这篇文章主要针对SparkStreaming的三个特性进行说明。
分别是Dstream Window checkpoint
sparkStreaming是一个实时数据流框架。数据可以从多种源输入，比如Kafka,Flume,TCP socket等，可以对输入进来的数据进行一系列处理，最终输出。
在内部处理时，会将流分解为多个批batch。最终交由spark处理。
Dstream
官方有这样的解释：Dstream(Discretized Stream)意为离散化流，是Spark Streaming提供的基本抽象，他代表了一个连续的数据流，无论是从数据源接受的数据流还是处理过程中产生的数据流。Dstream的内部属性中一串连续的RDD。每个RDD在一定间隔时间后产生。
对于每个Dstream的操作，最终都会变为对RDD的操作，所以Streaming操作最后会交给Spark引擎(engine)进行计算。
我们可以通过源码进行验证。
首先我们以一个程序为例
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(5));
ssc.checkpoint(".");
// Initial state RDD input to mapWithState
@SuppressWarnings("unchecked")
List<Tuple2<String, Integer>> tuples =
Arrays.asList(new Tuple2<>("hello", 1), new Tuple2<>("world", 1));
JavaPairRDD<String, Integer> initialRDD = ssc.sparkContext().parallelizePairs(tuples);
JavaReceiverInputDStream<String> lines = ssc.socketTextStream(
"localhost",9999, StorageLevels.MEMORY_AND_DISK_SER_2);
JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator());
JavaPairDStream<String, Integer> wordsDstream = newWords.mapToPair(s -> new Tuple2<>(s, 1));
该程序第二步生成一个JavaReceiverInputDStream对象，由于是通过socketTextStream获得，所以获得了的是JavaReceiverInputDStream对象。
private[streaming] class SocketInputDStream[T](
_ssc : org.apache.spark.streaming.StreamingContext,
host : scala.Predef.String,
port : scala.Int,
bytesToObjects : scala.Function1[java.io.InputStream, scala.Iterator[T]], storageLevel : org.apache.spark.storage.StorageLevel)
(implicit evidence$1 : scala.reflect.ClassTag[T])
extends org.apache.spark.streaming.dstream.ReceiverInputDStream[T] {...}
这里是从socket端获得流的Dstream它继承的是ReceiverInputDStream。
而这个ReceiverInputDStream是一个重要的类
abstract class ReceiverInputDStream[T](
_ssc :org.apache.spark.streaming.StreamingContext)
(implicit evidence$1 : scala.reflect.ClassTag[T])
extends org.apache.spark.streaming.dstream.InputDStream[T] {
override def compute(validTime : org.apache.spark.streaming.Time) : scala.Option[org.apache.spark.rdd.RDD[T]] = { /* compiled code */ }
....
}
它重写了compute方法，这个方法是用来生成RDD的。
然后看它继承的父类InputDStream
abstract class InputDStream[T](
_ssc : org.apache.spark.streaming.StreamingContext)
(implicit evidence$1 : scala.reflect.ClassTag[T])
extends org.apache.spark.streaming.dstream.DStream[T] {
override def dependencies : scala.List[org.apache.spark.streaming.dstream.DStream[_]] = { /* compiled code */ }
override def slideDuration : org.apache.spark.streaming.Duration = { /* compiled code */ }
}
最后终于来到了Dstream了
abstract class DStream[T](
@scala.transient private[streaming] var ssc : org.apache.spark.streaming.StreamingContext)
(implicit evidence$1 : scala.reflect.ClassTag[T])
extends scala.AnyRef with scala.Serializable with org.apache.spark.internal.Logging {
private[streaming] var generatedRDDs : scala.collection.mutable.HashMap[org.apache.spark.streaming.Time, org.apache.spark.rdd.RDD[T]] = { /* compiled code */ }
private[streaming] var zeroTime : org.apache.spark.streaming.Time = { /* compiled code */ }
private[streaming] var rememberDuration : org.apache.spark.streaming.Duration = { /* compiled code */ }
private[streaming] var storageLevel : org.apache.spark.storage.StorageLevel = { /* compiled code */ }
private[streaming] val mustCheckpoint : scala.Boolean = { /* compiled code */ }
private[streaming] var checkpointDuration : org.apache.spark.streaming.Duration = { /* compiled code */ }
private[streaming] val checkpointData : org.apache.spark.streaming.dstream.DStreamCheckpointData[T] = { /* compiled code */ }
private[streaming] var graph : org.apache.spark.streaming.DStreamGraph = { /* compiled code */ }
}
这里我们注重关注它的属性：
generatedRDDs 用来放RDD的一个Hash表，key为time
其他的自行了解，有开始时间，checkPoint点，时间间隔，图等等。
下面我们来看看Dstream怎么生成RDD的。
由例子往下走
private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {
// If RDD was already generated, then retrieve it from HashMap,
// or else compute the RDD
generatedRDDs.get(time).orElse {
// Compute the RDD if time is valid (e.g. correct time in a sliding window)
// of RDD generation, else generate nothing.
if (isTimeValid(time)) {
val rddOption = createRDDWithLocalProperties(time, displayInnerRDDOps = false) {
// Disable checks for existing output directories in jobs launched by the streaming
// scheduler, since we may need to write output to an existing directory during checkpoint
// recovery; see SPARK-4835 for more details. We need to have this call here because
// compute() might cause Spark jobs to be launched.
SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
compute(time)
}
}
override def compute(validTime: Time): Option[RDD[U]] = {
parent.getOrCompute(validTime).map(_.map[U](mapFunc))
}
Dstream实现了getOrCompute方法，该方法会根据该RDD是否在Hash表中进行判断，没有则创建有就提取。
map,reduce等操作都重写了compute方法但是都是调用父类的方法。
关于定时任务，job调度和生成详见博客
https://www.cnblogs.com/sparkbigdata/p/5513339.html
https://blog.csdn.net/xiaojun220/article/details/51441242
Window
window的思想和TCP中window十分类似，用一个窗口化操作，对数据偏移进行控制。每次想操作多少秒的数据，多久进行窗口移动都是可设定值。sparkStreaming也提供了相关的API可以针对不同操作进行使用，官网很详细，必须再说明
http://spark.apachecn.org/docs/cn/2.2.0/streaming-programming-guide.html#input-dstreams-%E5%92%8C-receivers%E6%8E%A5%E6%94%B6%E5%99%A8
checkpoint
通过整理几篇博文进行理解。
checkpoint意为检查点。类似于快照的意思。如果在spark计算中，DAG十分巨大，出错后因为丢失了中间数据而需要重新计算是很费事的。这时候，将文件写入缓存或者是持久化写入硬盘是很必要的。但是也不能保证硬盘不会出问题。所以我们需要检查故有了检查点，他的作用是将重要的中间数据写入一个高可用的地方。
3
