Spark中ContinuousExecution執(zhí)行流程是怎么樣的

這篇文章主要介紹Spark中ContinuousExecution執(zhí)行流程是怎么樣的，文中介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！

最重要的是看ContinuousExecution怎么重寫LogicalPlan的，詳細(xì)代碼不貼了，最后是創(chuàng)建了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作為入?yún)?chuàng)建了IncrementalExecution。

triggerLogicalPlan是StreamingDataSourceV2Relation類。

IncrementalExecution本身沒啥，只是在每一個(gè)處理的時(shí)候包裝了一些額外的輔助處理而已。

WriteToDataSourceV2的作用是將triggerLogicalPlan的物理計(jì)劃的執(zhí)行結(jié)果通過writer寫入到外部存儲中，所有這里也不看WriteToDataSourceV2了，就看看triggerLogicalPlan的對應(yīng)的物理計(jì)劃是什么，前面說過了它對應(yīng)的邏輯計(jì)劃是：StreamingDataSourceV2Relation。

直接找是不是StreamingDataSourceV2Relation對應(yīng)的物理計(jì)劃的，所以我們先看看StreamingDataSourceV2Relation類的定義：

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

原來是DataSourceV2Relation的子類??！

直接找DataSourceV2Relation的物理計(jì)劃吧，在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對應(yīng)的物理計(jì)劃是DataSourceV2ScanExec。

DataSourceV2ScanExec的代碼也不多。

DataSourceV2ScanExec是用DataSourceReader來作為數(shù)據(jù)源的讀取器的，它的inputRDDs返回的是DataSourceRDD或者ContinuousDataSourceRDD，ContinuousDataSourceRDD肯定是對應(yīng)的ContinuousExecution，其他方式就是DataSourceRDD了。

不管是DataSourceRDD或者ContinuousDataSourceRDD，他們的讀取數(shù)據(jù)源的類都是一樣的，都是DataSourceReader過來的。DataSourceRDD或者ContinuousDataSourceRDD這兩者的代碼都非常少，一看就知道怎么回事了。

以上是“Spark中ContinuousExecution執(zhí)行流程是怎么樣的”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對大家有幫助，更多相關(guān)知識，歡迎關(guān)注億速云行業(yè)資訊頻道！