9 SparkStreaming 开荒

た入场券 2022-12-10 03:51 30阅读 0赞

### 文章目录 ###

*  一. Overview
 *   *  1.2 Streaming 特点
     *  1.3 Spark Streaming架构
     *   *   *  1.3.2 背压机制
     *  1.4 WC案例
     *   *   *  1.4.2 WC解析
     *  1.5 DStream创建
     *   *   *   *  1.5.1 RDD队列(queue)

[Spark官网介绍][Spark]

# 一. Overview #

SparkStreaming 也是基于Spark core API做的 高宽带 容错的数据流  
数据源 & 数据出口如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]  
事实上，虽然称为流处理，底层实现还是微批处理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]

**微批处理原理**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]  
**DStream**  
和Spark基于RDD的概念很相似，Spark Streaming使用离散化流(discretized stream)作为抽象表示，叫作DStream

DSteram是随着时间推移而收到数据的序列  
每个 时间区间 收到的数据都作为RDD存在，而DStream是 这些RDD 组成的序列(离散化)  
所以，DStream就是对RDD在实时数据处理场景的一种封装

## 1.2 Streaming 特点 ##

1.  易用

![在这里插入图片描述][20200923195226941.png_pic_center]

1.  容错  
    ![在这里插入图片描述][20200923195257939.png_pic_center]
2.  Spark全栈解决

![在这里插入图片描述][20200923195322132.png_pic_center]

## 1.3 Spark Streaming架构 ##

整体架构图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]

SparkStreaming架构图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]

#### 1.3.2 背压机制 ####

Spark 1.5以前版本，用户如果要限制Receiver的数据接收速率，可以通过设置静态配制参数“spark.streaming.receiver.maxRate”的值来实现，此举虽然可以通过限制接收速率，来适配当前的处理能力，防止内存溢出，但也会引入其它问题。比如：producer数据生产高于maxRate，当前集群处理能力也高于maxRate，这就会造成资源利用率下降等问题。  
为了更好的协调数据接收速率与资源处理能力，1.5版本开始Spark Streaming可以动态控制数据接收速率来适配集群数据处理能力。背压机制（即Spark Streaming Backpressure）: 根据JobScheduler反馈作业的执行信息来动态调整Receiver数据接收率。  
通过属性“spark.streaming.backpressure.enabled”来控制是否启用backpressure机制，默认值false，即不启用。

## 1.4 WC案例 ##

1.  添加依赖

<dependency>
                    <groupId>org.apache.spark</groupId>
                    <artifactId>spark-streaming_${
        scala.tools.version}</artifactId>
                    <version>${
        spark.version}</version>
                </dependency>

def main(args: Array[String]): Unit = {
        
        //1. 初始化Spark配置信息
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamingWC")
        //2. 初始化SparkStreamingContext
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        //3. 通过监控端口创建DStream , 读进来的数据为一行行
        val lineStreams = ssc.socketTextStream("ifeng", 9527)
    
        //wc
        val wc = lineStreams.flatMap(_.split(",")).map((_, 1))countByValue()
        wc.print()
    
        //启动SparkStreamingContext(采集器)
        ssc.start()
        //等待采集器的结束，
        ssc.awaitTermination()

五秒打印一次  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]

#### 1.4.2 WC解析 ####

Discretized Stream 是 Spark Streaming 的基础抽象，代表持续性的 数据流 和 经过各种Spark原语句操作后的结果数据流。  
在内部实现上，DStream是一系列连续的RDD来表示，每个RDD含有一段时间间隔内的数据

![在这里插入图片描述][20200924113931341.png_pic_center]  
对数的操作也是按照RDD为单位来进行的

![在这里插入图片描述][20200924113958282.png_pic_center]  
计算过程由Spark Engine来完成

![在这里插入图片描述][20200924114031400.png_pic_center]

## 1.5 DStream创建 ##

##### 1.5.1 RDD队列(queue) #####

测试过程中，可以通过使用

ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD 都会作为一个DStream处理

**案例**

def main(args: Array[String]) {
        
    
        //1.初始化Spark配置信息
        val conf = new SparkConf().setMaster("local[*]").setAppName("RDDStream")
    
        //2.初始化SparkStreamingContext
        val ssc = new StreamingContext(conf, Seconds(4))
    
        //3.创建RDD队列
        val rddQueue = new mutable.Queue[RDD[Int]]()
    
        //4.创建QueueInputDStream
        val inputStream = ssc.queueStream(rddQueue,oneAtATime = false)
    
        //5.处理队列中的RDD数据
        val mappedStream = inputStream.map((_,1))
        val reducedStream = mappedStream.reduceByKey(_ + _)
    
        //6.打印结果
        reducedStream.print()
    
        //7.启动任务
        ssc.start()
    
    //8.循环创建并向RDD队列中放入RDD
        for (i <- 1 to 5) {
        
          rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
          Thread.sleep(2000)
        }
    
        ssc.awaitTermination()
    
      }
    }

[Spark]: http://spark.apache.org/docs/latest/streaming-programming-guide.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/2ab57330cfa649c48c1373e5ef816f24.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/428e81ead3744568b9f4e4e7b342da20.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/d60f436f12364dbf93e14d7fc54732aa.png
[20200923195226941.png_pic_center]: /images/20221123/22884cb108b747e79149e794bcfd4927.png
[20200923195257939.png_pic_center]: /images/20221123/ea518e90651e45c1a081641a6e197ba8.png
[20200923195322132.png_pic_center]: /images/20221123/96a00a5bc12b44569a288ee8d620c49c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/c98e4b9856e047e892231b6a80865258.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/461572e3f7494d7fb9068c8e9c8d0523.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/ac634ab1da0b4bdeba467be986fa181f.png
[20200924113931341.png_pic_center]: /images/20221123/a254279d282643bf946ddf8bcbda750a.png
[20200924113958282.png_pic_center]: /images/20221123/d22e6bcd54724915a863ee2ebc7d8f90.png
[20200924114031400.png_pic_center]: /images/20221123/8f60078d81cb42a098c100d44d287245.png