Spark中ContinuousExecution執行流程是怎么樣的

這篇文章主要介紹Spark中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中ContinuousExecution執行流程是怎么樣的”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業資訊頻道！