您當前位置：首頁 > 服務器 > 第16課：Spark Streaming源碼解讀之數據清理內幕徹底解密

第16課：Spark Streaming源碼解讀之數據清理內幕徹底解密

來源：程序員人生發布時間：2016-06-15 08:24:56 閱讀次數：3271次

本篇博客的主要目的是：
1. 理清楚Spark Streaming中數據清算的流程

組織思路以下：
a) 背景
b) 如何研究Spark Streaming數據清算？
c) 源碼解析

1：背景
Spark Streaming數據清算的工作不管是在實際開發中，還是自己動手實踐中都是會面臨的，Spark Streaming中Batch Durations中會不斷的產生RDD，這樣會不斷的有內存對象生成，其中包括元數據和數據本身。由此Spark Streaming本身會有1套產生元數據和數據的清算機制。

2：如何研究Spark Streaming數據清算？

操作DStream的時候會產生元數據，所以要解決RDD的數據清算工作就1定要從DStream入手。由于DStream是RDD的模板，DStream之間有依賴關系。
DStream的操作產生了RDD,接收數據也靠DStream，數據的輸入，數據的計算，輸出全部生命周期都是由DStream構建的。由此，DStream負責RDD的全部生命周期。因此研究的入口的是DStream。
基于Kafka數據來源，通過Direct的方式訪問Kafka,DStream隨著時間的進行，會不斷的在自己的內存數據結構中保護1個HashMap,HashMap保護的就是時間窗口，和時間窗口下的RDD.依照Batch Duration來存儲RDD和刪除RDD.
Spark Streaming本身是1直在運行的，在自己計算的時候會不斷的產生RDD，例如每秒Batch Duration都會產生RDD,除此以外可能還有累加器，廣播變量。由于不斷的產生這些對象，因此Spark Streaming有自己的1套對象，元數據和數據的清算機制。
Spark Streaming對RDD的管理就相當于JVM的GC.

3：源碼解析
generatedRDDs:安照Batch Duration的方式來存儲RDD和刪除RDD。

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

我們在實際開發中，可能手動緩存，即便不緩存的話，它在內存generatorRDD中也有對象，如何釋放他們？不單單是RDD本身，也包括數據源(數據來源)和元數據(metada),因此釋放RDD的時候這3方面都需要斟酌。
釋放跟時鐘Click有關系，由于數據是周期性產生，所以肯定是周期性釋放。
因此下1步就需要找JobGenerator

RecurringTimer: 消息循環器將消息不斷的發送給EventLoop

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

2.  eventLoop：onReceive接收到消息。

/** 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)
    }
  }

3.  processEvent:中就會接收到ClearMetadata和ClearCheckpointData。

/** 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)
  }
}

4.  clearMetadata：清楚元數據信息。

/** Clear DStream metadata for the given `time`. */
private def clearMetadata(time: Time) {
  ssc.graph.clearMetadata(time)

  // If checkpointing is enabled, then checkpoint,
  // else mark batch to be fully processed
  if (shouldCheckpoint) {
    eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = true))
  } else {
    // If checkpointing is not enabled, then delete metadata information about
    // received blocks (block data not saved in any case). Otherwise, wait for
    // checkpointing of this batch to complete.
    val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
    jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
    jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
    markBatchFullyProcessed(time)
  }
}

5.  DStreamGraph:首先會清算outputDStream，其實就是forEachDStream

def clearMetadata(time: Time) {
  logDebug("Clearing metadata for time " + time)
  this.synchronized {
    outputStreams.foreach(_.clearMetadata(time))
  }
  logDebug("Cleared old metadata for time " + time)
}

6.  DStream.clearMetadata:除清除RDD，也能夠清除metadata元數據。如果想RDD跨Batch Duration的話可以設置rememberDuration時間. rememberDuration1般都是Batch Duration的倍數。

/**
 * Clear metadata that are older than `rememberDuration` of this DStream.
 * This is an internal method that should not be called directly. This default
 * implementation clears the old generated RDDs. Subclasses of DStream may override
 * this to clear their own metadata along with the generated RDDs.
 */
private[streaming] def clearMetadata(time: Time) {
  val unpersistData = ssc.conf.getBoolean("spark.streaming.unpersist", true)
// rememberDuration記憶周期 查看下RDD是不是是oldRDD
  val oldRDDs = generatedRDDs.filter(_._1 <= (time - rememberDuration))
  logDebug("Clearing references to old RDDs: [" +
    oldRDDs.map(x => s"${x._1} -> ${x._2.id}").mkString(", ") + "]")
//從generatedRDDs中將key清算掉。
  generatedRDDs --= oldRDDs.keys
  if (unpersistData) {
    logDebug("Unpersisting old RDDs: " + oldRDDs.values.map(_.id).mkString(", "))
    oldRDDs.values.foreach { rdd =>
      rdd.unpersist(false)
      // Explicitly remove blocks of BlockRDD
      rdd match {
        case b: BlockRDD[_] =>
          logInfo("Removing blocks of RDD " + b + " of time " + time)
          b.removeBlocks() //清算掉RDD的數據
        case _ =>
      }
    }
  }
  logDebug("Cleared " + oldRDDs.size + " RDDs that were older than " +
    (time - rememberDuration) + ": " + oldRDDs.keys.mkString(", "))
//依賴的DStream也需要清算掉。
  dependencies.foreach(_.clearMetadata(time))
}

7.  在BlockRDD中,BlockManagerMaster根據blockId將Block刪除。刪除Block的操作是不可逆的。

/**
 * Remove the data blocks that this BlockRDD is made from. NOTE: This is an
 * irreversible operation, as the data in the blocks cannot be recovered back
 * once removed. Use it with caution.
 */
private[spark] def removeBlocks() {
  blockIds.foreach { blockId =>
    sparkContext.env.blockManager.master.removeBlock(blockId)
  }
  _isValid = false
}

回到上面JobGenerator中的processEvent
1. clearCheckpoint：清除緩存數據。

/** Clear DStream checkpoint data for the given `time`. */
private def clearCheckpointData(time: Time) {
  ssc.graph.clearCheckpointData(time)

  // All the checkpoint information about which batches have been processed, etc have
  // been saved to checkpoints, so its safe to delete block metadata and data WAL files
  val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
  jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
  jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
  markBatchFullyProcessed(time)
}

2.  clearCheckpointData:

def clearCheckpointData(time: Time) {
  logInfo("Clearing checkpoint data for time " + time)
  this.synchronized {
    outputStreams.foreach(_.clearCheckpointData(time))
  }
  logInfo("Cleared checkpoint data for time " + time)
}

3.  ClearCheckpointData: 和清除元數據信息1樣，還是清除DStream依賴的緩存數據。

private[streaming] def clearCheckpointData(time: Time) {
  logDebug("Clearing checkpoint data")
  checkpointData.cleanup(time)
  dependencies.foreach(_.clearCheckpointData(time))
  logDebug("Cleared checkpoint data")
}

4.  DStreamCheckpointData:清除緩存的數據

/**
 * Cleanup old checkpoint data. This gets called after a checkpoint of `time` has been
 * written to the checkpoint directory.
 */
def cleanup(time: Time) {
  // Get the time of the oldest checkpointed RDD that was written as part of the
  // checkpoint of `time`
  timeToOldestCheckpointFileTime.remove(time) match {
    case Some(lastCheckpointFileTime) =>
      // Find all the checkpointed RDDs (i.e. files) that are older than `lastCheckpointFileTime`
      // This is because checkpointed RDDs older than this are not going to be needed
      // even after master fails, as the checkpoint data of `time` does not refer to those files
      val filesToDelete = timeToCheckpointFile.filter(_._1 < lastCheckpointFileTime)
      logDebug("Files to delete:\n" + filesToDelete.mkString(","))
      filesToDelete.foreach {
        case (time, file) =>
          try {
            val path = new Path(file)
            if (fileSystem == null) {
              fileSystem = path.getFileSystem(dstream.ssc.sparkContext.hadoopConfiguration)
            }
            fileSystem.delete(path, true)
            timeToCheckpointFile -= time
            logInfo("Deleted checkpoint file '" + file + "' for time " + time)
          } catch {
            case e: Exception =>
              logWarning("Error deleting old checkpoint file '" + file + "' for time " + time, e)
              fileSystem = null
          }
      }
    case None =>
      logDebug("Nothing to delete")
  }
}

至此我們也知道了清算的進程，全流程以下：
這里寫圖片描述

但是清算是甚么時候被觸發的？
1. 在終究提交Job的時候，是交給JobHandler去履行的。

private class JobHandler(job: Job) extends Runnable with Logging {
    import JobScheduler._

    def run() {
      try {
        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)

        // 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⑷835 for more details.
          PairRDDFunctions.disableOutputSpecValidation.withValue(true) {
            job.run()
          }
          _eventLoop = eventLoop
          if (_eventLoop != null) {
//當Job完成的時候，eventLoop會發消息初始化onReceive
            _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
          }
        } else {
          // JobScheduler has been stopped.
        }
      } finally {
        ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null)
        ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null)
      }
    }
  }
}

2.  OnReceive初始化接收到消息JobCompleted.

def start(): Unit = synchronized {
  if (eventLoop != null) return // scheduler has already been started

  logDebug("Starting JobScheduler")
  eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
    override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

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

3.  processEvent:

private def processEvent(event: JobSchedulerEvent) {
  try {
    event match {
      case JobStarted(job, startTime) => handleJobStart(job, startTime)
      case JobCompleted(job, completedTime) => handleJobCompletion(job, completedTime)
      case ErrorReported(m, e) => handleError(m, e)
    }
  } catch {
    case e: Throwable =>
      reportError("Error in job scheduler", e)
  }
}

4.  調用JobGenerator的onBatchCompletion方法清楚元數據。

private def handleJobCompletion(job: Job, completedTime: Long) {
  val jobSet = jobSets.get(job.time)
  jobSet.handleJobCompletion(job)
  job.setEndTime(completedTime)
  listenerBus.post(StreamingListenerOutputOperationCompleted(job.toOutputOperationInfo))
  logInfo("Finished job " + job.id + " from job set of time " + jobSet.time)
  if (jobSet.hasCompleted) {
    jobSets.remove(jobSet.time)
    jobGenerator.onBatchCompletion(jobSet.time)
    logInfo("Total delay: %.3f s for time %s (execution: %.3f s)".format(
      jobSet.totalDelay / 1000.0, jobSet.time.toString,
      jobSet.processingDelay / 1000.0
    ))
    listenerBus.post(StreamingListenerBatchCompleted(jobSet.toBatchInfo))
  }
  job.result match {
    case Failure(e) =>
      reportError("Error running job " + job, e)
    case _ =>
  }
}