Spark版本定制第8天：RDD生成生命周期彻底

Posted 2020-07-10

本期内容：

1 RDD生成生命周期

2 深度思考

　　一切不能进行实时流处理的数据都是无效的数据。在流处理时代，SparkStreaming有着强大吸引力，而且发展前景广阔，加之Spark的生态系统，Streaming可以方便调用其他的诸如SQL，MLlib等强大框架，它必将一统天下。

　　Spark Streaming运行时与其说是Spark Core上的一个流式处理框架，不如说是Spark Core上的一个最复杂的应用程序。如果可以掌握Spark streaming这个复杂的应用程序，那么其他的再复杂的应用程序都不在话下了。这里选择Spark Streaming作为版本定制的切入点也是大势所趋。

 

　　在sparkStreaming中DStream是RDD的模板，每隔一个batchInterval会根据DStream模板生成一个对应的RDD。生成的RDD会存放到generatedRDD中。生成了一系列的RDD后，会通过foreachRDD对RDD进行操作。

　　

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

　　在foreachRDD中new出了一个ForEachDStream对象。并将这个注册给DStreamGraph 。

　　当每一个batchInterval的时间内，会调用DStreamGraph中的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 
} 

　　在这里面会根据时间调用generateJob方法

override def generateJob(time: Time): Option[Job] = { 
  parent.getOrCompute(time) match { 
    case Some(rdd) => 
      val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) { 
        foreachFunc(rdd, time) 
      } 
      Some(new Job(time, jobFunc)) 
    case None => None 
  } 
} 

　　然后沿着graph回溯，再下一个batchInterval后生成新的RDD，不断循环，将所有生成的rdd都放入集合generatrdRDD中。

 

备注：

资料来源于：DT_大数据梦工厂（Spark发行版本定制）

更多私密内容，请关注微信公众号：DT_Spark

如果您对大数据Spark感兴趣，可以免费听由王家林老师每天晚上20：00开设的Spark永久免费公开课，地址YY房间号：68917580