1.概述

Spark Streaming将不同的数据源，经过处理之后，结果输出到外部文件系统。

特点：低延时；能从错误中高效地恢复过来；能够运行在成百上千的节点上；能够将批处理、机器学习、图计算等子框架综合使用

工作原理：

粗粒度：Spark Streaming接收到实时数据流，把数据按照指定的时间段切成一片片小的数据块，然后把小的数据块传给Spark Engine处理。

细粒度：Spark应用程序运行在Driver端（Spark Context、Streaming Context），在Driver端会要求在Executor端开启Receiver接收器，当收到输入过来的数据（Input Stream）时，拆分成数据块先存进内存，如果是多副本可以拷贝到其他的Executor中。拷贝完成之后，Receiver会把block信息（在哪些机器上面）发送给Streaming Context；每隔几秒钟（时间周期），就会通知Spark Context开启几个jobs，然后把jobs分发到Executor去执行。

2.核心概念

DStreams：源源不断的数据流【一系列的RDD】，这里面的每个RDD，都是包含这一批次的数据

对DStreams操作算子，比如map/flatmap，其实底层会被翻译成对DStreams中的每个RDD都做相同的操作，因为一个DStreams是由不同批次的RDD构成的。

Input DStreams and Receiver ：每一个Input DStreams（除了文件系统上的数据）都需要关联一个Receiver，Receiver是用来从源头接收数据的，然后把数据存入spark内存中。

Transformations：从Input DStreams过来的数据进行修改（map、flatMap、filter ... ）

Output Operation：把DStreams中的数据写进外部系统（数据库或者文件系统）

3.案例

3.1 Spark Streaming处理socket数据

依赖：

    <properties><scala.version>2.11.8</scala.version><kafka.version>0.8.2.1</kafka.version><spark.version>2.2.0</spark.version><hadoop.version>2.6.0-cdh5.7.0</hadoop.version><hbase.version>1.2.0-cdh5.7.0</hbase.version></properties><!-- 添加cdh版本下载url --><repositories><repository><id>cloudera</id><url>https://repository.cloudera.com/artifactory/cloudera-repos</url></repository></repositories><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><!-- Kafka 依赖 --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_2.11</artifactId><version>${kafka.version}</version></dependency><!-- Hadoop 依赖 --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><!-- HBase 依赖 --><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>${hbase.version}</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>${hbase.version}</version></dependency><!-- Spark Streaming 依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><version>${spark.version}</version></dependency><!-- Spark Streaming 整合 Flume 依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-flume_2.11</artifactId><version>${spark.version}</version></dependency><!-- Spark Streaming 整合 Flume pull方式依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-flume-sink_2.11</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.5</version></dependency><!-- Spark Streaming 整合 Kafka 依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>${spark.version}</version></dependency><!-- Spark SQL的依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>${spark.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.module</groupId><artifactId>jackson-module-scala_2.11</artifactId><version>2.6.5</version></dependency><dependency><groupId>net.jpountz.lz4</groupId><artifactId>lz4</artifactId><version>1.3.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.37</version></dependency><!-- 将Logger日志写到Flume里面所需要的依赖 --><dependency><groupId>org.apache.flume.flume-ng-clients</groupId><artifactId>flume-ng-log4jappender</artifactId><version>1.6.0</version></dependency></dependencies>

/*** Spark Streaming处理Socket数据* 测试：nc*/
object NetworkWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话System.setProperty("hadoop.home.dir", "E:/winutils/")val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")//创建StreamingContext需要两个参数：SparkConf和batch intervalval ssc = new StreamingContext(sparkConf, Seconds(5))val lines = ssc.socketTextStream("localhost", 6789)val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)result.print()ssc.start()ssc.awaitTermination()}
}

用nc -lk 6789命令，输入单词进行测试。

3.2 Spark Streaming处理文件系统的数据

/*** 使用Spark Streaming处理文件系统(local/hdfs)数据*/
object FileWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话System.setProperty("hadoop.home.dir", "E:/winutils/")val sparkConf = new SparkConf().setMaster("local[2]").setAppName("FileWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))// 文件系统的数据不需要Receiverval lines = ssc.textFileStream("file:///D:/BidDataTestFile/ss/")val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)result.print()ssc.start()ssc.awaitTermination()}
}

注意：它只会读取新文件的数据，不会处理已经处理过的文件。

4.Spark Streaming进阶

4.1 统计累积出现单词的次数

/*** 使用Spark Streaming 完成已有单词统计*/
object StatefulWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话System.setProperty("hadoop.home.dir", "E:/winutils/")val sparkConf = new SparkConf().setMaster("local[2]").setAppName("StatefulWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))//如果使用了stateful的算子，必须要设置checkpoint//在生产环境中，建议大家把checkpoint设置到HDFS的某个文件夹中//"." 表示把检查文件放在当前目录ssc.checkpoint(".")val lines = ssc.socketTextStream("localhost", 6789)val result = lines.flatMap(_.split(" ")).map((_, 1))//调用updateStateByKey算子，统计单词在全局中出现的次数val state = result.updateStateByKey[Int](updateFunction _)state.print()ssc.start()ssc.awaitTermination()}/*** 把当前的数据去更新已有的或者是老的数据* @param currentValues 当前的数据* @param preValues 老的数据* @return*/def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int] = {//相当于遍历currentValues，把当前值相加求和val current = currentValues.sumval pre = preValues.getOrElse(0)  //如果取不到值的话，就返回0Some(current + pre)}
}

4.2 统计单词个数写入到MySQL数据库

/*** 使用Spark Streaming完成词频统计，并将结果写入到MySQL数据库中*/
object ForeachRDDApp {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话System.setProperty("hadoop.home.dir", "E:/winutils/")val sparkConf = new SparkConf().setMaster("local[2]").setAppName("ForeachRDDApp")val ssc = new StreamingContext(sparkConf, Seconds(5))val lines = ssc.socketTextStream("localhost", 6789)val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)result.print()//将结果写入MySQL数据库result.foreachRDD(rdd => {rdd.foreachPartition(partitionOfRecords => {val connection = createConnection()partitionOfRecords.foreach(record => {val sql = "insert into wordcount(word, wordcount) values('" + record._1 + "'," + record._2 + ")"connection.createStatement().execute(sql)})connection.close()})})ssc.start()ssc.awaitTermination()}/*** 获取MySQL连接* @return*/def createConnection() = {Class.forName("com.mysql.jdbc.Driver")DriverManager.getConnection("jdbc:mysql://localhost:3306/imooc_spark", "root", "MySQLKenan_07")}
}

4.3 Window窗体的使用

定时的进行一个时间段内的数据处理

window length：窗口的长度

sliding interval：窗口的间隔

这2个参数和batch size有关系：整数倍

eg.每隔10秒计算前10分钟的wc ==> 窗口间隔10s，窗口长度10min

// 窗口长度，窗口间隔
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int, b:Int) => (a+b), Seconds(30), Seconds(10))

4.4 黑名单过滤

访问日志：20180808,zs 20180808,ls 20180808,ww【 ==> (zs:20180808, zs)】

黑名单列表：zs、ls 【 ==> (zs,true)】

left join ==> (zs:[<20180808,zs>,<true>]) (zs:[<20180808,zs>,<false>]) ===> 我们需要的是<20180808,zs>

/*** 黑名单过滤*/
object TransformApp {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[2]").setAppName("TransformApp")val ssc = new StreamingContext(sparkConf, Seconds(5))//构建黑名单val blacks = List("zs", "ls")val blacksRDD = ssc.sparkContext.parallelize(blacks).map(x => (x, true))val lines = ssc.socketTextStream("localhost", 6789)//前面是名字后面是完整的信息val clickLog = lines.map(x=>(x.split(",")(1), x)).transform(rdd => {rdd.leftOuterJoin(blacksRDD).filter(x => x._2._2.getOrElse(false) != true).map(x => x._2._1)})clickLog.print()ssc.start()ssc.awaitTermination()}
}

4.5 Spark Streaming 整合 Spark SQL

/*** Spark Streaming 整合 Spark SQL完成词频统计*/
object SqlNetworkWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话System.setProperty("hadoop.home.dir", "E:/winutils/")val sparkConf = new SparkConf().setAppName("SqlNetworkWordCount").setMaster("local[2]")val ssc = new StreamingContext(sparkConf, Seconds(5))val lines = ssc.socketTextStream("localhost", 6789)val words = lines.flatMap(_.split(" "))//从DStream里面遍历每一个批次words.foreachRDD((rdd, time) => {val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)import spark.implicits._val wordsDataFrame = rdd.map(w => Record(w)).toDF()wordsDataFrame.createOrReplaceTempView("words")val wordCountsDataFrame = spark.sql("select word, count(*) as total from words group by word")println(s"=========== $time ===========")wordCountsDataFrame.show()})ssc.start()ssc.awaitTermination()}//用来将RDD转成DataFramecase class Record(word: String)object SparkSessionSingleton {@transient private var instance: SparkSession = _def getInstance(sparkConf: SparkConf): SparkSession = {if (instance == null) {instance = SparkSession.builder.config(sparkConf).getOrCreate()}instance}}
}

5.Spark Streaming整合Flume

5.1 Push方式

Flume采集数据，直接传给Spark Streaming

vim flume_push_streaming.conf

simple-agent.sources = netcat-source
simple-agent.sinks = avro-sink
simple-agent.channels = memory-channelsimple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = hadoop000
simple-agent.sources.netcat-source.port = 44444simple-agent.sinks.avro-sink.type = avro
# 本地测试的时候，这里应该是本地ip
simple-agent.sinks.avro-sink.hostname = hadoop000
simple-agent.sinks.avro-sink.port = 41414 simple-agent.channels.memory-channel.type = memorysimple-agent.sources.netcat-source.channels = memory-channel
simple-agent.sinks.avro-sink.channel = memory-channel

依赖

        <!-- Spark Streaming 整合 Flume 依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-flume_2.11</artifactId><version>${spark.version}</version></dependency>

/*** Spark Streaming 整合 Flume的第一种方式 --- Push*/
object FlumePushWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话//System.setProperty("hadoop.home.dir", "E:/winutils/")if (args.length != 2) {System.err.println("Usage: FlumePushWordCount <hostname> <port>")System.exit(1)}val Array(hostname, port) = args//通过submit方式提交到服务器，需要注释掉下面的内容val sparkConf = new SparkConf()//.setMaster("local[2]").setAppName("FlumePushWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))//如何使用SparkStreaming整合Flume//从hostname 的 port 中获取传过来的数据val flumeStream = FlumeUtils.createStream(ssc, hostname, port.toInt)flumeStream.map(x => new String(x.event.getBody.array()).trim).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print(10)ssc.start()ssc.awaitTermination()}
}

测试：先启动应用程序，再启动Flume

flume-ng agent \
--name simple-agent \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/flume_push_streaming.conf \
-Dflume.root.logger=INFO,console

本地测试总结：
1> 启动Spark Streaming作业
2> 启动flume agent
3> 通过telnet输入数据，观察IDEA控制台的输出
打包方式：mvn clean package -DskipTests

提交到spark服务器调试

./spark-submit \
--class com.xq.spark.examples.FlumePushWordCount \
--name FlumePushWordCount \
--master local[2] \
--executor-memory 1G \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0,org.apache.flume:flume-ng-sdk:1.6.0 \
/home/Kiku/lib/sparktrain-1.0-SNAPSHOT.jar \
hadoop000 41414

flume-ng agent \
--name simple-agent \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/flume_push_streaming.conf \
-Dflume.root.logger=INFO,console

telnet hadoop000 44444

测试成功！！

5.2 Pull方式（推荐！！）

Flume把数据采集过来之后，丢给sink，然后spark streaming到sink去获取数据。

依赖

        <!-- Spark Streaming 整合 Flume pull方式依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-flume-sink_2.11</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.5</version></dependency>

vim flume_pull_streaming.conf

simple-agent.sources = netcat-source
simple-agent.sinks = spark-sink
simple-agent.channels = memory-channelsimple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = hadoop000
simple-agent.sources.netcat-source.port = 44444simple-agent.sinks.spark-sink.type = org.apache.spark.streaming.flume.sink.SparkSink
simple-agent.sinks.spark-sink.hostname = hadoop000
simple-agent.sinks.spark-sink.port = 41414 simple-agent.channels.memory-channel.type = memorysimple-agent.sources.netcat-source.channels = memory-channel
simple-agent.sinks.spark-sink.channel = memory-channel

/*** Spark Streaming 整合 Flume的第二种方式 --- Pull*/
object FlumePullWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话//System.setProperty("hadoop.home.dir", "E:/winutils/")if (args.length != 2) {System.err.println("Usage: FlumePullWordCount <hostname> <port>")System.exit(1)}val Array(hostname, port) = args//通过submit方式提交到服务器，需要注释掉下面的内容val sparkConf = new SparkConf()//.setMaster("local[2]").setAppName("FlumePullWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))val flumeStream = FlumeUtils.createPollingStream(ssc, hostname, port.toInt)flumeStream.map(x => new String(x.event.getBody.array()).trim).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()ssc.start()ssc.awaitTermination()}
}

本地调试：

先启动Flume，后启动Spark应用程序

flume-ng agent \
--name simple-agent \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/flume_pull_streaming.conf \
-Dflume.root.logger=INFO,console

服务器调试：

项目打包【注意：注释掉local[2]，appName参数】

./spark-submit \
--class com.xq.spark.examples.FlumePullWordCount \
--name FlumePullWordCount \
--master local[2] \
--executor-memory 1G \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0,org.apache.flume:flume-ng-sdk:1.6.0,org.apache.spark:spark-streaming-flume-sink_2.11:2.2.0 \
/home/Kiku/lib/sparktrain-1.0-SNAPSHOT.jar \
hadoop000 41414

6.Spark Streaming整合Kafka

6.1 Receiver方式整合

通过Receiver接受从Kafka过来的数据，存储到Spark Executor；job启动的时候，处理存储的这些数据。

1）启动ZK zkServer.sh start

2）启动Kafka kafka-server-start.sh -daemon /home/Kiku/app/kafka_2.11-0.9.0.0/config/server.properties

3）创建topic kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic xxxxxx

4）通过控制台测试topic能否正确的发送、接受消息

kafka-console-producer.sh --broker-list hadoop000:9092 --topic kafka_streaming_topic

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic kafka_streaming_topic

依赖

        <!-- Spark Streaming 整合 Kafka 依赖 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>${spark.version}</version></dependency>

/*** Spark Streaming 对接 Kafka的第一种方式 -- Receiver*/
object KafkaReceiverWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话//System.setProperty("hadoop.home.dir", "E:/winutils/")if (args.length != 4) {System.err.println("Usage: KafkaReceiverWordCount <zkQuorum> <group> <topics> <numThreads>")}val Array(zkQuorum, group, topics, numThreads) = argsval sparkConf = new SparkConf()//.setMaster("local[2]").setAppName("KafkaReceiverWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap//Spark Streaming 如何对接Kafkaval messages = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)//第二位是我们字符串的值messages.map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()ssc.start()ssc.awaitTermination()}
}

提交到服务器

代码中去掉，local[2]、appName

spark-submit \
--class com.xq.spark.examples.KafkaReceiverWordCount \
--name KafkaReceiverWordCount \
--master local[2] \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/Kiku/lib/sparktrain-1.0-SNAPSHOT.jar \
192.168.6.130:2181 test kafka_streaming_topic 1

6.2 Direct方式整合

无Receiver，端到端的数据保障。周期性的从每一个topic分区中查询Kafka偏移量，周期性的把这个范围的数据通过每个批次进行处理。

优点：简化并行度，不需要创建多个Input Stream，使用direct Stream进行处理

0数据丢失，Receiver方式中，我们需要WAL(Write Ahead Log)写入日志中，以副本的方式存储，才能保证数据不丢失。

效率不高。Direct方式，不再需要WAL，性能提升。

缺点：无法更新偏移量到Zookeeper，需要手动更新。

/*** Spark Streaming 对接 Kafka的第二种方式 -- Direct*/
object KafkaDirectWordCount {def main(args: Array[String]): Unit = {//因为主机没有配置hadoop环境，所以需要加上这句话//System.setProperty("hadoop.home.dir", "E:/winutils/")if (args.length != 2) {System.err.println("Usage: KafkaReceiverWordCount <brokers> <topics>")System.exit(1)}val Array(brokers, topics) = argsval sparkConf = new SparkConf()//.setMaster("local[2]").setAppName("KafkaDirectWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))val topicsSet = topics.split(",").toSetval kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)//第二位是我们字符串的值messages.map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()ssc.start()ssc.awaitTermination()}
}

提交到服务器

spark-submit \
--class com.xq.spark.examples.KafkaDirectWordCount \
--name KafkaDirectWordCount \
--master local[2] \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/Kiku/lib/sparktrain-1.0-SNAPSHOT.jar \
192.168.6.130:9092 kafka_streaming_topic

大数据入门之分布式计算框架Spark(3) -- Spark Streaming相关推荐

大数据入门之分布式计算框架Spark(2) -- Spark SQL
1.Spark SQL概述一个运行在Spark上执行sql的处理框架,可以用来处理结构化的数据[外部数据源(访问hive.json.parquet等文件的数据)]. Spark SQL提供了SQL的 ...
大数据入门(五)-分布式计算框架MapReduce
1 概述源自于Google的MapReduce论文,发表于2004年12月. Hadoop MapReduce是Google MapReduce的克隆版优点海量数量离线处理易开发易运行缺点 ...
Hadoop大数据原理(3) - 分布式计算框架MapReduce
文章目录 1. 大数据的通用计算 2 MapReduce编程模型 3. MapReduce计算框架 3.1 三类关键进程大数据应用进程 JobTracker进程 TaskTracker进程 3.2 ...
【大数据入门笔记系列】第六节分布式计算框架MapReduce的工作流程
[大数据入门笔记系列]第六节分布式计算框架MapReduce的工作流程前言 MapReduce分布式运算 MapReduceApplication MapTask ReduceTask split ...
车联网大数据框架_大数据基础：ORM框架入门简介
作为大数据开发技术者,需要掌握扎实的Java基础,这是不争的事实,所以对于Java开发当中需要掌握的重要框架技术,也需要有相应程度的掌握,比如说ORM框架.今天的大数据基础分享,我们就来具体讲一讲OR ...
大数据入门：Spark RDD、DataFrame、DataSet
在Spark的学习当中,RDD.DataFrame.DataSet可以说都是需要着重理解的专业名词概念.尤其是在涉及到数据结构的部分,理解清楚这三者的共性与区别,非常有必要.今天的大数据入门分享,我们 ...
女友问粉丝过万如何庆祝，我发万字长文《保姆级大数据入门篇》感恩粉丝们支持，学姐|学妹|学弟|小白看了就懂
2021大数据领域优质创作博客,带你从入门到精通,该博客每天更新,逐渐完善大数据各个知识体系的文章,帮助大家更高效学习. 有对大数据感兴趣的可以关注微信公众号:三帮大数据目录粉丝破万了新星计划申 ...
大数据入门概念及应用场景
参考内容: [知乎]深入浅出大数据:到底什么是Hadoop? [知乎]五万字 | Hive知识体系保姆级教程大数据入门概念及应用场景一.入门概念 1.1 大数据的4V 1.2 大数据处理的最佳工 ...
女友问粉丝过万如何庆祝，我发长文《保姆级大数据入门篇》感恩粉丝们支持，学姐|学弟看了就懂
文章目录粉丝破万了新星计划申请时粉丝数新星内卷抢热榜之旅运营整顿新星执行新规重整旗鼓输出内容为王女友问粉丝过万如何庆祝保姆级大数据入门篇一.学习重点划定二.Java和大数据关系三. ...

大数据入门之分布式计算框架Spark(3) -- Spark Streaming

1.概述

2.核心概念

3.案例

3.1 Spark Streaming处理socket数据

3.2 Spark Streaming处理文件系统的数据

4.Spark Streaming进阶

4.1 统计累积出现单词的次数

4.2 统计单词个数写入到MySQL数据库

4.3 Window窗体的使用

4.4 黑名单过滤

5.Spark Streaming整合Flume

5.1 Push方式

提交到spark服务器调试

5.2 Pull方式（推荐！！）

6.Spark Streaming整合Kafka

6.1 Receiver方式整合

提交到服务器

6.2 Direct方式整合

大数据入门之分布式计算框架Spark(3) -- Spark Streaming相关推荐

最新文章

热门文章

大数据入门之分布式计算框架Spark(3) -- Spark Streaming

1.概述

2.核心概念

3.案例

3.1 Spark Streaming处理socket数据

3.2 Spark Streaming处理文件系统的数据

4.Spark Streaming进阶

4.1 统计累积出现单词的次数

4.2 统计单词个数写入到MySQL数据库

4.3 Window窗体的使用

4.4 黑名单过滤

5.Spark Streaming整合Flume

5.1 Push方式

提交到spark服务器调试

5.2 Pull方式 （推荐！！）

6.Spark Streaming整合Kafka

6.1 Receiver方式整合

提交到服务器

6.2 Direct方式整合

大数据入门之分布式计算框架Spark(3) -- Spark Streaming相关推荐

最新文章

热门文章

5.2 Pull方式（推荐！！）