设为首页 加入收藏

TOP

Spark Streaming 数据计算阶段分析
2019-02-13 01:09:26 】 浏览:16
Tags:Spark Streaming 数据 计算 阶段 分析

——————————————————————————————
Spark Streaming概述
Spark Streaming 初始化过程
Spark Streaming Receiver启动过程分析
Spark Streaming 数据准备阶段分析(Receiver方式)
Spark Streaming 数据计算阶段分析
SparkStreaming Backpressure分析
Spark Streaming Executor DynamicAllocation 机制分析

——————————————————————————————

SparkStreaming的全过程分为两个阶段:数据准备阶段和数据计算阶段。两个阶段在功能上相互独立,仅通过数据联系在一起。"Spark Streaming 数据准备阶段分析"从源码角度分析了Spark Streaming数据准备阶段的具体流程。本文将从源码的角度对数据计算阶段的具体流程进行分析。
Spark Streaming数据计算阶段包含批次数据划分,批作业生成,批wt提交三个部分。

1、 JobGenerator 启动

JobGenerator用于定期生成Job并进行提交 。"Spark Streaming 初始化过程分析"中提到,在启动JobScheduler时,其会调用JobGenerator的start方法,启动JobGenerator.
JobGenerator的start方法实现如下:

  /** Start generation of jobs */
  def start(): Unit = synchronized {
    if (eventLoop != null) return // generator has already been started

    // Call checkpointWriter here to initialize it before eventLoop uses it to avoid a deadlock.
    // See SPARK-10125
    checkpointWriter

    eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {
      override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event)

      override protected def onError(e: Throwable): Unit = {
        jobScheduler.reportError("Error in job generator", e)
      }
    }
    eventLoop.start()

    if (ssc.isCheckpointPresent) {
      restart()
    } else {
      startFirstTime()
    }
  }

通过分析上述代码可知,在JobGenerator.start()被调用时,其将创建

  • eventLoop对象并启动,其中eventLoop定义事件交由processEvent(event).
    processEvent其依据事件的类型,对其进行不同的处理。
  • 调用startFirstTime()方法。通过分析startFirstTime的实现逻辑,可知其进行两项主要工作:
  • 调用 timer.start方法、 定期生成Job
  • 调用graph.start方法
  /** Starts the generator for the first time */
  private def startFirstTime() {
    val startTime = new Time(timer.getStartTime())
    graph.start(startTime - graph.batchDuration)
    timer.start(startTime.milliseconds)
    logInfo("Started JobGenerator at " + startTime)
  }

下面我分别对这二者进行解析。

2、Job 生成及提交

2.1 周期性触发Job生成事件

startFirstTime()方法中调用了timer.start方法,其中timer[RecurringTimer]为定时器,与Spark Streaming 数据准备阶段分析一文中介绍切片时所有定时器一样。其按设置的时间周期,重复的执行计划的任务。此处Timer的具体实现为:

  private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds,
    longTime => eventLoop.post(GenerateJobs(new Time(longTime))), "JobGenerator")

其每个batchDuration规定时间,都会向eventLoop发送一GenerateJobs事件,eventLoop收到GenerateJobs事件,则使用processEvent进行相应处理,此处为调用 generateJobs()方法 ,生成job.

  /** Processes all events */
  private def processEvent(event: JobGeneratorEvent) {
    logDebug("Got event " + event)
    event match {
      case GenerateJobs(time) => generateJobs(time)
      case ClearMetadata(time) => clearMetadata(time)
      case DoCheckpoint(time, clearCheckpointDataLater) =>
        doCheckpoint(time, clearCheckpointDataLater)
      case ClearCheckpointData(time) => clearCheckpointData(time)
    }
  }

2.2 Job生成详细过程

上文搞到Timer分周期性的触发Job生成事件,并通过generateJobs来生成Job.
JobGenerator在每个Batch Interval都会为应用中的每个Output Stream建立一个Job, 该批次中的所有Job组成一个JobSet.使用JobScheduler的submitJobSet进行批量Job提交。
下面来分析generateJobs的实现逻辑。

  /** Generate jobs and perform checkpointing for the given `time`.  */
  private def generateJobs(time: Time) {
    // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
    // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
    ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
    Try {
      jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
      graph.generateJobs(time) // generate jobs using allocated block
    } match {
      case Success(jobs) =>
        val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
        jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
      case Failure(e) =>
        jobScheduler.reportError("Error generating jobs for time " + time, e)
        PythonDStream.stopStreamingContextIfPythonProcessIsDead(e)
    }
    eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
  }

在generateJobs生成Job时, 其首先通过ReceiverTracker 取得其中注册的未分配的数据信息。然后通过DStreamGraph生成Job。

2.2.1 批数据信息划分

本部分会将Spark Streaming 数据准备阶段分析 中生成的未分配的block,划归到某个批次进行处理。具体过程如下:

在生成Job时,首先调用如下语句:

jobScheduler.receiverTracker.allocateBlocksToBatch(time)

该语句用来划分某批次(time)要处理的数据。下面对其获取过程进行详说说明。

其中allocateBlocksToBatch的实现如下:


  /** Allocate all unallocated blocks to the given batch. */
  def allocateBlocksToBatch(batchTime: Time): Unit = {
    if (receiverInputStreams.nonEmpty) {
      receivedBlockTracker.allocateBlocksToBatch(batchTime)
    }
  }

其将调用receivedBlockTracker的allocateBlocksToBatch方法,将未分配数据信息取出,并划分给batchTime所指批次。首先receivedBlockTracker从streamIdToUnallocatedBlockQueues中取出未分配的block信息,将其包装为AllocatedBlocks,并注册在timeToAllocatedBlocks表中,等待某批次(batchTime)生成Job时,与Job进行绑定。

  /**
   * Allocate all unallocated blocks to the given batch.
   * This event will get written to the write ahead log (if enabled).
   */
  def allocateBlocksToBatch(batchTime: Time): Unit = synchronized {
    if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
      val streamIdToBlocks = streamIds.map { streamId =>
          (streamId, getReceivedBlockQueue(streamId).dequeueAll(x => true))
      }.toMap
      val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
      if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
        timeToAllocatedBlocks.put(batchTime, allocatedBlocks)
        lastAllocatedBatchTime = batchTime
      } else {
        logInfo(s"Possibly processed batch $batchTime needs to be processed again in WAL recovery")
      }
    } else {
      // This situation occurs when:
      // 1. WAL is ended with BatchAllocationEvent, but without BatchCleanupEvent,
      // possibly processed batch job or half-processed batch job need to be processed again,
      // so the batchTime will be equal to lastAllocatedBatchTime.
      // 2. Slow checkpointing makes recovered batch time older than WAL recovered
      // lastAllocatedBatchTime.
      // This situation will only occurs in recovery time.
      logInfo(s"Possibly processed batch $batchTime needs to be processed again in WAL recovery")
    }
  }

2.2.2 批作业(Job)生成

通过graph.generateJobs(time)方法分别将DStreamGraph中的每个OutputStream转换了一个Job(如果应用中有多个OutputStream算子,则一个批次会生成多个Job)。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
  }

通过分析源码,其将逐个调用OutputStream的generateJob方法来将每个OutputStream转化为Job. OutputStream不同于其它DStream的地方为其重写了generateJob方法, 以WordCount程序中使用的print算子中的ForEachDStream为例,其 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
    }
  }

通过程序,可以看出,其将调用父DStream中的getOrCompute方法,生成RDD,然后包装成Job。

2.2.2.1 RDD 生成

以WordCount为例,先来看一下WordCount应用中DStream的转换,转换关系如下:


4093076-59088721c2c33870.png
WordCount应用中DStream转换关系

通过分析,getOrCompute( compute方法与之类似)方法由DStream基类创建, 如果子类重写该方法,则执行子类方法; 若未重写,则执行基类中的方法。通过查看上述转换关系链中ForEachDStream的父亲-ShuffledDStream,发现其未重写getOrCompute方法,因此将使用继承自基类DStream中的getOrCompute, 代码如下。

  /**
   * Get the RDD corresponding to the given time; either retrieve it from cache
   * or compute-and-cache it.
   */
  private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {
    // If RDD was already generated, then retrieve it from HashMap,
    // or else compute the RDD
    generatedRDDs.get(time).orElse {
      // Compute the RDD if time is valid (e.g. correct time in a sliding window)
      // of RDD generation, else generate nothing.
      if (isTimeva lid(time)) {

        val rddOption = createRDDWithLocalProperties(time, displayInnerRDDOps = false) {
          // Disable checks for existing output directories in jobs launched by the streaming
          // scheduler, since we may need to write output to an existing directory during checkpoint
          // recovery; see SPARK-4835 for more details. We need to have this call here because
          // compute() might cause Spark jobs to be launched.
          SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
            compute(time)
          }
        }

        rddOption.foreach { case newRDD =>
          // Register the generated RDD for caching and checkpointing
          if (storageLevel != StorageLevel.NONE) {
            newRDD.persist(storageLevel)
            logDebug(s"Persisting RDD ${newRDD.id} for time $time to $storageLevel")
          }
          if (checkpointDuration != null && (time - zeroTime).isMultipleOf(checkpointDuration)) {
            newRDD.checkpoint()
            logInfo(s"Marking RDD ${newRDD.id} for time $time for checkpointing")
          }
          generatedRDDs.put(time, newRDD)
        }
        rddOption
      } else {
        None
      }
    }
  }

此代码将调用ShuffledDStream的compute生成RDD,其compute实现为:

override def compute(validTime: Time): Option[RDD[(K, C)]] = {
    parent.getOrCompute(validTime) match {
      case Some(rdd) => Some(rdd.combineByKey[C](
          createCombiner, mergeva lue, mergeCombiner, partitioner, mapSideCombine))
      case None => None
    }
  }

通过分析,其将调用其父DStream的compute方法,其父DStream继续递归向上调用父DStream的compute直到源头DStream(SocketInputDStream),
SocketInputDStream的compute方法继承自ReceiverInputDStream,其compute方法将生成源头RDD,并按DStream递归逆向生成RDD Graph.

ReceiverInputDStream定义的compute的实现如下:

 /**
   * Generates RDDs with blocks received by the receiver of this stream. */
  override def compute(validTime: Time): Option[RDD[T]] = {
    val blockRDD = {

      if (validTime < graph.startTime) {
        // If this is called for any time before the start time of the context,
        // then this returns an empty RDD. This may happen when recovering from a
        // driver failure without any write ahead log to recover pre-failure data.
        new BlockRDD[T](ssc.sc, Array.empty)
      } else {
        // Otherwise, ask the tracker for all the blocks that have been allocated to this stream
        // for this batch
        val receiverTracker = ssc.scheduler.receiverTracker
        val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

        // Register the input blocks information into InputInfoTracker
        val inputInfo = StreamInputInfo(id, blockInfos.flatMap(_.numRecords).sum)
        ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)

        // Create the BlockRDD
        createBlockRDD(validTime, blockInfos)
      }
    }
    Some(blockRDD)
  }

此处,通过如下逻辑

val receiverTracker = ssc.scheduler.receiverTracker 
val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

将 2.2.1 节中提到的划分过批次的数据信息(blockInfos)取出,包装成StreamInputInfo,然后通过createBlockRDD方法生成RDD. 此处,如果blockInfos信息不空,则生成正常的RDD;若blockInfos为空,则没有Block的空RDD(new BlockRDD(ssc.sc, Array.empty))。

2.2.3 Job 的提交

当成功转化为Job之后,然后通过JobScheduler对JobSet进行提交。

case Success(jobs) =>
        val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
        jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))

其中submitJobSet方法实现如下:

  def submitJobSet(jobSet: JobSet) {
    if (jobSet.jobs.isEmpty) {
      logInfo("No jobs added for time " + jobSet.time)
    } else {
      listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
      jobSets.put(jobSet.time, jobSet)
      jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
      logInfo("Added jobs for time " + jobSet.time)
    }
  }
  • 首先将JobSet加入JobSets表中,以便监控系统可以追踪。
  • 将Job通过JobHandler进行包装,然后由ThreadPoolExecutor的execute增加到其workQueue中,等待被调度执行。如果线程池有空闲线程,则其将被调度。(此部分为Java并发编程中Executor的相关内容。)
    其中线程池的定义如下所示:
  private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1)
  private val jobExecutor =
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor")

通过分析代码可知,JobScheduler创建一固定长度的daemon线程池jobExecutor ,大小由“spark.streaming.concurrentJobs”,默认为1。 线程池中有多个线程则可以同时执行多少个Job, 默认情况下每次只能提交一个Job。当Job来不及执行时,会产生堆集,堆集的Job会保存在ThreadPoolExecutor中的workQueue队列中,等待有空闲线程时被调度。

  • JobHandler是ThreadPoolExecutor中Executor运行的主要任务,其功能是对提交的Job进行处理,实现如下, 其将通过EventLoop对Job状态进行管理,并通过调用job.run方法,使用Job开始运行。
    def run() {
      val oldProps = ssc.sparkContext.getLocalProperties
      try {
        ssc.sparkContext.setLocalProperties(SerializationUtils.clone(ssc.savedProperties.get()))
        val formattedTime = UIUtils.formatBatchTime(
          job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
        val batchUrl = s"/streaming/batch/id=${job.time.milliseconds}"
        val batchLinkText = s"[output operation ${job.outputOpId}, batch time ${formattedTime}]"

        ssc.sc.setJobDescription(
          s"""Streaming job from <a href="$batchUrl">$batchLinkText</a>""")
        ssc.sc.setLocalProperty(BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
        ssc.sc.setLocalProperty(OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
        // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
        // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
        ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")

        // We need to assign `eventLoop` to a temp variable. Otherwise, because
        // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then
        // it's possible that when `post` is called, `eventLoop` happens to null.
        var _eventLoop = eventLoop
        if (_eventLoop != null) {
          _eventLoop.post(JobStarted(job, clock.getTimeMillis()))
          // Disable checks for existing output directories in jobs launched by the streaming
          // scheduler, since we may need to write output to an existing directory during checkpoint
          // recovery; see SPARK-4835 for more details.
          SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
            job.run()
          }
          _eventLoop = eventLoop
          if (_eventLoop != null) {
            _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
          }
        } else {
          // JobScheduler has been stopped.
        }
      } finally {
        ssc.sparkContext.setLocalProperties(oldProps)
      }
    }

其中Job.run方法,实现如下:

 def run() {
    _result = Try(func())
  }

其将执行创建Job时的方法func。WordCount应用是ForEachDStream中进行Job创建。其创建方法 上文已经提到:

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
    }
  }

分析代码可知,其将调用foreachFunc, 该方法是创建ForEachDStream时引入的参数,由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(s"Time: $time")
        println("-------------------------------------------")
        firstNum.take(num).foreach(println)
        if (firstNum.length > num) println("...")
        println()
        // scalastyle:on println
      }
    }
    foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
  }

其中,会调用rdd.take()算子, take算子属于action算子,会触发SparkJob的提交,接下来的处理流程与spark 批处理相同。
前述生成的Job,只是Streaming中定义的抽象,与SparkJob(真正进行调度,生成Task)不同。


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Spark Streaming 数据准备阶段分.. 下一篇Spark Streaming Receiver启动过..

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }