Spark学习（5）- Spark Streaming

缺乏、安全感 2022-03-12 03:16 201阅读 0赞

**目录：**

1.  Spark Streaming简介
    
    1.  概述
    2.  工作原理
    3.  离散流 DStreams
2.  Spark Streaming架构及运行流程
3.  Spark Streaming编程

### 一.Spark Streaming简介 ###

**1.概述：**[官方网站][Link 1]

1.  Spark Streaming是一个构建在Spark之上，是Spark四大组件之一
2.  是Spark系统中用于处理流式数据的分布式流式处理框架
3.  具有可伸缩、高吞吐量、容错能力强等特点。
4.  处理的数据源可以来自Kafka，Flume，Twitter，ZeroMQ，Kinesis or TCP sockets等，结果集可保存到HDFS、数据库或者实时Dashboard展示，如下图所示   
    ![这里写图片描述][70]  
    总的来说我们可以从三点进行考虑：**输入—–计算—–输出**。
    
    1.  输入：可以从Kafka，Flume,HDFS等获取数据 
    2.  计算：我们可以通过map，reduce,join等一系列算子通过spark计算引擎进行计算（基本和RDD一样，使用起来更方便。） 
    3.  输出：可以输出到HDFS,数据库，HBase等。

**2.工作原理**

在内部，它的工作原理如下图所示,。Spark Streaming接收实时的输入数据流，然后将这些数据切分为批数据供Spark引擎处理，Spark引擎将数据生成最终的结果数据。

![这里写图片描述][70 1]  
从图中也能看出它将输入的数据分成多个batch进行处理，严格来说spark streaming 并不是一个真正的实时框架,因为它是分批次进行处理的。

Spark Streaming支持一个高层的抽象，叫做离散流(`discretized stream`)或者`DStream ，`它表示连续的数据流。 DStream可以通过Kafka，Flume和Kinesis等来源的输入数据流创建，也可以通过在其他DStream上应用高级操作来创建。在内部，DStream表示为一系列RDD。

**3.离散流（DStreams）**

离散流或者DStreams是Spark Streaming提供的基本的抽象，它代表一个连续的数据流。它要么是从源中获取的输入流，要么是输入流通过转换算子生成的处理后的数据流。在内部，DStreams由一系列连续的RDD组成。DStreams中的每个RDD都包含确定时间间隔内的数据，如下图所示：

![这里写图片描述][70 2]

任何对DStreams的操作都转换成了对DStreams隐含的RDD的操作。在前面的[例子][Link 2]中，`flatMap`操作应用于`lines`这个DStreams的每个RDD，生成`words`这个DStreams的RDD。过程如下图所示：

![这里写图片描述][70 3]

通过Spark引擎计算这些隐含RDD的转换算子。DStreams操作隐藏了大部分的细节，并且为了更便捷，为开发者提供了更高层的API。

### 二.Spark Streaming 架构及运行流程 ###

我们应该知道spark有很多种运行模式，下面通过spark on yarn (cluster模式)的模式图进行介绍，所以想要对spark streaming的运行架构进行理解，你要知道在yarn上提交作业的流程，以及spark的运行流程，下面是我在网上找的一幅图，我们根据这幅图 进行一个学习：   
![70 4][]

说明：符号表示：1，2，3….代表Spark on Yarn启动流程 ；(1)(2)(3)….代表Spark Streaming执行过程。

**Spark on Yarn启动流程**

1.  通过spark client提交作业到RM 
2.  ResouceManager为该作业分配第一个Container，并与对应的NodeManager通信，创建Spark ApplicationMaster（每个SparkContext都有一个ApplicationMaster） 
3.  NodeManager启动Spark AppMaster。 
4.  Spark AppMaster并向ResourceManager AsM注册，用户就可以通过UI查看作业情况。 
5.  ResourceManager通知NodeManager分配Container。（每个container的对应一个executor） 
6.  NodeManager准备资源，并分配给executor。

**Spark Streaming执行过程**

spark on Yarn 模式Driver运行在NM的container之中，运行application的main()函数并自动创建SparkContext，在SparkContext之上会创建一个 StreamingContext（因为途中并没有标出，这里说明下）。

*  （1）SparkContext向资源管理器注册并申请运行Executor资源； 
 *  （2）Executor会启动Receive接收数据（Data Received），分批处理。 
 *  （3）Receive接收到数据后汇报给streamingcontext(底层调用的sparkcontext)，他会以多个副本存储，默认两个（后面进行源码解读就知道了。） 
 *  （4）Spark ApplicationMaster和executor（container）进行交互，分配task。 
 *  （5）每个executor上会运行多个task执行任务。

最后把结果保存在HDFS上。

**Saprk Streaming数据处理过程**

![这里写图片描述][70 5]  
首先，Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块。Spark Streaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据。每个块都会生成一个Spark Job处理，最终结果也返回多块。这里也和上面所说的一致。

### 三.Spark Streaming 编程 ###

**步骤：**

1.  创建maven项目
2.  在maven项目中引入依赖：[引入依赖][Link 3]
3.  代码
    
    1.  初始化StreamingContext：[初始化StreamingContext][StreamingContext]
    2.  获取**DStreams**：[输入DStreams][DStreams]
    3.  转换**DStreams**：[DStream中的转换][DStream]
    4.  输出**DStreams**：[DStream的输出操作][DStream 1]
    5.  **DStreams**缓存或持久化：[DStreams缓存或持久化][DStreams 1]

**实例1.SparkStreaming接受socket数据，实现单词计数WordCount**

流程：

1.  创建项目
2.  引入相应的jar包（在[*Maven* *Repository*][Maven_Repository]中找到和自己安装版本对应的jar包：[https://mvnrepository.com/search?q=spark+streaming][https_mvnrepository.com_search_q_spark_streaming]）
    
        <dependency>
                    <groupId>org.apache.spark</groupId>
                    <artifactId>spark-streaming_2.11</artifactId>
                    <version>2.0.2</version>
        </dependency>
3.  编写代码，并运行
    
        import org.apache.spark.SparkConf
        import org.apache.spark.SparkContext
        import org.apache.spark.SparkContext._
        import org.apache.spark.rdd.RDD
        import org.apache.spark.streaming.{Seconds, StreamingContext}
        import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
        
        object SparkFirst {
          def main(args: Array[String]): Unit = {
            
            //1、创建sparkConf   设置master的地址local[N] ,n必须大于1，其中1个线程负责去接受数据，另一线程负责处理接受到的数据
            val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocket").setMaster("local[2]")
            //2、创建sparkContext
            val sc = new SparkContext(sparkConf)
            sc.setLogLevel("WARN")
            //3、创建streamingContext,需要sparkContext和以多久时间间隔为一个批次
            val ssc = new StreamingContext(sc,Seconds(5))
            //4、通过streaming接受socket数据
            val stream: ReceiverInputDStream[String] = ssc.socketTextStream("localhost",9999)
            //5、切分每一行
            val words: DStream[String] = stream.flatMap(_.split(" "))
            //6、每个单词记为1
            val wordAndOne: DStream[(String, Int)] = words.map((_,1))
            //7、相同单词出现的次数累加
            val result:DStream[(String,Int)] = wordAndOne.reduceByKey(_+_)
            //8、打印
            result.print()
            //9、开启流式计算
            ssc.start()
            //一直会阻塞，等待退出
            ssc.awaitTermination()
          }
        }
4.  新开一个客户端，通过nc命令向9999端口传送数据
    
        [root@dn02 ~]# nc -l 9999
        who you are today 
        why you are here
        why you are here
5.  结果：
    
        18/08/18 09:22:30 WARN BlockManager: Block input-0-1533720150200 replicated to only 0 peer(s) instead of 1 peers
        -------------------------------------------
        Time: 1533720155000 ms
        -------------------------------------------
        (today,1)
        (are,1)
        (who,1)
        (you,1)

**实例2.SparkStreaming接受socket数据，实现单词计数累加 **

*  **代码**
    
        import org.apache.spark.{SparkConf, SparkContext}
        import org.apache.spark.streaming.{Seconds, StreamingContext}
        import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
        
        //todo:利用sparkStreaming接受socket数据，实现所有批次单词计数结果累加
        object SparkStreamingSocketTotal {
        
          //定义一个方法
          //currentValues:他表示在当前批次每个单词出现的所有的1   (hadoop,1) (hadoop,1)(hadoop,1)
          //historyValues:他表示在之前所有批次中每个单词出现的总次数   (hadoop,100)
          def updateFunc(currentValues:Seq[Int], historyValues:Option[Int]):Option[Int] = {
                  val newValue: Int = currentValues.sum+historyValues.getOrElse(0)
                  Some(newValue)
          }
        
          def main(args: Array[String]): Unit = {
              //1、创建sparkConf
              val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketTotal").setMaster("local[2]")
              //2、创建sparkContext
              val sc = new SparkContext(sparkConf)
              sc.setLogLevel("WARN")
             //3、创建streamingContext
              val ssc = new StreamingContext(sc,Seconds(5))
             //设置checkpoint目录
              ssc.checkpoint("./ck")
            //4、接受socket数据
              val stream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
            //5、切分每一行
              val words: DStream[String] = stream.flatMap(_.split(" "))
            //6、把每一个单词计为1
              val wordAndOne: DStream[(String, Int)] = words.map((_,1))
            //7、相同单词出现的次数累加
              val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
        
            //8、打印结果数据
              result.print()
        
            //9、开启流式计算
              ssc.start()
              ssc.awaitTermination()
            }
        }

**实例3.SparkStreaming开窗函数reduceByKeyAndWindow，实现单词计数**

*  **代码**
    
        import org.apache.spark.{SparkConf, SparkContext}
        import org.apache.spark.streaming.{Seconds, StreamingContext}
        import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
        
        //todo:利用sparkStreaming开窗函数 reducebyKeyAndWindow实现单词计数
        object SparkStreamingSocketWindow {
        
          //定义一个方法
          //currentValues:他表示在当前批次每个单词出现的所有的1   (hadoop,1) (hadoop,1)(hadoop,1)
          //historyValues:他表示在之前所有批次中每个单词出现的总次数   (hadoop,100)
          def updateFunc(currentValues:Seq[Int], historyValues:Option[Int]):Option[Int] = {
            val newValue: Int = currentValues.sum+historyValues.getOrElse(0)
            Some(newValue)
          }
        
          def main(args: Array[String]): Unit = {
            //1、创建sparkConf
            val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketWindow").setMaster("local[2]")
            //2、创建sparkContext
            val sc = new SparkContext(sparkConf)
            sc.setLogLevel("WARN")
            //3、创建streamingContext
            val ssc = new StreamingContext(sc,Seconds(5))
            //设置checkpoint目录
            ssc.checkpoint("./ck")
            //4、接受socket数据
            val stream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
            //5、切分每一行
            val words: DStream[String] = stream.flatMap(_.split(" "))
            //6、把每一个单词计为1
            val wordAndOne: DStream[(String, Int)] = words.map((_,1))
            //7、相同单词出现的次数累加
            //reduceByKeyAndWindow该方法需要三个参数
            //reduceFunc：需要一个函数
            //windowDuration:表示窗口的长度
            //slideDuration:表示窗口滑动时间间隔，即每隔多久计算一次
            val result: DStream[(String, Int)] = wordAndOne.reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(5),Seconds(10))
        
            //8、打印结果数据
            result.print()
        
            //9、开启流式计算
            ssc.start()
            ssc.awaitTermination()
          }
        }

[Link 1]: http://spark.apache.org/docs/latest/streaming-programming-guide.html
[70]: /images/20220312/86e23826f211469a8ca93ff66b22c7cf.png
[70 1]: /images/20220312/d55f6e210b69426da0296f4bc24ab60e.png
[70 2]: /images/20220312/7a03597219dc4523a1ba40f78eb09b2f.png
[Link 2]: https://www.w3cschool.cn/spark/cshr7ozt.html
[70 3]: /images/20220312/69146598f9e94660a0e584dd46c5064d.png
[70 4]: /images/20220312/b93957bc643640d09a7ad9984d3733f8.png
[70 5]: /images/20220312/9e7e2036571042a4a2768b708484bacf.png
[Link 3]: https://www.w3cschool.cn/spark/xb7e5ozt.html
[StreamingContext]: https://www.w3cschool.cn/spark/5y8coozt.html
[DStreams]: https://www.w3cschool.cn/spark/yshfqozt.html
[DStream]: https://www.w3cschool.cn/spark/7xrbnozt.html
[DStream 1]: https://www.w3cschool.cn/spark/eu5d2ozt.html
[DStreams 1]: https://www.w3cschool.cn/spark/9nhyfozt.html
[Maven_Repository]: https://www.baidu.com/link?url=uJ3Q53tXCIpJIgSHho2aWvk31L-5shDZU4X5VDDdZ8Zaxd9tzDNPJc63Qy2RnQki&wd=&eqid=be0c904200065d44000000025c834aff
[https_mvnrepository.com_search_q_spark_streaming]: https://mvnrepository.com/search?q=spark+streaming