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

成都創(chuàng)新互聯(lián)公司-專業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設(shè)、高性價比臺前網(wǎng)站開發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫,直接使用。一站式臺前網(wǎng)站制作公司更省心,省錢,快速模板網(wǎng)站建設(shè)找我們，業(yè)務(wù)覆蓋臺前地區(qū)。費用合理售后完善，10多年實體公司更值得信賴。

最重要的是看ContinuousExecution怎么重寫LogicalPlan的，詳細代碼不貼了，最后是創(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本身沒啥，只是在每一個處理的時候包裝了一些額外的輔助處理而已。

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

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

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

原來是DataSourceV2Relation的子類?。?/p>

直接找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對應(yīng)的物理計劃是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)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！

新聞名稱：Spark中ContinuousExecution執(zhí)行流程是怎么樣的
轉(zhuǎn)載來于：http://www.rwnh.cn/article16/jsdcdg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供搜索引擎優(yōu)化、響應(yīng)式網(wǎng)站、關(guān)鍵詞優(yōu)化、移動網(wǎng)站建設(shè)、定制開發(fā)、網(wǎng)站建設(shè)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)