一、DStream的定义

DStream是离散流，Spark Streaming提供的一种高级抽象，代表了一个持续不断的数据流。DStream可以通过输入数据源来创建，比如Kafka、Flume，也可以通过对其他DStream应用高阶函数来创建，比如map、reduce、join、window。
DStream的内部，其实是一系列持续不断产生的RDD，RDD是Spark Core的核心抽象，即不可变的，分布式的数据集。
对DStream应用的算子，其实在底层会被翻译为对DStream中每个RDD的操作，比如对一个DStream执行一个map操作，会产生一个新的DStream，其底层原理为，对输入DStream中的每个时间段的RDD，都应用一遍map操作，然后生成的RDD，即作为新的DStream中的那个时间段的一个RDD。

二、DStream的操作

1，普通的转换操作

2，窗口转换函数

在Spark Streaming中，数据处理是按批进行的，而数据采集是逐条进行的。因此在Spark Streaming中会先设置好批处理间隔（batch duration），当超过批处理间隔的时候就会把采集到的数据汇总起来成为一批数据交给系统去处理。

对于窗口操作而言，在其窗口内部会有N个批处理数据，批处理数据的大小由窗口间隔（window duration）决定，而窗口间隔指的就是窗口的持续时间，在窗口操作中，只有窗口的长度满足了才会触发批数据的处理。除了窗口的长度，窗口操作还有另一个重要的参数就是滑动间隔（slide duration），它指的是经过多长时间窗口滑动一次形成新的窗口，滑动窗口默认情况下和批次间隔的相同，而窗口间隔一般设置的要比它们两个大。

3，输出操作

三、常用操作详解

1，transform(func)

该transform操作（转换操作）及其类似的transformWith操作，允许在DStream上应用任意的RDD-to-RDD函数。它可以实现DStream API中未提供的操作，比如两个数据流的连接操作。
示例代码：

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information

val cleanedDStream = wordCounts.transform { rdd =>
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
}

2，updateStateByKey操作

使用的一般操作都是不记录历史数据的，也就说只记录当前定义时间段内的数据，跟前后时间段无关。如果要统计历史时间内的总共数据并且实时更新，如何解决呢？该updateStateByKey操作可以让你保持任意状态，同时不断有新的信息进行更新。

要使用updateStateByKey操作，必须进行下面两个步骤 ：
（1）定义状态： 状态可以是任意的数据类型。
（2）定义状态更新函数：用一个函数指定如何使用先前的状态和从输入流中获取的新值更新状态。
对DStream通过updateStateByKey(updateFunction)来实现实时更新。

更新函数有两个参数 ：
（1）newValues是当前新进入的数据。
（2）runningCount 是历史数据，被封装到了Option中。

示例：
首先我们需要了解数据的类型
编写处理方法
封装结果
代码：

//定义更新函数
//我们这里使用的Int类型的数据，因为要做统计个数
def updateFunc(newValues : Seq[Int],state :Option[Int]) ：Some[Int] = {
 //传入的newVaules将当前的时间段的数据全部保存到Seq中
 //调用foldLeft(0)(_+_) 从0位置开始累加到结束   
 val currentCount = newValues.foldLeft(0)(_+_) 
 //获取历史值,没有历史数据时为None，有数据的时候为Some
 //getOrElse（x）方法，如果获取值为None则用x代替
 val  previousCount = state.getOrElse(0)
 //计算结果，封装成Some返回
 Some(currentCount+previousCount) 
}
//使用
val stateDStream = DStream.updateStateByKey[Int](updateFunc)