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

本期內(nèi)容：

• Spark Streaming數(shù)據(jù)清理原因和現(xiàn)象

• Spark Streaming數(shù)據(jù)清理代碼解析

對Spark Streaming解析了這么多課之后，我們越來越能感知，Spark Streaming只是基于Spark Core的一個應(yīng)用程序，因此掌握Spark Streaming對于我們怎么編寫Spark應(yīng)用是絕對有好處的。

Spark Streaming 不像Spark Core的應(yīng)用程序，Spark Core的應(yīng)用的數(shù)據(jù)是存儲在底層文件系統(tǒng)，如HDFS等別的存儲系統(tǒng)中，而Spark Streaming一直在運(yùn)行，不斷計算，每一秒中在不斷運(yùn)行都會產(chǎn)生大量的累加器、廣播變量，所以需要對對象及元數(shù)據(jù)需要定期清理。每個batch duration運(yùn)行時不斷觸發(fā)job后需要清理rdd和元數(shù)據(jù)。Client模式可以看到打印的日志，從文件日志也可以看到清理日志內(nèi)容。

Spark運(yùn)行在jvm上，jvm會產(chǎn)生對象，jvm需要對對象進(jìn)行回收工作，如果我們不管理gc（對象產(chǎn)生和回收），jvm很快耗盡。現(xiàn)在研究的是Spark Streaming的Spark GC。Spark Streaming對rdd的數(shù)據(jù)管理、元數(shù)據(jù)管理相當(dāng)jvm對gc管理。數(shù)據(jù)、元數(shù)據(jù)是操作DStream時產(chǎn)生的，數(shù)據(jù)、元數(shù)據(jù)的回收則需要研究DStream的產(chǎn)生和回收。

數(shù)據(jù)輸入靠InputDStream，數(shù)據(jù)輸入、數(shù)據(jù)操作、數(shù)據(jù)輸出，整個生命周期都是基于DStream構(gòu)建的，DStream負(fù)責(zé)rdd的生命周期，rrd是DStream產(chǎn)生的，對rdd的操作也是對DStream的操作，所以不斷產(chǎn)生batchDuration的循環(huán)，所以研究對rdd的操作也就是研究對DStream的操作。 以從kafka中 Direct方式為例, DirectKafkaInputDStream會產(chǎn)生KafkaRDD

override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {  val untilOffsets = clamp(latestLeaderOffsets(maxRetries))  val rdd = KafkaRDD[K, V, U, T, R](
    context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)  // Report the record number and metadata of this batch interval to InputInfoTracker.
  val offsetRanges = currentOffsets.map { case (tp, fo) =>
    val uo = untilOffsets(tp)    OffsetRange(tp.topic, tp.partition, fo, uo.offset)
  }  val description = offsetRanges.filter { offsetRange =>
    // Don't display empty ranges.
    offsetRange.fromOffset != offsetRange.untilOffset
  }.map { offsetRange =>
    s"topic: ${offsetRange.topic}\tpartition: ${offsetRange.partition}\t" +
      s"offsets: ${offsetRange.fromOffset} to ${offsetRange.untilOffset}"
  }.mkString("\n")  // Copy offsetRanges to immutable.List to prevent from being modified by the user
  val metadata = Map(    "offsets" -> offsetRanges.toList,    StreamInputInfo.METADATA_KEY_DESCRIPTION -> description)  val inputInfo = StreamInputInfo(id, rdd.count, metadata)
  ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
  currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)  Some(rdd)
}

DStream隨著時間進(jìn)行,數(shù)據(jù)周期性產(chǎn)生和周期性釋放，在JobGenerator中有一個定時器：

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

而JobGenerator中也有一個EventLoop來周期性的接收消息事件：

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

里面就有清理元數(shù)據(jù)和清理checkpoint數(shù)據(jù)的方法 clearMetadata：清楚元數(shù)據(jù)信息。

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

DStreamGraph:首先會清理outputDStream，其實(shí)就是ForEachDStream

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

DStream.clearMetadata:除了清除RDD，也可以清除metadata元數(shù)據(jù)。如果想RDD跨Batch Duration的話可以設(shè)置rememberDuration時間. rememberDuration

/** * 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的數(shù)據(jù)
        case _ =>
      }
    }
  }
  logDebug("Cleared " + oldRDDs.size + " RDDs that were older than " +
    (time - rememberDuration) + ": " + oldRDDs.keys.mkString(", "))//依賴的DStream也需要清理掉。
  dependencies.foreach(_.clearMetadata(time))
}

在BlockRDD中,BlockManagerMaster根據(jù)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看看 clearCheckpoint：清除緩存數(shù)據(jù)

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

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

ClearCheckpointData: 和清除元數(shù)據(jù)信息一樣，還是清除DStream依賴的緩存數(shù)據(jù)。

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

DStreamCheckpointData:清除緩存的數(shù)據(jù)

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

至此，我們知道了怎么清理舊的數(shù)據(jù)以及清理什么數(shù)據(jù)，但是清理數(shù)據(jù)什么時候被觸發(fā)的？在最終提交Job的時候，是交給JobHandler去執(zhí)行的。

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 $batchLinkText""")
        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-4835 for more details.
          PairRDDFunctions.disableOutputSpecValidation.withValue(true) {
            job.run()
          }
          _eventLoop = eventLoop          if (_eventLoop != null) {//當(dāng)Job完成的時候，eventLoop會發(fā)消息初始化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)
      }
    }
  }
}

EventLoop 的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()

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

調(diào)用JobGenerator的onBatchCompletion方法清除元數(shù)據(jù)。

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

至此我們明白了什么時候觸發(fā)清楚舊數(shù)據(jù)的過程。

備注：

1、DT大數(shù)據(jù)夢工廠微信公眾號DT_Spark 
2、IMF晚8點(diǎn)大數(shù)據(jù)實(shí)戰(zhàn)YY直播頻道號：68917580
3、新浪微博: http://www.weibo.com/ilovepains