本篇博客將詳細(xì)探討DStream模板下的RDD是如何被創(chuàng)建，然后被執(zhí)行的。在開始敘述之前，先來(lái)思考幾個(gè)問(wèn)題，本篇文章也就是基于此問(wèn)題構(gòu)建的。 
1. RDD是誰(shuí)產(chǎn)生的？ 
2. 如何產(chǎn)生RDD？ 
帶著這兩個(gè)問(wèn)題開啟我們的探索之旅。

創(chuàng)新互聯(lián)是專業(yè)的紅崗網(wǎng)站建設(shè)公司，紅崗接單;提供做網(wǎng)站、網(wǎng)站制作,網(wǎng)頁(yè)設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行紅崗網(wǎng)站開發(fā)網(wǎng)頁(yè)制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來(lái)合作!

DStream是RDD的模板，每隔一個(gè)Batch Interval會(huì)根據(jù)DStream模板生成一個(gè)對(duì)應(yīng)的RDD，然后將RDD存儲(chǔ)到DStream中的generatedRDDs數(shù)據(jù)結(jié)構(gòu)中，下面是存儲(chǔ)結(jié)構(gòu)格式。

// RDDs generated, marked as private[streaming] so that testsuites can access it
@transient
private[streaming] var generatedRDDs = new HashMap[Time, RDD[T]] ()

1、簡(jiǎn)單的WordCount程序

object WordCount {  def main(args:Array[String]): Unit ={
    val sparkConf = new SparkConf().setMaster("Master:7077").setAppName("WordCount")
    val ssc = new StreamingContext(sparkConf,Seconds(10)) // Timer觸發(fā)頻率

    val lines = ssc.socketTextStream("Master",9999) //接收數(shù)據(jù)
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x,1)).reduceByKey(_+_)
    wordCounts.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

首先我們先看看print方法，具體的代碼如下：

/**
* 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.
*/
def print(num: Int): Unit = ssc.withScope {
 def foreachFunc: (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)
}

首先定義了一個(gè)函數(shù)，該函數(shù)用來(lái)從RDD中取出前幾條數(shù)據(jù)，并打印出結(jié)果與時(shí)間等，后面會(huì)調(diào)用foreachRDD函數(shù)。

private def foreachRDD(
   foreachFunc: (RDD[T], Time) => Unit,
   displayInnerRDDOps: Boolean): Unit = {
   new ForEachDStream(this,context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
}

/**
* Register this streaming as an output stream. This would ensure that RDDs of this
* DStream will be generated.
*/
private[streaming] def register(): DStream[T] = {
 ssc.graph.addOutputStream(this)
 this
}

def addOutputStream(outputStream: DStream[_]) {
 this.synchronized {
   outputStream.setGraph(this)
   outputStreams += outputStream
 }

在foreachRDD中new出了一個(gè)ForEachDStream對(duì)象，并將這個(gè)注冊(cè)給DStreamGraph，F(xiàn)orEachDStream對(duì)象也就是DStreamGraph中的outputStreams。

當(dāng)每到達(dá)一個(gè)BatchInterval時(shí)候，就會(huì)調(diào)用DStreamingGraph中的generateJobs.

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

這里就會(huì)調(diào)用outputStream的generateJob方法

private[streaming] def generateJob(time: Time): Option[Job] = {
 getOrCompute(time) match {
   case Some(rdd) => {
     val jobFunc = () => {
       val emptyFunc = { (iterator: Iterator[T]) => {} }
       context.sparkContext.runJob(rdd, emptyFunc)
     }
     Some(new Job(time, jobFunc))
   }
   case None => None
 }
}

這里會(huì)調(diào)用getOrCompute(time)來(lái)產(chǎn)生新RDD，并將其存入到generatedRDDs中，整理的過(guò)程如下圖：


本文標(biāo)題：(版本定制)第8課：SparkStreaming源碼解讀之
當(dāng)前路徑：http://jinyejixie.com/article2/jjicoc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供服務(wù)器托管、做網(wǎng)站、微信小程序、Google、關(guān)鍵詞優(yōu)化、移動(dòng)網(wǎng)站建設(shè)

廣告

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