免费观看又色又爽又黄的小说免费_美女福利视频国产片_亚洲欧美精品_美国一级大黄大色毛片

如何進行JobScheduler內幕實現和深度思考

本篇文章為大家展示了如何進行JobScheduler內幕實現和深度思考,內容簡明扼要并且容易理解,絕對能使你眼前一亮,通過這篇文章的詳細介紹希望你能有所收獲。

創新互聯專業為企業提供五華網站建設、五華做網站、五華網站設計、五華網站制作等企業網站建設、網頁設計與制作、五華企業網站模板建站服務,十多年五華做網站經驗,不只是建網站,更提供有價值的思路和整體網絡服務。

DStream的foreachRDD方法,實例化ForEachDStream對象,并將用戶定義的函數foreachFunc傳入到該對象中。foreachRDD方法是輸出操作,foreachFunc方法會作用到這個DStream中的每個RDD。

/**
 * Apply a function to each RDD in this DStream. This is an output operator, so
 * 'this' DStream will be registered as an output stream and therefore materialized.
 * @param foreachFuncforeachRDD function
 * @param displayInnerRDDOpsWhether the detailed callsites and scopes of the RDDs generated
 *                           in the `foreachFuncto be displayed in the UI. If `false`, then
 *                           only the scopes and callsites of `foreachRDDwill override those
 *                           of the RDDs on the display.
 */
private defforeachRDD(
    foreachFunc: (RDD[T], Time) => Unit,
    displayInnerRDDOps: Boolean): Unit = {
  newForEachDStream(this,
    context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
}

ForEachDStream對象中重寫了generateJob方法,調用父DStream的getOrCompute方法來生成RDD并封裝Job,傳入對該RDD的操作函數foreachFunc和time。dependencies方法定義為父DStream的集合。

/**
 * An internal DStream used to represent output operations like DStream.foreachRDD.
 * @param parent        Parent DStream
 * @param foreachFunc   Function to apply on each RDD generated by the parent DStream
 * @param displayInnerRDDOpsWhether the detailed callsites and scopes of the RDDs generated
 *                           by `foreachFuncwill be displayed in the UI; only the scope and
 *                           callsite of `DStream.foreachRDDwill be displayed.
 */
private[streaming]
classForEachDStream[T: ClassTag] (
    parent: DStream[T],
    foreachFunc: (RDD[T], Time) => Unit,
    displayInnerRDDOps: Boolean
  ) extendsDStream[Unit](parent.ssc) {

  override defdependencies: List[DStream[_]] = List(parent)

  override defslideDuration: Duration = parent.slideDuration

  override defcompute(validTime: Time): Option[RDD[Unit]] = None

  override defgenerateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match{
      caseSome(rdd) =>
        valjobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
          foreachFunc(rdd, time)
        }
        Some(newJob(time, jobFunc))
      caseNone => None
    }
  }
}

DStreamGraph的generateJobs方法中會調用outputStream的generateJob方法,就是調用ForEachDStream的generateJob方法。

defgenerateJobs(time: Time): Seq[Job] = {
  logDebug("Generating jobs for time "+ time)
  valjobs = this.synchronized {
    outputStreams.flatMap { outputStream =>
      valjobOption = outputStream.generateJob(time)
      jobOption.foreach(_.setCallSite(outputStream.creationSite))
      jobOption
    }
  }
  logDebug("Generated "+ jobs.length + " jobs for time "+ time)
  jobs
}

DStream的generateJob定義如下,其子類中只有ForEachDStream重寫了generateJob方法。

/**
 * Generate a SparkStreaming job for the given time. This is an internal method that
 * should not be called directly. This default implementation creates a job
 * that materializes the corresponding RDD. Subclasses of DStream may override this
 * to generate their own jobs.
 */
private[streaming] defgenerateJob(time: Time): Option[Job] = {
  getOrCompute(time) match{
    caseSome(rdd) => {
      valjobFunc = () => {
        valemptyFunc = { (iterator: Iterator[T]) => {} }
        context.sparkContext.runJob(rdd, emptyFunc)
      }
      Some(newJob(time, jobFunc))
    }
    caseNone => None
  }
}

DStream的print方法內部還是調用foreachRDD來實現,傳入了內部方法foreachFunc,來取出num+1個數后打印輸出。

/**
 * Print the first num elements of each RDD generated in this DStream. This is an output
 * operator, so this DStream will be registered as an output stream and there materialized.
 */
defprint(num: Int): Unit = ssc.withScope {
  defforeachFunc: (RDD[T], Time) => Unit = {
    (rdd: RDD[T], time: Time) => {
      val firstNum = rdd.take(num + 1)
      // scalastyle:off println
      println("-------------------------------------------")
      println("Time: "+ time)
      println("-------------------------------------------")
      firstNum.take(num).foreach(println)
      if(firstNum.length > num) println("...")
      println()
      // scalastyle:on println
    }
  }
  foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
}

總結:JobScheduler是SparkStreaming 所有Job調度的中心,內部有兩個重要的成員:

JobGenerator負責Job的生成,ReceiverTracker負責記錄輸入的數據源信息。

JobScheduler的啟動會導致ReceiverTracker和JobGenerator的啟動。ReceiverTracker的啟動導致運行在Executor端的Receiver啟動并且接收數據,ReceiverTracker會記錄Receiver接收到的數據meta信息。JobGenerator的啟動導致每隔BatchDuration,就調用DStreamGraph生成RDD Graph,并生成Job。JobScheduler中的線程池來提交封裝的JobSet對象(時間值,Job,數據源的meta)。Job中封裝了業務邏輯,導致最后一個RDD的action被觸發,被DAGScheduler真正調度在Spark集群上執行該Job。

上述內容就是如何進行JobScheduler內幕實現和深度思考,你們學到知識或技能了嗎?如果還想學到更多技能或者豐富自己的知識儲備,歡迎關注創新互聯行業資訊頻道。

分享標題:如何進行JobScheduler內幕實現和深度思考
本文來源:http://m.newbst.com/article26/ispgjg.html

成都網站建設公司_創新互聯,為您提供面包屑導航網站設計標簽優化自適應網站網站收錄App設計

廣告

聲明:本網站發布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網站立場,如需處理請聯系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經允許不得轉載,或轉載時需注明來源: 創新互聯

微信小程序開發