SparkStreaming（二）：DStream操作

Bertha 。 2022-12-24 09:54 397阅读 0赞

### 文章目录 ###

*  一、DStream创建
 *   *  1.1 RDD 队列
     *  1.2 自定义数据源
     *  1.3 Kafka 数据源
     *   *  1.3.1 版本选型
         *  1.3.2 Kafka 0-10 Direct模式
 *  二、DStream转换
 *   *  2.1 无状态转化操作
     *   *  2.1.1 Transform
         *  2.1.2 join
     *  2.2 有状态转化操作
     *   *  2.2.1 UpdateStateByKey
         *  2.2.2 WindowOperations
 *  三、DStream输出
 *  四、优雅关闭

# 一、DStream创建 #

## 1.1 RDD 队列 ##

**用法**：通过使用`ssc.queueStream(queueOfRDDs)`来创建`DStream`，每一个推送到这个队列中的`RDD`，都会作为一个`DStream`处理。

**需求**：循环创建几个`RDD`，将`RDD`放入队列。通过`SparkStream`创建`Dstream`，计算`WordCount`。

**编写代码**：

//1、初始化Spark配置信息
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingTest")
        //2、初始化SparkStreamingContext
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        //3、创建RDD队列
        val rddQueue = new mutable.Queue[RDD[Int]]()
        //4、创建QueueInputStream
        val inputStream = ssc.queueStream(rddQueue, oneAtATime = false)
        //5、对采集数据进行操作
        val wordAndOneStream = inputStream.map((_, 1))
        val wordCountStream = wordAndOneStream.reduceByKey(_ + _)
        //6、打印结果
        wordCountStream.print()
    
        //7、启动任务
        ssc.start()
        //8、循环创建并向RDD队列中添加RDD
        for (i <- 1 to 5) { 
          rddQueue += ssc.sparkContext.makeRDD(List(i))
          Thread.sleep(1000)
        }
        ssc.awaitTermination()

**结果展示**：

-------------------------------------------
    Time: 1606374672000 ms
    -------------------------------------------
    (1,1)
    (2,1)
    (3,1)
    
    -------------------------------------------
    Time: 1606374675000 ms
    -------------------------------------------
    (4,1)
    (5,1)

## 1.2 自定义数据源 ##

**用法**：需要继承`Receiver`，并实现`onStart`、`onStop`方法来自定义数据源采集。

**需求**：自定义数据源，实现监控某个端口号，获取该端口号内容

**代码实现：**

//使用自定义Receiver创建DStream
        val lineStreams = ssc.receiverStream(new CustomerReceiver("192.168.182.200", 9559))
    
    
    class CustomerReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) { 
      //启动时，调用该方法，作用：读数据并将数据发送给Spark
      override def onStart(): Unit = { 
        new Thread("Socket Receiver") { 
          override def run(): Unit = { 
            receive()
          }
        }.start()
      }
    
      def receive(): Unit = { 
        //创建一个Socket
        val socket = new Socket(host, port)
        //定义一个变量用来接收端口传来的数据
        var input: String = null
        //创建一个BufferReader用于读取端口传来的数据
        val reader =
          new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))
        //读取数据
        input = reader.readLine()
        //当receiver没有关闭并且输入数据不为空，则循环发送数据给Spark
        while (!isStopped() && input != null) { 
          store(input)
          input = reader.readLine()
        }
        //跳出循环则关闭资源
        reader.close()
        socket.close()
        //重启任务
        restart("restart")
      }
    
      override def onStop(): Unit = { }
    }

## 1.3 Kafka 数据源 ##

### 1.3.1 版本选型 ###

**ReceiverAPI**：需要一个专门的`Executor`去接收数据，然后发送给其他的`Executor`做计算。存在的问题，接收数据的`Executor`和计算的`Executor`速度会有所不同，特别在接收数据的`Executor`速度大于计算的`Executor`速度，会导致计算数据的节点内存溢出。早期版本中提供此方式，当前版本不适用

**DirectAPI**：是由计算的`Executor`来主动消费`Kafka`的数据，速度由自身控制。

### 1.3.2 Kafka 0-10 Direct模式 ###

**需求**：通过`SparkStreaming`从`Kafka`读取数据，并将读取过来的数据做简单计算，最终打印到控制台。

**导入依赖**:

<dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
         <version>2.4.5</version>
    </dependency>

**编写代码：**

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingKafka")
        val ssc = new StreamingContext(sparkConf, Seconds(3))
    
        //定义Kafka参数
        val kafkaPara: Map[String, Object] = Map[String, Object](
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop100:9092",
          ConsumerConfig.GROUP_ID_CONFIG -> "hucheng",
          "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
          "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
        )
        //读取Kafka数据创建DStream
        val kafkaDStream =
          KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
            ConsumerStrategies.Subscribe[String, String](Set("hucheng"), kafkaPara))
        //将每条消息的KV取出
        val valueDStream = kafkaDStream.map(record => record.value())
        valueDStream.flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKey(_ + _)
          .print()
        //开启任务
        ssc.start()
        ssc.awaitTermination()

**测试：**

创建`topic`：

[root@hadoop100 kafka-0.11.0.0]# bin/kafka-topics.sh --zookeeper hadoop100:2181 \
    --create --replication-factor 3 --partitions 1 --topic hucheng

发送消息：

[root@hadoop100 kafka-0.11.0.0]# bin/kafka-console-producer.sh \
    --broker-list hadoop100:9092 --topic first
    >b  
    >c
    >a
    >c

控制台打印：

-------------------------------------------
    Time: 1606443630000 ms
    -------------------------------------------
    (b,1)
    (c,1)
    
    -------------------------------------------
    Time: 1606443633000 ms
    -------------------------------------------
    (a,1)
    (c,1)

查看`Kafka`消费进度：

[root@hadoop100 kafka-0.11.0.2]# bin/kafka-consumer-groups.sh --describe \
    --bootstrap-server hadoop100:9092 --group hucheng
    
    TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG CONSUMER-ID HOST  CLIENT-ID
    hucheng      0          6               7           1    -           -     -

# 二、DStream转换 #

`DStream`上的操作与`RDD`的类似，分为`Transformations`（转换）和`Output Operations`（输出）两种，此外转换操作中还有一些比较特殊的原语，如：`updateStateByKey()`、`transform()`以及各种`Window`相关的原语。

## 2.1 无状态转化操作 ##

无状态转化操作就是把简单的`RDD`转化操作应用到每个批次上，也就是转化`DStream`中的每一个`RDD`。部分无状态转化操作列在了下表中。注意，针对键值对的`DStream`转化操作(比如 `reduceByKey()`)要添加`import StreamingContext._`才能在`Scala`中使用。  
![在这里插入图片描述][20201127111714805.png]  
需要记住的是，尽管这些函数看起来像作用在整个流上一样，但事实上每个`DStream`在内部是由许多`RDD`（批次）组成，且无状态转化操作是分别应用到每个`RDD`上的。

例如：`reduceByKey()`会归约每个时间区间中的数据，但不会归约不同区间之间的数据。

### 2.1.1 Transform ###

`Transform`允许`DStream`上执行任意的`RDD-to-RDD`函数。即使这些函数并没有在`DStream`的`API`中暴露出来，通过该函数可以方便的扩展`Spark API`。该函数每一批次调度一次。其实也就是对`DStream`中的`RDD`应用转换。

**代码编写：**

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingTransform")
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        val lineDStream = ssc.socketTextStream("192.168.182.200", 9559)
    
        val wordCountDStream = lineDStream.transform(rdd => { 
          rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
        })
        wordCountDStream.print()
        //启动SparkStreamingContext
        ssc.start()
        ssc.awaitTermination()

测试：

-------------------------------------------
    Time: 1606463652000 ms
    -------------------------------------------
    (a,2)
    (b,1)
    
    -------------------------------------------
    Time: 1606463655000 ms
    -------------------------------------------
    (c,1)

### 2.1.2 join ###

两个流之间的`join`需要两个流的批次大小一致，这样才能做到同时触发计算。计算过程就是对当前批次的两个流中各自的`RDD`进行`join`，与两个`RDD`的`join`效果相同。

//1.创建SparkConf
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JoinTest")
    
        //2.创建StreamingContext
        val ssc = new StreamingContext(sparkConf, Seconds(5))
    
        //3.从端口获取数据创建流
        val lineDStream1: ReceiverInputDStream[String] = ssc.socketTextStream("linux1", 9999)
        val lineDStream2: ReceiverInputDStream[String] = ssc.socketTextStream("linux2", 8888)
    
        //4.将两个流转换为KV类型
        val wordToOneDStream: DStream[(String, Int)] = lineDStream1.flatMap(_.split(" ")).map((_, 1))
        val wordToADStream: DStream[(String, String)] = lineDStream2.flatMap(_.split(" ")).map((_, "a"))
    
        //5.流的JOIN
        val joinDStream: DStream[(String, (Int, String))] = wordToOneDStream.join(wordToADStream)
    
        //6.打印
        joinDStream.print()
    
        //7.启动任务
        ssc.start()
        ssc.awaitTermination()

## 2.2 有状态转化操作 ##

### 2.2.1 UpdateStateByKey ###

`UpdateStateByKey`原语用于记录历史记录，有时，我们需要在`DStream`中跨批次维护状态(例如流计算中累加`wordcount`)。针对这种情况，`updateStateByKey()`为我们提供了对一个状态变量的访问，用于键值对形式的`DStream`。给定一个由(键，事件)对构成的`DStream`，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的`DStream`，其内部数据为(键，状态) 对。

`UpdateStateByKey`的结果会是一个新的`DStream`，其内部的`RDD`序列是由每个时间区间对应的(键，状态)对组成的。

**代码实现：**

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        ssc.sparkContext.setCheckpointDir("cp")
    
        val lineDStream = ssc.socketTextStream("192.168.182.200", 9559)
    
        val wordCountDStream = lineDStream.
          flatMap(_.split(" "))
          .map((_, 1L))
          /** * updateStateByKey 是有状态计算方法 * 第一个参数表示 相同key的value集合 * 第二个参数表示 相同key的缓冲区的数据，有可能为空 * 这里中间结果需要保存到检查点的位置中，需要设定检查点 */
          .updateStateByKey(
            (seq: Seq[Long], buffer: Option[Long]) => { 
              val newBufferValue = buffer.getOrElse(0L) + seq.sum
              Option(newBufferValue)
            }
          )
        wordCountDStream.print()
        //启动 SparkStreamingContext
        ssc.start()
        ssc.awaitTermination()

**输入：**

[root@rich ~]# nc -lp 9559
    a
    a
    b

**控制台打印：**

-------------------------------------------
    Time: 1606478217000 ms
    -------------------------------------------
    (a,1)
    
    -------------------------------------------
    Time: 1606478223000 ms
    -------------------------------------------
    (a,2)
    (b,1)

### 2.2.2 WindowOperations ###

`Window Operations`可以设置窗口的大小和滑动窗口的间隔来动态的获取当前`Steaming`的允许状态。所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。

*  窗口时长：计算内容的时间范围；
 *  滑动步长：隔多久触发一次计算。

注意：这两者都必须为采集周期大小的整数倍。  
![在这里插入图片描述][20201128153615894.png]

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingTest8")
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        ssc.sparkContext.setCheckpointDir("cp")
    
        val lineDStream = ssc.socketTextStream("192.168.182.200", 9559)
    
        val wordCountDStream = lineDStream.
          flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKeyAndWindow((a: Int, b: Int) => (a + b), Seconds(12), Seconds(6))
        wordCountDStream.print()
        //启动SparkStreamingContext
        ssc.start()
        ssc.awaitTermination()

关于`Window`的操作还有如下方法：

*  `window(windowLength, slideInterval)`: 基于对源`DStream`窗化的批次进行计算返回一个新的`Dstream`；
 *  `countByWindow(windowLength, slideInterval)`: 返回一个滑动窗口计数流中的元素个数；
 *  `reduceByWindow(func, windowLength, slideInterval)`: 通过使用自定义函数整合滑动区间流元素来创建一个新的单元素流；
 *  `reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])`: 当在一个`(K,V)`对的`DStream`上调用此函数，会返回一个新`(K,V)`对的`DStream`，此处通过对滑动窗口中批次数据使用`reduce`函数来整合每个`key`的`value`值。

# 三、DStream输出 #

输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库或输出到屏幕上)。与`RDD`中的惰性求值类似，如果一个`DStream`及其派生出的`DStream`都没有被执行输出操作，那么这些`DStream`就都不会被求值。如果`StreamingContext`中没有设定输出操作，整个`context`就都不会启动。

输出操作如下：

*  **print()**：在运行流程序的驱动结点上打印`DStream`中每一批次数据的最开始10个元素。这用于开发和调试。在`Python API`中，同样的操作叫`print()`。
 *  **saveAsTextFiles(prefix, \[suffix\])**：以`text`文件形式存储这个`DStream`的内容。每一批次的存储文件名基于参数中的`prefix`和`suffix`。`prefix-Time_IN_MS[.suffix]`。
 *  **saveAsObjectFiles(prefix, \[suffix\])**：以`Java`对象序列化的方式将`Stream`中的数据保存为`SequenceFiles`。每一批次的存储文件名基于参数中的为`prefix-TIME_IN_MS[.suffix]`，`Python`中目前不可用。
 *  **saveAsHadoopFiles(prefix, \[suffix\])**：将`Stream`中的数据保存为 `Hadoop files`。每一批次的存储文件名基于参数中的为`prefix-TIME_IN_MS[.suffix]`。`Python API`中目前不可用。
 *  **foreachRDD(func)**：这是最通用的输出操作，即将函数`func`用于产生于`stream`的每一个`RDD`。其中参数传入的函数`func`应该实现将每一个`RDD`中数据推送到外部系统，如将`RDD`存入文件或者通过网络将其写入数据库。

# 四、优雅关闭 #

流式任务需要7\*24小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所有配置优雅的关闭就显得至关重要了。

使用外部文件系统来控制内部程序关闭。

class MonitorStop(ssc: StreamingContext) extends Runnable { 
      override def run(): Unit = { 
        val fs = FileSystem.get(new URI("hdfs://hadoop100:9000"),
          new Configuration(), "root")
        while (true) { 
          try
            Thread.sleep(5000)
          catch { 
            case e: InterruptedException =>
              e.printStackTrace()
          }
          val state = ssc.getState()
          val isExists = fs.exists(new Path("hdfs://hadoop100:9000/stopSpark"))
          if (isExists) { 
            if (state == StreamingContextState.ACTIVE) { 
              ssc.stop(stopSparkContext = true, stopGracefully = true)
              System.exit(0)
            }
          }
        }
      }
    }

主函数代码：

def main(args: Array[String]): Unit = { 
        ......
        new Thread(new MonitorStop(ssc)).start()
        ssc.start()
        ssc.awaitTermination()
      }

[20201127111714805.png]: /images/20221120/d46e0ea90a96466c96c6e5330cdd215e.png
[20201128153615894.png]: /images/20221120/51e0d0a02f1545b2ae74c781915059c2.png