Spark从入门到精通31：DStream对象详解

Spark Streaming就是将连续的数据流切分成离散的数据流，即DStream。本节就来介绍DStream的详细知识。

1.什么是DStream

Discretized Stream（简称DStream）是Spark Streaming对流式数据的基本抽象。它表示离散的数据流，这些离散的数据流可以是直接从数据源接收的输入数据流，也可以是通过对输入数据流执行转换操作而生成的经处理的数据流。

DStream内部是一个个时间片上的RDD，如下图所示：

image

举例分析：在之前的NetworkWordCount例子中，我们将一行行文本组成的数据流转换成单词流，具体做法是：将flatMap算子作用于名为lines的DStream中的每一个RDD上，以生成words DStream的RDD，像下面这样：

image

2.DStream和RDD的关系

我们知道，RDD的内部由一系列分区组成，而DStream的内部又是由一系列RDD组成，对DStream的操作实际上是对内部每个RDD的操作，它们二者的关系可以用下图表示：

image

3.DStream常用算子介绍

因为DStream的操作实际上是对内部RDD的操作，所以DStream的算子基本和RDD的算子相同。下面是一些DStream常用的transformation算子（action算子参考RDD的action算子）：

image

最后两个算子需要重点介绍一下：

（1）transform(func)：通过RDD-to-RDD类型的函数func作用于源DStream中的各个RDD，这里的func中可以是任意的RDD操作，完了返回一个新的RDD。

举例：在NetworkWordCount程序中，使用transform算子来生成元组对

package SparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

object NetworkWordCount{
    def main(args:Array[String]) {
        //配置：程序名字为NetworkWordCount，运行模式为本地，开启2个线程
        val sparkconf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[2]")
        //创建StreamingContext对象，设置批处理时间间隔为3秒
        val ssc = new StreamingContext(sparkconf, Seconds(3))
        //创建DStream对象，连接到Netcat服务器
        val lines = ssc.socketTextStream("192.168.126.110",9999,StorageLevel.MEMORY_AND_DISK_SER)
        //处理采集到的数据
        val words = lines.flatMap(_.split(" "))
        val wordCounts = words.**transform(rdd=>rdd.map(word=>(word,1)))**.reduceByKey(_+_)
        //输出结果
        wordCounts.print()
        //启动StreamingContext对象，开始采集处理数据
        ssc.start()
        //等待计算结束
        ssc.awaitTermination()
    }
}

（2）updateStateByKey(func)：该操作可以通过设置检查点来记录状态信息，在计算的过程中不断更新状态信息。func函数用于更新状态信息，这里的状态信息可以是任何数据类型。

举例：重写NetworkWordCount程序，这次来累计每个单词出现过的次数。

package SparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

object NetworkWordCountSum{
    def main(args:Array[String]) {
        //配置：程序名字为NetworkWordCount，运行模式为本地，开启2个线程
        val sparkconf = new SparkConf().setAppName("NetworkWordCountSum").setMaster("local[2]")
        //创建StreamingContext对象，设置批处理时间间隔为3秒
        val ssc = new StreamingContext(sparkconf, Seconds(3))
        //设置检查点
 ssc.checkPoint("hdfs://localhost:9000/spark/checkpoint")
        //创建DStream对象，连接到Netcat服务器
        val lines = ssc.socketTextStream("192.168.126.110",9999,StorageLevel.MEMORY_AND_DISK_SER)
        //处理采集到的数据
        val words = lines.flatMap(_.split(" "))
        val pairs = words.map(word=>(word,1))
        //定义状态更新函数：用于累加每个单词出现的次数
 //参数：curValues是某个单词在当前时间窗口内出现的次数的List，需要求和
 //参数：preValueState是该单词之前出现的总次数
        val addFunc = (curValues:Seq[Int], preValueState:Option[Int]) => {
 val curCount = curValues.sum
 val preCount = preValueState.getOrElse(0)
 Some(curCount + preCount)
 }
 val totalWordCounts = pairs.updateStateByKey[Int](addFunc)
        //输出结果
        totalWordCounts.print()
        //启动StreamingContext对象，开始采集处理数据
        ssc.start()
        //等待计算结束
        ssc.awaitTermination()
    }
}

小技巧：如果在IDEA中，不想输出log4j的日志INFO信息，可以将log4j.properties文件（放在src的目录下）的第一行改为：log4j.rootCategory=ERROR, console，这样只有出错的时候才打印日志。