Spark的ContinuousExecution执行流程

前一章说过，spark sql的连续流处理经过ContinuousExecution执行，这一章看看ContinuousExecution是怎么往下一步步实现的。

最重要的是看ContinuousExecution怎么重写LogicalPlan的，详细代码不贴了，最后是建立了Sink类型的LogicalPlan。

val writer = sink.createStreamWriter(
      s"$runId",
      triggerLogicalPlan.schema,
      outputMode,
      new DataSourceOptions(extraOptions.asJava))
    val withSink = WriteToDataSourceV2(writer, triggerLogicalPlan)

    val reader = withSink.collect {
      case DataSourceV2Relation(_, r: ContinuousReader) => r
    }.head

这里的sink能够当作就是DataSource。而后用withSink做为入参建立了IncrementalExecution。

triggerLogicalPlan是StreamingDataSourceV2Relation类。

IncrementalExecution自己没啥，只是在每个处理的时候包装了一些额外的辅助处理而已。

WriteToDataSourceV2的做用是将triggerLogicalPlan的物理计划的执行结果经过writer写入到外部存储中，全部这里也不看WriteToDataSourceV2了，就看看triggerLogicalPlan的对应的物理计划是什么，前面说过了它对应的逻辑计划是：StreamingDataSourceV2Relation。

直接找是否是StreamingDataSourceV2Relation对应的物理计划的，因此咱们先看看StreamingDataSourceV2Relation类的定义：

class StreamingDataSourceV2Relation(
    output: Seq[AttributeReference],
    reader: DataSourceReader) extends DataSourceV2Relation(output, reader) {
  override def isStreaming: Boolean = true
}

原来是DataSourceV2Relation的子类啊！

直接找DataSourceV2Relation的物理计划吧，在DataSourceV2Strategy.scala文件中定义了。

object DataSourceV2Strategy extends Strategy {
  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
    case DataSourceV2Relation(output, reader) =>
      DataSourceV2ScanExec(output, reader) :: Nil

    case WriteToDataSourceV2(writer, query) =>
      WriteToDataSourceV2Exec(writer, planLater(query)) :: Nil

    case _ => Nil
  }
}

DataSourceV2Relation对应的物理计划是DataSourceV2ScanExec。

DataSourceV2ScanExec的代码也很少。

DataSourceV2ScanExec是用DataSourceReader来做为数据源的读取器的，它的inputRDDs返回的是DataSourceRDD或者ContinuousDataSourceRDD，ContinuousDataSourceRDD确定是对应的ContinuousExecution，其余方式就是DataSourceRDD了。

无论是DataSourceRDD或者ContinuousDataSourceRDD，他们的读取数据源的类都是同样的，都是DataSourceReader过来的。DataSourceRDD或者ContinuousDataSourceRDD这二者的代码都很是少，一看就知道怎么回事了。

从这分析能够看出，Spark sql对于流处理的支持设计的很是巧妙的，很是简洁，值得好好学习。