Spark的RDD检查点实现分析

  def checkpoint() {
    if (context.checkpointDir.isEmpty) {
      throw new SparkException("Checkpoint directory has not been set in the SparkContext")
    } else if (checkpointData.isEmpty) {
      checkpointData = Some(new RDDCheckpointData(this))
      checkpointData.get.markForCheckpoint()
    }
  }

  def markForCheckpoint() {
    RDDCheckpointData.synchronized {
      if (cpState == Initialized) cpState = MarkedForCheckpoint
    }
  }

markForCheckpoint方法中将cpState置为MarkedForCheckpoint，有着重要意义：一方面表示启用检查点，另一方面只有当cpState等于MarkedForCheckpoint时，才能够保存检查点。

检查点的保存

  def doCheckpoint() {
    RDDCheckpointData.synchronized {
      if (cpState == MarkedForCheckpoint) {
        cpState = CheckpointingInProgress
      } else {
        return
      }
    }

    // Create the output path for the checkpoint
    val path = new Path(rdd.context.checkpointDir.get, "rdd-" + rdd.id)
    val fs = path.getFileSystem(rdd.context.hadoopConfiguration)
    if (!fs.mkdirs(path)) {
      throw new SparkException("Failed to create checkpoint path " + path)
    }

    // Save to file, and reload it as an RDD
    val broadcastedConf = rdd.context.broadcast(
      new SerializableWritable(rdd.context.hadoopConfiguration))
    rdd.context.runJob(rdd, CheckpointRDD.writeToFile[T](path.toString, broadcastedConf) _)
    val newRDD = new CheckpointRDD[T](rdd.context, path.toString)
    if (newRDD.partitions.size != rdd.partitions.size) {
      throw new SparkException(
        "Checkpoint RDD " + newRDD + "(" + newRDD.partitions.size + ") has different " +
          "number of partitions than original RDD " + rdd + "(" + rdd.partitions.size + ")")
    }

    // Change the dependencies and partitions of the RDD
    RDDCheckpointData.synchronized {
      cpFile = Some(path.toString)
      cpRDD = Some(newRDD)
      rdd.markCheckpointed(newRDD)   // Update the RDD‘s dependencies and partitions
      cpState = Checkpointed
    }
    logInfo("Done checkpointing RDD " + rdd.id + " to " + path + ", new parent is RDD " + newRDD.id)
  }

通过阅读doCheckpoint方法，其执行步骤总结如下：

1)        校验用户作业是否启用了检查点，即是否调用了checkpoint方法将cpState置为MarkedForCheckpoint。如果没有启用检查点，则直接返回，不继续进行检查点的保存。

2)        在HDFS上创建用于保存检查点数据的文件路径。其中checkpointDir必须由用户作业调用SparkContext的setCheckpointDir方法（见代码清单5-67）设置。

3)        运行作业，此作业实际执行了CheckpointRDD的writeToFile方法（见代码清单5-68），将检查点数据保存的HDFS上。

  def writeToFile[T: ClassTag](
      path: String,
      broadcastedConf: Broadcast[SerializableWritable[Configuration]],
      blockSize: Int = -1
    )(ctx: TaskContext, iterator: Iterator[T]) {
    val env = SparkEnv.get
    val outputDir = new Path(path)
    val fs = outputDir.getFileSystem(broadcastedConf.value.value)

    val finalOutputName = splitIdToFile(ctx.partitionId)
    val finalOutputPath = new Path(outputDir, finalOutputName)
    val tempOutputPath = new Path(outputDir, "." + finalOutputName + "-attempt-" + ctx.attemptId)

    if (fs.exists(tempOutputPath)) {
      throw new IOException("Checkpoint failed: temporary path " +
        tempOutputPath + " already exists")
    }
    val bufferSize = env.conf.getInt("spark.buffer.size", 65536)

    val fileOutputStream = if (blockSize < 0) {
      fs.create(tempOutputPath, false, bufferSize)
    } else {
      // This is mainly for testing purpose
      fs.create(tempOutputPath, false, bufferSize, fs.getDefaultReplication, blockSize)
    }
    val serializer = env.serializer.newInstance()
    val serializeStream = serializer.serializeStream(fileOutputStream)
    serializeStream.writeAll(iterator)
    serializeStream.close()

    if (!fs.rename(tempOutputPath, finalOutputPath)) {
      if (!fs.exists(finalOutputPath)) {
        logInfo("Deleting tempOutputPath " + tempOutputPath)
        fs.delete(tempOutputPath, false)
        throw new IOException("Checkpoint failed: failed to save output of task: "
          + ctx.attemptId + " and final output path does not exist")
      } else {
        // Some other copy of this task must‘ve finished before us and renamed it
        logInfo("Final output path " + finalOutputPath + " already exists; not overwriting it")
        fs.delete(tempOutputPath, false)
      }
    }
  }

  private[spark] def markCheckpointed(checkpointRDD: RDD[_]) {
    clearDependencies()
    partitions_ = null
    deps = null    // Forget the constructor argument for dependencies too
  }

  protected def clearDependencies() {
    dependencies_ = null
  }

使用检查点

前两个小节分别讲解了检查点如何启用以及启用后如何实现保存的原理和分析，在5.7节一开始介绍了检查点的两种使用场景：

1)        获取RDD的依赖时，如果有了检查点，则从检查点中读取；

2)        获取RDD的分区时，如果有了检查点，则从检查点中读取。

  private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] =
  {
    if (isCheckpointed) firstParent[T].iterator(split, context) else compute(split, context)
  }

  def isCheckpointed: Boolean = checkpointData.exists(_.isCheckpointed)

isCheckpointed实际代理了RDDCheckpointData的isCheckpointed方法（见代码清单5-71），用于判断当前RDD是否已经设置了检查点。

代码清单5-71         判断RDD是否已经保持了检查点

  def isCheckpointed: Boolean = {
    RDDCheckpointData.synchronized { cpState == Checkpointed }
  }

根据之前的分析，我们知道如果已经保存了检查点，那么cpState必然等于Checkpointed，所以isCheckpointed方法将返回true。因此代码清单5-70将会继续执行firstParent[T].iterator(split, context)。而firstParent（见代码清单5-13）首先会调用代码清单5-28所示的dependencies方法，这样计算过程中调用computeOrReadCheckpoint，使用检查点的过程实际退化为我们说的获取RDD依赖时使用检查点的方式。而此时的依赖已经被CheckpointRDD所替代，经过迭代计算（请参考第6章），最终会调用CheckpointRDD的compute方法（见代码清单5-72），从其实现可知从检查点读取计算结果实际就是读取之前分析的写入HDFS的数据。

  override def compute(split: Partition, context: TaskContext): Iterator[T] = {
    val file = new Path(checkpointPath, CheckpointRDD.splitIdToFile(split.index))
    CheckpointRDD.readFromFile(file, broadcastedConf, context)
  }