ContextCleaner是SparkContext中的组件之一。ContextCleaner用于清理那些超出应用范围的RDD、Shuffle对应的map任务状态、Shuffle元数据、Broadcast对象以及RDD的Checkpoint数据。
创建ContextCleaner
创建ContextCleaner的代码如下。
_cleaner = if (_conf.getBoolean("spark.cleaner.referenceTracking", true)) {
Some(new ContextCleaner(this)) } else { None } _cleaner.foreach(_.start())
根据上述代码,我们知道可以通过配置属性spark.cleaner.referenceTracking(默认是true)来决定是否启用
ContextCleaner。
ContextCleaner的组成如下:
* referenceQueue:缓存顶级的AnyRef引用;
* referenceBuffer:缓存AnyRef的虚引用;
* listeners:缓存清理工作的监听器数组;
* cleaningThread:用于具体清理工作的线程。此线程为守护线程,名称为Spark Context Cleaner。
* periodicGCService:类型为ScheduledExecutorService,用于执行GC(garbage
collection,即垃圾收集)的调度线程池,此线程池只包含一个线程,启动的线程名称以context-cleaner-periodic-gc开头。
*
periodicGCInterval:执行GC的时间间隔。可通过spark.cleaner.periodicGC.interval属性进行配置,默认是30分钟。
*
blockOnCleanupTasks:清理非Shuffle的其它数据是否是阻塞式的。可通过spark.cleaner.referenceTracking.blocking属性进行配置,默认是true。
*
blockOnShuffleCleanupTasks:清理Shuffle数据是否是阻塞式的。可通过spark.cleaner.referenceTracking.blocking.shuffle属性进行配置,默认是false。清理Shuffle数据包括:清理MapOutputTracker中指定ShuffleId对应的map任务状态和ShuffleManager中注册的ShuffleId对应的Shuffle元数据。
* stopped:ContextCleaner是否停止的状态标记。
启动ContextCleaner
SparkContext在初始化的过程中会启动ContextCleaner,只有这样ContextCleaner才能够清理那些超出应用范围的RDD、Shuffle对应的map任务状态、Shuffle元数据、Broadcast对象以及RDD的Checkpoint数据。启动ContextCleaner的代码如下:
def start(): Unit = { cleaningThread.setDaemon(true)
cleaningThread.setName("Spark Context Cleaner") cleaningThread.start()
periodicGCService.scheduleAtFixedRate(new Runnable { override def run(): Unit =
System.gc() }, periodicGCInterval, periodicGCInterval, TimeUnit.SECONDS) }
根据上述代码,启动ContextCleaner的步骤如下。
* 将cleaningThread设置为守护线程,并指定名称为Spark Context Cleaner。
* 启动cleaningThread。
* 给periodicGCService设置以periodicGCInterval作为时间间隔定时进行GC操作的任务。
除了GC的定时器,ContextCleaner其余部分的工作原理和listenerBus一样(也采用监听器模式,由异步线程来处理),因此不再赘述。异步线程实际只是调用keepCleaning方法,其实现见代码清单1。
代码清单1 keepCleaning的实现
private def keepCleaning(): Unit = Utils.tryOrStopSparkContext(sc) { while
(!stopped) { try { val reference =
Option(referenceQueue.remove(ContextCleaner.REF_QUEUE_POLL_TIMEOUT))
.map(_.asInstanceOf[CleanupTaskWeakReference]) synchronized { reference.foreach
{ ref => logDebug("Got cleaning task " + ref.task) referenceBuffer.remove(ref)
ref.task match { case CleanRDD(rddId) => doCleanupRDD(rddId, blocking =
blockOnCleanupTasks) case CleanShuffle(shuffleId) =>
doCleanupShuffle(shuffleId, blocking = blockOnShuffleCleanupTasks) case
CleanBroadcast(broadcastId) => doCleanupBroadcast(broadcastId, blocking =
blockOnCleanupTasks) case CleanAccum(accId) => doCleanupAccum(accId, blocking =
blockOnCleanupTasks) case CleanCheckpoint(rddId) => doCleanCheckpoint(rddId) }
} } } catch { case ie: InterruptedException if stopped => // ignore case e:
Exception => logError("Error in cleaning thread", e) } } }
从代码清单1可以看出,异步线程将匹配各种引用,并执行相应的方法进行清理。以doCleanupRDD为例,其实现见代码清单2。
代码清单2 清理RDD数据
def doCleanupRDD(rddId: Int, blocking: Boolean): Unit = { try {
logDebug("Cleaning RDD " + rddId) sc.unpersistRDD(rddId, blocking)
listeners.asScala.foreach(_.rddCleaned(rddId)) logInfo("Cleaned RDD " + rddId)
} catch { case e: Exception => logError("Error cleaning RDD " + rddId, e) } }
根据代码清单2,doCleanupRDD的执行步骤如下。
* 调用SparkContext的unpersistRDD方法从内存或磁盘中移除RDD。
* 从persistentRdds中移除对RDD的跟踪。
* 调用所有监听器的rddCleaned方法。
热门工具 换一换