zoukankan      html  css  js  c++  java

  • 16.Spark Streaming源码解读之数据清理机制解析

    原创文章，转载请注明:转载自 听风居士博客(http://www.cnblogs.com/zhouyf/)

    本期内容：

    一、Spark Streaming 数据清理总览

    二、Spark Streaming 数据清理过程详解

    三、Spark Streaming 数据清理的触发机制


        Spark Streaming不像普通Spark 的应用程序，普通Spark程序运行完成后,中间数据会随着SparkContext的关闭而被销毁,而Spark Streaming一直在运行，不断计算，每一秒中在不断运行都会产生大量的中间数据，所以需要对对象及元数据需要定期清理。每个batch duration运行时不断触发job后需要清理rdd和元数据。下面我们就结合源码详细解析一下Spark Streaming程序的数据清理机制。


    一、数据清理总览

        Spark Streaming 运行过程中,随着时间不断产生Job,当job运行结束后,需要清理相应的数据(RDD,元数据信息,Checkpoint数据),Job由JobGenerator定时产生,数据的清理也是有JobGenerator负责。

        JobGenerator负责数据清理控制的代码位于一个消息循环体eventLoop中：

    1.     eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {
    2.       override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event)
    4.       override protected def onError(e: Throwable): Unit = {
    5.         jobScheduler.reportError("Error in job generator", e)
    6.       }
    7.     }
    8.     eventLoop.start()

    其中的核心逻辑位于processEvent(event)函数中:

    1.   /** Processes all events */
    2.   private def processEvent(event: JobGeneratorEvent) {
    3.     logDebug("Got event " + event)
    4.     event match {
    5.       case GenerateJobs(time) => generateJobs(time)
    6.       case ClearMetadata(time) => clearMetadata(time)
    7.       case DoCheckpoint(time, clearCheckpointDataLater) =>
    8.         doCheckpoint(time, clearCheckpointDataLater)
    9.       case ClearCheckpointData(time) => clearCheckpointData(time)
    10.     }
    11.   }

    可以看到当JobGenerator收到ClearMetadata(time) 和 ClearCheckpointData(time)是会进行相应的数据清理,其中 clearMetadata(time)会清理RDD数据和一些元数据信息, ClearCheckpointData(time)会清理Checkpoint数据。

    二、数据清理过程详解
        2.1 ClearMetaData 过程详解
    首先看一下clearMetaData函数的处理逻辑：
    1.   /** Clear DStream metadata for the given `time`. */
    2.   private def clearMetadata(time: Time) {
    3.     ssc.graph.clearMetadata(time)
    5.     // If checkpointing is enabled, then checkpoint,
    6.     // else mark batch to be fully processed
    7.     if (shouldCheckpoint) {
    8.       eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = true))
    9.     } else {
    10.       // If checkpointing is not enabled, then delete metadata information about
    11.       // received blocks (block data not saved in any case). Otherwise, wait for
    12.       // checkpointing of this batch to complete.
    13.       val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
    14.       jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
    15.       jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
    16.       markBatchFullyProcessed(time)
    17.     }
    18.   }

    首先调用了DStreamGraph的clearMetadata方法:

    1.   def clearMetadata(time: Time) {
    2.     logDebug("Clearing metadata for time " + time)
    3.     this.synchronized {
    4.       outputStreams.foreach(_.clearMetadata(time))
    5.     }
    6.     logDebug("Cleared old metadata for time " + time)
    7.   }

    这里调用了所有OutputDStream (关于DStream 的分类请参考http://blog.csdn.net/zhouzx2010/article/details/51460790)的clearMetadata方法

    1.   private[streaming] def clearMetadata(time: Time) {
    2.     val unpersistData = ssc.conf.getBoolean("spark.streaming.unpersist", true)
    3.      //获取需要清理的RDD
    4.     val oldRDDs = generatedRDDs.filter(_._1 <= (time - rememberDuration))
    5.     logDebug("Clearing references to old RDDs: [" +
    6.       oldRDDs.map(x => s"${x._1} -> ${x._2.id}").mkString(", ") + "]")

    8.     //将要清除的RDD从generatedRDDs 中清除 
    9.     generatedRDDs --= oldRDDs.keys
    10.     if (unpersistData) {
    11.       logDebug(s"Unpersisting old RDDs: ${oldRDDs.values.map(_.id).mkString(", ")}")
    12.       oldRDDs.values.foreach { rdd =>
    13.       //将RDD 从persistence列表中移除
    14.         rdd.unpersist(false)
    15.         // Explicitly remove blocks of BlockRDD
    16.         rdd match {
    17.           case b: BlockRDD[_] =>
    18.             logInfo(s"Removing blocks of RDD $b of time $time")
    19.             //移除RDD的block 数据
    20.             b.removeBlocks()
    21.           case _ =>
    22.         }
    23.       }
    24.     }
    25.     logDebug(s"Cleared ${oldRDDs.size} RDDs that were older than " +
    26.       s"${time - rememberDuration}: ${oldRDDs.keys.mkString(", ")}")
    27.     //清除依赖的DStream
    28.     dependencies.foreach(_.clearMetadata(time))
    29.   }

    关键的清理逻辑在代码中做了详细注释,首先清理DStream对应的RDD的元数据信息,然后清理RDD的数据,最后对DStream所依赖的DStream进行清理。

    回到JobGenerator的clearMetadata函数：

    1.     if (shouldCheckpoint) {
    2.       eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = true))
    3.     } else {
    4.       // If checkpointing is not enabled, then delete metadata information about
    5.       // received blocks (block data not saved in any case). Otherwise, wait for
    6.       // checkpointing of this batch to complete.
    7.       val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
    8.       jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
    9.       jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
    10.       markBatchFullyProcessed(time)
    11.     }

    调用了ReceiverTracker的 cleanupOldBlocksAndBatches方法，最后调用了clearupOldBatches方法：

    1.   def cleanupOldBatches(cleanupThreshTime: Time, waitForCompletion: Boolean): Unit = synchronized {
    2.     require(cleanupThreshTime.milliseconds < clock.getTimeMillis())
    3.     val timesToCleanup = timeToAllocatedBlocks.keys.filter { _ < cleanupThreshTime }.toSeq
    4.     logInfo(s"Deleting batches: ${timesToCleanup.mkString(" ")}")
    5.     if (writeToLog(BatchCleanupEvent(timesToCleanup))) {
    6.       //将要删除的Batch数据清除
    7.       timeToAllocatedBlocks --= timesToCleanup
    8.       //清理WAL日志
    9.       writeAheadLogOption.foreach(_.clean(cleanupThreshTime.milliseconds, waitForCompletion))
    10.     } else {
    11.       logWarning("Failed to acknowledge batch clean up in the Write Ahead Log.")
    12.     }
    13.   }

    可以看到ReceiverTracker的clearupOldBatches方法清理了Receiver数据,也就是Batch数据和WAL日志数据。
    最后对InputInfoTracker信息进行清理：
    1.   def cleanup(batchThreshTime: Time): Unit = synchronized {
    2.     val timesToCleanup = batchTimeToInputInfos.keys.filter(_ < batchThreshTime)
    3.     logInfo(s"remove old batch metadata: ${timesToCleanup.mkString(" ")}")
    4.     batchTimeToInputInfos --= timesToCleanup
    5.   }
    这简单的清除了batchTimeToInputInfos 的输入信息。

    2.2 ClearCheckPoint 过程详解
    看一下clearCheckpointData的处理逻辑:
    1.   /** Clear DStream checkpoint data for the given `time`. */
    2.   private def clearCheckpointData(time: Time) {
    3.     ssc.graph.clearCheckpointData(time)
    5.     // All the checkpoint information about which batches have been processed, etc have
    6.     // been saved to checkpoints, so its safe to delete block metadata and data WAL files
    7.     val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
    8.     jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
    9.     jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
    10.     markBatchFullyProcessed(time)
    11.   }

    后面的ReceiverTraker和InputInforTracker的清理逻辑和ClearMetaData的相同,这分析DStreamGraph的clearCheckpointData方法:

    1.   def clearCheckpointData(time: Time) {
    2.     logInfo("Clearing checkpoint data for time " + time)
    3.     this.synchronized {
    4.       outputStreams.foreach(_.clearCheckpointData(time))
    5.     }
    6.     logInfo("Cleared checkpoint data for time " + time)
    7.   }

    同样的调用了DStreamGraph中所有OutputDStream的clearCheckPiontData 方法:

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

    这里的核心逻辑在checkpointData.cleanup(time)方法,这里的CheckpointData 是 DStreamCheckpointData对象, DStreamCheckpointData的clearup方法如下:
    1. def cleanup(time: Time) {
    2.     // 获取需要清理的Checkpoint 文件 时间
    3.     timeToOldestCheckpointFileTime.remove(time) match {
    4.       case Some(lastCheckpointFileTime) =>
    5.         //获取需要删除的文件
    6.         val filesToDelete = timeToCheckpointFile.filter(_._1 < lastCheckpointFileTime)
    7.         logDebug("Files to delete:
" + filesToDelete.mkString(","))
    8.         filesToDelete.foreach {
    9.           case (time, file) =>
    10.             try {
    11.               val path = new Path(file)
    12.               if (fileSystem == null) {
    13.                 fileSystem = path.getFileSystem(dstream.ssc.sparkContext.hadoopConfiguration)
    14.               }
    15.           //  删除文件     
    16.               fileSystem.delete(path, true)
    17.               timeToCheckpointFile -= time
    18.               logInfo("Deleted checkpoint file '" + file + "' for time " + time)
    19.             } catch {
    20.               case e: Exception =>
    21.                 logWarning("Error deleting old checkpoint file '" + file + "' for time " + time, e)
    22.                 fileSystem = null
    23.             }
    24.         }
    25.       case None =>
    26.         logDebug("Nothing to delete")
    27.     }
    28.   }

    可以看到checkpoint的清理,就是删除了指定时间以前的checkpoint文件。

    三、数据清理的触发
        3.1 ClearMetaData 过程的触发
    JobGenerator 生成job后,交给JobHandler执行, JobHandler的run方法中,会在job执行完后给JobScheduler 发送JobCompleted消息:
    1.           _eventLoop = eventLoop
    2.           if (_eventLoop != null) {
    3.             _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
    5.           }
    JobScheduler 收到JobCompleted 消息调用 handleJobCompletion 方法,源码如下:

    1.   private def processEvent(event: JobSchedulerEvent) {
    2.     try {
    3.       event match {
    4.         case JobStarted(job, startTime) => handleJobStart(job, startTime)
    5.         case JobCompleted(job, completedTime) => handleJobCompletion(job, completedTime)
    6.         case ErrorReported(m, e) => handleError(m, e)
    7.       }
    8.     } catch {
    9.       case e: Throwable =>
    10.         reportError("Error in job scheduler", e)
    11.     }
    12.   }

    在 JobScheduler 的handleJobCompletion方法中会调用JobGenerator的onBatchCompletion方法,我们看一下JobGenerator的 onBatchCompletion 方法的源码:

    1.   def onBatchCompletion(time: Time) {
    2.     eventLoop.post(ClearMetadata(time))
    3.   }

    可以看到JobGenerator的onBatchCompletion方法给自己发送了ClearMetadata消息从而触发了ClearMetaData操作。

    3.2 ClearCheckPoint 过程的触发
        清理CheckPoint数据发生在CheckPoint完成之后,我们先看一下CheckPointHandler的run方法:

    1.           // All done, print success
    2.           val finishTime = System.currentTimeMillis()
    3.           logInfo("Checkpoint for time " + checkpointTime + " saved to file '" + checkpointFile +
    4.             "', took " + bytes.length + " bytes and " + (finishTime - startTime) + " ms")
    5.           //调用JobGenerator的方法进行checkpoint数据清理
    6.           jobGenerator.onCheckpointCompletion(checkpointTime, clearCheckpointDataLater)
    7.           return

    可以看到在checkpoint完成后,会调用JobGenerator的onCheckpointCompletion方法进行checkpoint数据清理,我查看JobGenerator的onCheckpointCompletion方法源码:

    1.   def onCheckpointCompletion(time: Time, clearCheckpointDataLater: Boolean) {
    2.     if (clearCheckpointDataLater) {
    3.       eventLoop.post(ClearCheckpointData(time))
    4.     }
    5.   }

    可以看到JobGenerator的onCheckpointCompletion方法中首先对传进来的 clearCheckpointDataLater 参数进行判断,如果该参数为true,就会给JobGenerator的eventLoop循环体发送ClearCheckpointData消息,从而触发clearCheckpointData 方法的调用,进行Checkpoint数据的清理。
    什么时候该参数会true呢？
    我们回到JobGenerator的 ClearMetadata 方法：

    1.   private def clearMetadata(time: Time) {
    2.     ssc.graph.clearMetadata(time)

    5.     if (shouldCheckpoint) {
    6.       //发送DoCheckpoint消息,并进行相应的Checkpoint数据清理
    7.       eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = true))
    8.     } else {

    10.       val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
    11.       jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
    12.       jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
    13.       markBatchFullyProcessed(time)
    14.     }
    15.   }

    可以看到在clearMetadata方法中,发送了DoCheckpoint消息,其中参数 clearCheckpointDataLater 为ture。Generator的eventLoop收到该消息后调用 doCheckpoint 方法：

    1.   private def doCheckpoint(time: Time, clearCheckpointDataLater: Boolean) {
    2.     if (shouldCheckpoint && (time - graph.zeroTime).isMultipleOf(ssc.checkpointDuration)) {
    3.       logInfo("Checkpointing graph for time " + time)
    4.       ssc.graph.updateCheckpointData(time)
    5.       checkpointWriter.write(new Checkpoint(ssc, time), clearCheckpointDataLater)
    6.     }
    7.   }

    这里关键一步：调用了CheckpointWriter的write方法,注意此时参数 clearCheckpointDataLater 为true。我们进入该方法：

    1.   def write(checkpoint: Checkpoint, clearCheckpointDataLater: Boolean) {
    2.     try {
    3.       val bytes = Checkpoint.serialize(checkpoint, conf)
    4.      //将参数clearCheckpointDataLater传入CheckpoitWriteHandler
    5.       executor.execute(new CheckpointWriteHandler(
    6.         checkpoint.checkpointTime, bytes, clearCheckpointDataLater))
    7.       logInfo("Submitted checkpoint of time " + checkpoint.checkpointTime + " writer queue")
    8.     } catch {
    9.       case rej: RejectedExecutionException =>
    10.         logError("Could not submit checkpoint task to the thread pool executor", rej)
    11.     }
    12.   }

    可以看到此时参数 clearCheckpointDataLater 传入CheckpointWriteHandler 。这样Checkpoint完成之后就会发送ClearCheckpointData消息给JobGenerator进行Checkpoint数据的清理。

    原创文章，转载请注明:转载自 听风居士博客(http://www.cnblogs.com/zhouyf/)










     
  • 相关阅读:
    如何在宿主机上查看kvm虚拟机的IP
    virt-viewer 连kvm　虚机
    用virsh console vhosts 卡住
    sshpass 实现shell脚本直接加载密登录服务器
    maven创建helloword项目
    linux 安装maven
    xz -d Python-3.4.7.tar.xz
    查看哪个用户登录过服务器　记录　时间　和　ip
    两台linux服务器各有两个不同的用户 其中一个服务器可以无密码登录服务器
    maven私服nexus
  • 原文地址：https://www.cnblogs.com/zhouyf/p/5633026.html
Copyright © 2011-2022 走看看