执行流程

数据的接收

StreamingContext实例化的时候，需要传入一个 SparkContext，然后指定要连接的 spark matser url，即连接一个 spark engine，用于获得executor。

实例化之后，首先，要指定一个接收数据的方式，如

  val lines = ssc.socketTextStream("localhost", 9999)

这样从socket接收文本数据。这个步骤返回的是一个 ReceiverInputDStream的实现，内含 Receiver，可接收数据并转化为RDD放内存里。

ReceiverInputDStream有一个需要子类实现的方法

  def getReceiver(): Receiver[T]

子类实现这个方法，worker节点调用后能得到 Receiver，使得数据接收的工作能分布到worker上。

如果是local跑，由于 Receiver接收数据在本地，所以在启动streaming application的时候，要注意分配的core数目要大于 Receiver数目，才能腾出cpu做计算任务的调度。

Receiver需要子类实现

  def onStart()
def onStop()

来定义一个数据接收器的初始化、接收到数据后如何存、如何在结束的时候释放资源。

Receiver提供了一系列 store()接口，如 store(ByteBuffer)， store(Iterator)等等。这些store接口是实现好了的，会由worker节点上初始化的 ReceiverSupervisor来完成这些存储功能。 ReceiverSupervisor还会对 Receiver做监控，如监控是否启动了、是否停止了、是否要重启、汇报error等等。

ReceiverSupervisor的存储接口的实现，借助的是 BlockManager，数据会以RDD的形式被存放，根据 StorageLevel选择不同存放策略。默认是序列化后存内存，放不下的话写磁盘(executor)。被计算出来的RDD中间结果，默认存放策略是序列化后只存内存。

ReceiverSupervisor在做putBlock操作的时候，会首先借助 BlockManager存好数据，然后往 ReceiverTracker发送一个 AddBlock的消息。 ReceiverTracker内部的 ReceivedBlockTracker用于维护一个receiver接收到的所有block信息，即 BlockInfo，所以 AddBlock会把信息存放在 ReceivedBlockTracker里。未来需要计算的时候， ReceiverTracker根据streamId，从 ReceivedBlockTracker取出对应的block列表。

RateLimiter帮助控制 Receiver速度， spark.streaming.receiver.maxRate参数。

数据源方面，普通的数据源为file, socket, akka, RDDs。高级数据源为Twitter, Kafka, Flume等。开发者也可以自己定制数据源。

任务调度

JobScheduler在context里初始化。当context start的时候，触发scheduler的start。

scheduler的start触发了 ReceiverTracker和 JobGenerator的start。这两个类是任务调度的重点。前者在worker上启动 Receiver接收数据，并且暴露接口能够根据streamId获得对应的一批Block地址。后者基于数据和时间来生成任务描述。

JobScheduler内含一个线程池，用于调度任务执行。 spark.streaming.concurrentJobs可以控制job并发度，默认是1，即它只能一个一个提job。

job来自 JobGenerator生成的 JobSet。 JobGenerator根据时间，生成job并且执行cp。

JobGenerator的生成job逻辑：
- 调用 ReceiverTracker的 allocateBlocksToBatch方法，为本批数据分配好block，即准备好数据
- 间接调用 DStream的 generateJob(time)方法，制造可执行的RDD

DStream切分RDD和生成可执行的RDD，即 getOrCompute(time)：
- 如果这个时间点的RDD已经生成好了，那么从内存hashmap里拿出来，否则下一步
- 如果时间是批次间隔的整数倍，则下一步，否则这个时间点不切
- 调用DStream的子类的 compute方法，得到RDD。可能是一个RDD，也可以是个RDD列表
- 对每个RDD,调用persist方法，制定默认的存储策略。如果时间点合适，同时调用RDD的 checkpoint方法，制定好cp策略
- 得到这些RDD后，调用 SparkContext.runJob(rdd, emptyFunction)。把这整个变成一个function，生成 Job类。未来会在executor上触发其 runJob

JobGenerator成功生成job后，调用 JobScheduler.submitJobSet(JobSet)， JobScheduler会使用线程池提交JobSet中的所有job。该方法调用结束后， JobGenerator发送一个 DoCheckpoint的消息， 注意这里的cp是driver端元数据的cp，而不是RDD本身的cp。如果time合适，会触发cp操作，内部的 CheckpointWriter类会完成 write(streamingContext, time)。

JobScheduler提交job的线程里，触发了job的run()方法，同时，job跑完后， JobScheduler处理 JobCompleted(job)。如果job跑成功了，调用 JobSet的 handleJobCompletion(Job)，做些计时和数数工作，如果整个JobSet完成了，调用 JobGenerator的 onBatchCompletion(time)方法， JobGenerator接着会做 clearMetadata的工作，然后 JobScheduler打印输出；如果job跑失败了， JobScheduler汇报error，最后会在context里抛异常。

更多说明

特殊操作

1. transform：可以与外部RDD交互，比如做维表的join

2. updateStateByKey：生成 StateDStream，比如做增量计算。 WordCount例子

   • 每一批都需要与增量RDD进行一次cogroup之后，然后执行update function。两个RDD做cogroup过程有些开销：RDD[K, V]和RDD[K, U]合成RDD[K, List[V], List[U]]，List[U]一般size是1，理解为oldvalue，即RDD[K, batchValueList, Option[oldValue]]。然后update function处理完，变成RDD[K, newValue]。
   • 批与批之间严格有序，即增量合并操作，是有序的，批之间没发并发
   • 增量RDD的分区数可以开大，即这步增量的计算可以调大并发

3. window：batch size，window length, sliding interval三个参数组成的滑窗操作。把多个批次的RDD合并成一个UnionRDD进行计算。

4. foreachRDD: 这个操作是一个输出操作，比较特殊。

    /**
   * Apply a function to each RDD in this DStream. This is an output operator, so
   * 'this' DStream will be registered as an output stream and therefore materialized.
   */
  def foreachRDD(foreachFunc: (RDD[T], Time) => Unit) {
    new ForEachDStream(this, context.sparkContext.clean(foreachFunc, false)).register()
  }

DStream.foreachRDD()操作使开发者可以直接控制RDD的计算逻辑，而不是通过 DStream映射过去。所以借助这个方法，可以实现MLlib, Spark SQL与Streaming的集合，如： 结合Spark SQL、DataFrame做Wordcount。

Cache

如果是window操作，默认接收的数据都persist在内存里。

如果是flume, kafka源头，默认接收的数据replicate成两份存起来。

Checkpoint

与state有关的流计算，计算出来的结果RDD，会被cp到HDFS上，原文如下：

Data checkpointing - Saving of the generated RDDs to reliable storage. This is necessary in some stateful transformations that combine data across multiple batches. In such transformations, the generated RDDs depends on RDDs of previous batches, which causes the length of the dependency chain to keep increasing with time. To avoid such unbounded increase in recovery time (proportional to dependency chain), intermediate RDDs of stateful transformations are periodically checkpointed to reliable storage (e.g. HDFS) to cut off the dependency chains.

cp的时间间隔也可以设定，可以多批做一次cp。

cp的操作是同步的。

简单的不带state操作的流任务，可以不开启cp。

driver端的metadata也有cp策略。driver cp的时候是将整个 StreamingContext对象写到了可靠存储里。

全文完 :)