3.5 RDD内部的计算机制_Spark大数据商业实战三部曲：内核解密|商业案例|性能调优-QQ阅读男生历史网

书名：Spark大数据商业实战三部曲：内核解密|商业案例|性能调优
作者名：王家林
本章字数：2361字
更新时间：2021-03-30 21:55:50

3.5　RDD内部的计算机制

RDD的多个Partition分别由不同的Task处理。Task分为两类：shuffleMapTask、resultTask。本节基于源码对RDD的计算过程进行深度解析。

3.5.1　Task解析

Task是计算运行在集群上的基本计算单位。一个Task负责处理RDD的一个Partition，一个RDD的多个Partition会分别由不同的Task去处理，通过之前对RDD的窄依赖关系的讲解，我们可以发现在RDD的窄依赖中，子RDD中Partition的个数基本都大于等于父RDD中Partition的个数，所以Spark计算中对于每一个Stage分配的Task的数目是基于该Stage中最后一个RDD的Partition的个数来决定的。最后一个RDD如果有100个Partition，则Spark对这个Stage分配100个Task。

Task运行于Executor上，而Executor位于CoarseGrainedExecutorBackend（JVM进程）中。

Spark Job中，根据Task所处Stage的位置，我们将Task分为两类：第一类为shuffleMapTask，指Task所处的Stage不是最后一个Stage，也就是Stage的计算结果还没有输出，而是通过Shuffle交给下一个Stage使用；第二类为resultTask，指Task所处Stage是DAG中最后一个Stage，也就是Stage计算结果需要进行输出等操作，计算到此已经结束；简单地说，Spark Job中除了最后一个Stage的Task为resultTask，其他所有Task都为shuffleMapTask。

3.5.2　计算过程深度解析

Spark中的Job本身内部是由具体的Task构成的，基于Spark程序内部的调度模式，即根据宽依赖的关系，划分不同的Stage，最后一个Stage依赖倒数第二个Stage等，我们从最后一个Stage获取结果；在Stage内部，我们知道有一系列的任务，这些任务被提交到集群上的计算节点进行计算，计算节点执行计算逻辑时，复用位于Executor中线程池中的线程，线程中运行的任务调用具体Task的run方法进行计算，此时，如果调用具体Task的run方法，就需要考虑不同Stage内部具体Task的类型，Spark规定最后一个Stage中的Task的类型为resultTask，因为我们需要获取最后的结果，所以前面所有Stage的Task是shuffleMapTask。

RDD在进行计算前，Driver给其他Executor发送消息，让Executor启动Task，在Executor启动Task成功后，通过消息机制汇报启动成功信息给Driver。Task计算示意图如图3-6所示。

详细情况如下：Driver中的CoarseGrainedSchedulerBackend给CoarseGrainedExecutor-Backend发送LaunchTask消息。

（1）首先反序列化TaskDescription。

Spark 2.1.1版本的CoarseGrainedExecutorBackend.scala的receive的源码如下。

1.    override def receive: PartialFunction[Any, Unit] = {
2.  .......
3.  case LaunchTask(data) =>
4.        if (executor == null) {
5.          exitExecutor(1, "Received LaunchTask command but executor was null")
6.        } else {
7.          val taskDesc = ser.deserialize[TaskDescription](data.value)
8.          logInfo("Got assigned task " + taskDesc.taskId)
9.          executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber
            = taskDesc.attemptNumber, taskDesc.name, taskDesc.serializedTask)
10.       }

图3-6　Task计算示意图

Spark 2.2.0版本的CoarseGrainedExecutorBackend.scala的receive的源码与Spark 2.1.1版本相比具有如下特点。

　上段代码中第7行调整为调用TaskDescription的decode方法，解析读取dataIn、taskId、attemptNumber、executorId、name、index等信息，读取相应的JAR、文件、属性，返回TaskDescription值。

　上段代码中第9行executor.launchTask传入的第二个参数更新为封装的taskDesc值。

1.  .......
2.     val taskDesc = TaskDescription.decode(data.value)
3.   ......
4.       executor.launchTask(this, taskDesc)
5.

launchTask中调用了decode方法，TaskDescription.scala的decode的源码如下。

1.  def decode(byteBuffer: ByteBuffer): TaskDescription = {
2.  val dataIn = new DataInputStream(new ByteBufferInputStream(byteBuffer))
3.  val taskId = dataIn.readLong()
4.  val attemptNumber = dataIn.readInt()
5.  val executorId = dataIn.readUTF()
6.  val name = dataIn.readUTF()
7.  val index = dataIn.readInt()
8.
9.  //读文件
10.     val taskFiles = deserializeStringLongMap(dataIn)
11.
12.     //读取jars包
13.     val taskJars = deserializeStringLongMap(dataIn)
14.
15.     //读取属性
16.     val properties = new Properties()
17.     val numProperties = dataIn.readInt()
18.     for (i <- 0 until numProperties) {
19.       val key = dataIn.readUTF()
20.       val valueLength = dataIn.readInt()
21.       val valueBytes = new Array[Byte](valueLength)
22.       dataIn.readFully(valueBytes)
23.       properties.setProperty(key, new String(valueBytes,
          StandardCharsets.UTF_8))
24.     }
25.
26.     //创建一个子缓冲用于序列化任务将其变成自己的缓冲区（被反序列化后）
27.     val serializedTask = byteBuffer.slice()
28.
29.     new TaskDescription(taskId, attemptNumber, executorId, name, index,
        taskFiles, taskJars,
30.       properties, serializedTask)
31.   }
32. }

（2）Executor会通过launchTask执行Task。

（3）Executor的launchTask方法创建一个TaskRunner实例在threadPool来运行具体的Task。

Spark 2.1.1版本的Executor.scala的launchTask的源码如下。

1.    def launchTask(
2.        context: ExecutorBackend,
3.        taskId: Long,
4.        attemptNumber: Int,
5.        taskName: String,
6.        serializedTask: ByteBuffer): Unit = {
7.  //调用TaskRunner句柄创建TaskRunner对象
8.      val tr = new TaskRunner(context, taskId = taskId, attemptNumber =
        attemptNumber, taskName, serializedTask)
9.  //将创建的TaskRunner对象放入即将进行的堆栈中
10.     runningTasks.put(taskId, tr)
11. //从线程池中分配一条线程给TaskRunner
12.     threadPool.execute(tr)
13.   }

Spark 2.2.0版本的Executor.scala的launchTask的源码与Spark 2.1.1版本相比具有如下特点。

　上段代码中第3～6行调整launchTask方法的第二个参数：传入封装的taskDescription任务描述信息。

　上段代码中第8行构建TaskRunner实例传入的也是taskDescription参数。

1.  def launchTask(context: ExecutorBackend, taskDescription:
    TaskDescription): Unit = {
2.     val tr = new TaskRunner(context, taskDescription)
3.  ......

在TaskRunner的run方法首先会通过statusUpdate给Driver发信息汇报自己的状态，说明自己处于running状态。同时，TaskRunner内部会做一些准备工作，如反序列化Task的依赖，通过网络获取需要的文件、Jar等；然后反序列化Task本身。

Spark 2.1.1版本的Executor.scala的run方法的源码如下。

1.      override def run(): Unit = {
2.        threadId = Thread.currentThread.getId
3.        Thread.currentThread.setName(threadName)
4.        val threadMXBean = ManagementFactory.getThreadMXBean
5.        val taskMemoryManager = new TaskMemoryManager(env.memoryManager,
          taskId)
6.        val deserializeStartTime = System.currentTimeMillis()
7.        val deserializeStartCpuTime = if (threadMXBean.
          isCurrentThreadCpuTimeSupported) {
8.          threadMXBean.getCurrentThreadCpuTime
9.        } else 0L
10.       Thread.currentThread.setContextClassLoader(replClassLoader)
11.       val ser = env.closureSerializer.newInstance()
12.       logInfo(s"Running $taskName (TID $taskId)")
13. //通过statusUpdate给Driver发信息汇报自己状态说明自己是running状态
14.       execBackend.statusUpdate(taskId, TaskState.RUNNING, EMPTY_BYTE_BUFFER)
15.       var taskStart: Long = 0
16.       var taskStartCpu: Long = 0
17.       startGCTime = computeTotalGcTime()
18.       try {
19. //反序列化Task的依赖
20.         val (taskFiles, taskJars, taskProps, taskBytes) =
21.           Task.deserializeWithDependencies(serializedTask)
22.
23.         Executor.taskDeserializationProps.set(taskProps)
24.         updateDependencies(taskFiles, taskJars)
25. //反序列化Task本身
26.        task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.
           getContextClassLoader)
27.         task.localProperties = taskProps
28.         task.setTaskMemoryManager(taskMemoryManager)
29. ......

Spark 2.2.0版本的Executor.scala的run方法的源码与Spark 2.1.1版本相比具有如下特点。

　上段代码中第20～21行删掉，即删掉val (taskFiles, taskJars, taskProps, taskBytes) =Task.deserializeWithDependencies(serializedTask)。

　第23行代码Executor.taskDeserializationProps.set方法的参数将taskProps调整为taskDescription.properties。

　第24行代码updateDependencies方法的参数将taskFiles、taskJars分别调整为taskDescription.addedFiles、taskDescription.addedJars。

　第26行代码ser.deserialize方法的第一个参数将taskBytes调整为taskDescription.serializedTask。

　第27行代码将taskProps调整为taskDescription.properties。

1.  ......
2.     Executor.taskDeserializationProps.set(taskDescription.properties)
3.    updateDependencies(taskDescription.addedFiles, taskDescription.
      addedJars)
4.    task = ser.deserialize[Task[Any]](
5.      taskDescription.serializedTask, Thread.currentThread.
        getContextClassLoader)
6.    task.localProperties = taskDescription.properties
7.  ......

（4）调用反序列化后的Task.run方法来执行任务，并获得执行结果。

Spark 2.1.1版本的Executor.scala的run方法的源码如下。

1.     override def run(): Unit = {
2.  ......
3.   //Task计算开始时间
4.    taskStart = System.currentTimeMillis()
5.          taskStartCpu = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
6.            threadMXBean.getCurrentThreadCpuTime
7.          } else 0L
8.          var threwException = true
9.          val value = try {
10.  //运行Task的run方法
11.           val res: Any = task.run(
12.             taskAttemptId = taskId,
13.             attemptNumber = attemptNumber,
14.             metricsSystem = env.metricsSystem)
15.           threwException = false
16.           res
17.         } finally {
18.           val   releasedLocks     =   env.blockManager.releaseAllLocksForTask
              (taskId)
19.           val freedMemory = taskMemoryManager.cleanUpAllAllocatedMemory()
20.
21.           if (freedMemory > 0 && !threwException) {
22.             val errMsg = s"Managed memory leak detected; size = $freedMemory
                bytes, TID = $taskId"
23.             if(conf.getBoolean("spark.unsafe.exceptionOnMemoryLeak",false)){
24.               throw new SparkException(errMsg)
25.             } else {
26.               logWarning(errMsg)
27.             }
28.           }
29.
30.           if (releasedLocks.nonEmpty && !threwException) {
31.             val errMsg =
32.               s"${releasedLocks.size} block locks were not released by TID
                  = $taskId:\n" +
33.                 releasedLocks.mkString("[", ", ", "]")
34.             if (conf.getBoolean("spark.storage.exceptionOnPinLeak",
                false)) {
35.               throw new SparkException(errMsg)
36.             } else {
37.               logWarning(errMsg)
38.             }
39.           }
40.         }
41. //计算完成时间
42.         val taskFinish = System.currentTimeMillis()
43. ......

Spark 2.2.0版本的Executor.scala的run方法的源码与Spark 2.1.1版本相比具有如下特点。

　上段代码中第13行attemptNumber调整为taskDescription.attemptNumber。

　上段代码中第40行之后新增一段代码：在任务完成前，循环遍历任务，抓取失败的情况，打印日志提醒用户业务代码可能导致任务失败。

1.  ......
2.    attemptNumber = taskDescription.attemptNumber,
3.  ......
4.          task.context.fetchFailed.foreach { fetchFailure =>
5.  //用户代码在不抛出任何错误的情况下捕获了故障。可能是用户打算这样做的（虽然不太可能），
    //因此我们将记录一个错误并继续下去
6.            logError(s"TID ${taskId} completed successfully though
              internally it encountered " + s"unrecoverable fetch failures!
              Most likely this means user code is incorrectly " + s"swallowing
              Spark's internal ${classOf[FetchFailedException]}",
              fetchFailure) }
7.        ......

task.run方法调用了runTask的方法，而runTask方法是一个抽象方法，runTask方法内部会调用RDD的iterator()方法，该方法就是针对当前Task对应的Partition进行计算的关键所在，在处理的方法内部会迭代Partition的元素，并交给我们自定义的function进行处理。

Task.scala的run方法的源码如下。

1.    final def run(
2.        taskAttemptId: Long,
3.        attemptNumber: Int,
4.        metricsSystem: MetricsSystem): T = {
5.      ......
6.      try {
7.        runTask(context)
8.      } catch
9.  ......

task有两个子类，分别是ShuffleMapTask和ResultTask，下面分别对两者进行讲解。

1．ShuffleMapTask

ShuffleMapTask.scala的源码如下。

1.   override def runTask(context: TaskContext): MapStatus = {
2.      //使用广播变量反序列化RDD
3.      val threadMXBean = ManagementFactory.getThreadMXBean
4.      val deserializeStartTime = System.currentTimeMillis()
5.      val deserializeStartCpuTime = if (threadMXBean.
        isCurrentThreadCpuTimeSupported) {
6.        threadMXBean.getCurrentThreadCpuTime
7.      } else 0L
8.  //创建序列化器
9.      val ser = SparkEnv.get.closureSerializer.newInstance()
10. //反序列化出RDD和依赖关系
11. val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
12.       ByteBuffer.wrap(taskBinary.value), Thread.currentThread.
          getContextClassLoader)
13. //RDD反序列化的时间
14.  _executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
15.     _executorDeserializeCpuTime = if (threadMXBean.
        isCurrentThreadCpuTimeSupported) {
16.       threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
17.     } else 0L
18.  //创建Shuffle的writer对象，用来将计算结果写入Shuffle管理器
19.     var writer: ShuffleWriter[Any, Any] = null
20.     try {
21. //实例化shuffleManager
22.       val manager = SparkEnv.get.shuffleManager
23.   //对writer对象赋值
24.      writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId,
         context)
25. //将计算结果通过writer对象的write方法写入shuffle过程
26.  writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_
     <: Product2[Any, Any]]])
27.       writer.stop(success = true).get
28.     } catch {
29.       case e: Exception =>
30.         try {
31.           if (writer != null) {
32.             writer.stop(success = false)
33.           }
34.         } catch {
35.           case e: Exception =>
36.             log.debug("Could not stop writer", e)
37.         }
38.         throw e
39.     }
40.   }

首先，ShuffleMapTask会反序列化RDD及其依赖关系，然后通过调用RDD的iterator方法进行计算，而iterator方法中进行的最终运算的方法是compute()。

RDD.scala的iterator方法的源码如下。

1.   final def iterator(split: Partition, context: TaskContext): Iterator[T]
     = {//判断此RDD的持久化等级是否为NONE(不进行持久化)
2.    if (storageLevel != StorageLevel.NONE) {
3.      getOrCompute(split, context)
4.    } else {
5.      computeOrReadCheckpoint(split, context)
6.    }
7.  }

其中，RDD.scala的computeOrReadCheckpoint的源码如下。

1.   private[spark] def computeOrReadCheckpoint(split: Partition, context:
     TaskContext): Iterator[T] =
2.  {
3.    if (isCheckpointedAndMaterialized) {
4.      firstParent[T].iterator(split, context)
5.    } else {
6.      compute(split, context)
7.    }
8.  }

RDD的compute方法是一个抽象方法，每个RDD都需要重写的方法。

此时，选择查看MapPartitionsRDD已经实现的compute方法，可以发现compute方法的实现是通过f方法实现的，而f方法就是我们创建MapPartitionsRDD时输入的操作函数。

1.      private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag](
2.      var prev: RDD[T],
3.      f: (TaskContext, Int, Iterator[T]) => Iterator[U],  //(TaskContext,
        partition index, iterator)
4.      preservesPartitioning: Boolean = false)
5.    extends RDD[U](prev) {
6.
7.    override val partitioner = if (preservesPartitioning) firstParent[T].
      partitioner else None
8.
9.    override def getPartitions: Array[Partition] = firstParent[T].
      partitions
10.
11.   override def compute(split: Partition, context: TaskContext): Iterator
      [U] =
12.     f(context, split.index, firstParent[T].iterator(split, context))
13.
14.   override def clearDependencies() {
15.     super.clearDependencies()
16.     prev = null
17.   }
18. }

MapPartitionsRDD.scala的源码如下。

注意：通过迭代器的不断叠加，将每个RDD的小函数合并成一个大的函数流。

然后在计算具体的Partition之后，通过shuffleManager获得的shuffleWriter把当前Task计算的结果根据具体的shuffleManager实现写入到具体的文件中，操作完成后会把MapStatus发送给Driver端的DAGScheduler的MapOutputTracker。

2．ResultTask

Driver端的DAGScheduler的MapOutputTracker把shuffleMapTask执行的结果交给ResultTask，ResultTask根据前面Stage的执行结果进行shuffle后产生整个job最后的结果。

ResultTask.scala的runTask的源码如下。

1.   override def runTask(context: TaskContext): U = {
2.     //使用广播变量反序列化RDD及函数
3.     val threadMXBean = ManagementFactory.getThreadMXBean
4.     val deserializeStartTime = System.currentTimeMillis()
5.     val deserializeStartCpuTime = if (threadMXBean.
       isCurrentThreadCpuTimeSupported) {
6.       threadMXBean.getCurrentThreadCpuTime
7.     } else 0L
8.     //创建序列化器
9.     val ser = SparkEnv.get.closureSerializer.newInstance()
10.    //反序列RDD和func处理函数
11.    val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T])
       => U)](
12.      ByteBuffer.wrap(taskBinary.value), Thread.currentThread.
         getContextClassLoader)
13.    _executorDeserializeTime = System.currentTimeMillis() -
       deserializeStartTime
14.    _executorDeserializeCpuTime = if (threadMXBean.
       isCurrentThreadCpuTimeSupported) {
15.      threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
16.    } else 0L
17.
18.    func(context, rdd.iterator(partition, context))
19.  }

而ResultTask的runTask方法中反序列化生成func函数，最后通过func函数计算出最终的结果。