Spark2.1.0——ContextCleaner的工作原理分析

创建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的步骤如下。

1. 将cleaningThread设置为守护线程，并指定名称为Spark Context Cleaner。
2. 启动cleaningThread。
3. 给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的执行步骤如下。

1. 调用SparkContext的unpersistRDD方法从内存或磁盘中移除RDD。
2. 从persistentRdds中移除对RDD的跟踪。
3. 调用所有***的rddCleaned方法。