执行流程

数据的接收

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

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

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

• 1

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

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

  def getReceiver(): Receiver[T]

• 1

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

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

  Receiver需要子类实现

  def onStart()def onStop()

• 1
• 2

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

  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里抛异常。

  更多说明

  特殊操作

• transform：可以与外部RDD交互，比如做维表的join
• 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的分区数可以开大，即这步增量的计算可以调大并发
• window：batch size，window length, sliding interval三个参数组成的滑窗操作。把多个批次的RDD合并成一个UnionRDD进行计算。
• 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()

  }

• 1
• 2
• 3
• 4
• 5
• 6
• 7

  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对象写到了可靠存储里