Spark Streaming就是将连续的数据流切分成离散的数据流，即DStream。本节就来介绍DStream的详细知识。
1.什么是DStream
Discretized Stream（简称DStream）是Spark Streaming对流式数据的基本抽象。它表示离散的数据流，这些离散的数据流可以是直接从数据源接收的输入数据流，也可以是通过对输入数据流执行转换操作而生成的经处理的数据流。
DStream内部是一个个时间片上的RDD，如下图所示：
举例分析：在之前的NetworkWordCount例子中，我们将一行行文本组成的数据流转换成单词流，具体做法是：将flatMap算子作用于名为lines的DStream中的每一个RDD上，以生成words DStream的RDD，像下面这样：
2.DStream和RDD的关系
我们知道，RDD的内部由一系列分区组成，而DStream的内部又是由一系列RDD组成，对DStream的操作实际上是对内部每个RDD的操作，它们二者的关系可以用下图表示：
3.DStream常用算子介绍
因为DStream的操作实际上是对内部RDD的操作，所以DStream的算子基本和RDD的算子相同。下面是一些DStream常用的transformation算子（action算子参考RDD的action算子）：
最后两个算子需要重点介绍一下：
（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，这样只有出错的时候才打印日志。
