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對於流處理的支持設計的很是巧妙的，很是簡潔，值得好好學習。