架构

    重新贴一下 Overview 中给出的部署图：

    接下来分阶段讨论并细化这个图。

    Job 提交

    下图展示了driver program（假设在 master node 上运行）如何生成 job，并提交到 worker node 上执行。

    JobSubmission

    Driver 端的逻辑如果用代码表示：

    代码的文字描述：

    当用户的 program 调用 时，这个语句会帮助 program 启动诸多有关 driver 通信、job 执行的对象、线程、actor等，该语句确立了 program 的 driver 地位。

    Driver program 中的 transformation() 建立 computing chain（一系列的 RDD），每个 RDD 的 compute() 定义数据来了怎么计算得到该 RDD 中 partition 的结果，getDependencies() 定义 RDD 之间 partition 的数据依赖。

    每个 action() 触发生成一个 job，在 dagScheduler.runJob() 的时候进行 stage 划分，在 submitStage() 的时候生成该 stage 包含的具体的 ShuffleMapTasks 或者 ResultTasks，然后将 tasks 打包成 TaskSet 交给 taskScheduler，如果 taskSet 可以运行就将 tasks 交给 sparkDeploySchedulerBackend 去分配执行。

    Worker 端接收到 tasks 后，执行如下操作

    1. coarseGrainedExecutorBackend ! LaunchTask(serializedTask)
    2. => executor.launchTask()
    3. => executor.threadPool.execute(new TaskRunner(taskId, serializedTask))

    executor 将 task 包装成 taskRunner，并从线程池中抽取出一个空闲线程运行 task。一个 CoarseGrainedExecutorBackend 进程有且仅有一个 executor 对象。

    Task 运行

    下图展示了 task 被分配到 worker node 上后的执行流程及 driver 如何处理 task 的 result。

    Executor 收到 serialized 的 task 后，先 deserialize 出正常的 task，然后运行 task 得到其执行结果 directResult，这个结果要送回到 driver 那里。但是通过 Actor 发送的数据包不宜过大，如果 result 比较大（比如 groupByKey 的 result）先把 result 存放到本地的“内存＋磁盘”上，由 blockManager 来管理，只把存储位置信息（indirectResult）发送给 driver，driver 需要实际的 result 的时候，会通过 HTTP 去 fetch。如果 result 不大（小于spark.akka.frameSize = 10MB），那么直接发送给 driver。

    上面的描述还有一些细节：如果 task 运行结束生成的 directResult > akka.frameSize，directResult 会被存放到由 blockManager 管理的本地“内存＋磁盘”上。BlockManager 中的 memoryStore 开辟了一个 LinkedHashMap 来存储要存放到本地内存的数据。LinkedHashMap 存储的数据总大小不超过 Runtime.getRuntime.maxMemory * spark.storage.memoryFraction(default 0.6) 。如果 LinkedHashMap 剩余空间不足以存放新来的数据，就将数据交给 diskStore 存放到磁盘上，但前提是该数据的 storageLevel 中包含“磁盘”。

    1. In TaskRunner.run()
    2. // deserialize task, run it and then send the result to 
    3. => coarseGrainedExecutorBackend.statusUpdate()
    4. => task = ser.deserialize(serializedTask)
    5. => value = task.run(taskId)
    6. => directResult = new DirectTaskResult(ser.serialize(value))
    7. => if( directResult.size() > akkaFrameSize() ) 
    8.        indirectResult = blockManager.putBytes(taskId, directResult, MEMORY+DISK+SER)
    9.    else
    10.        return directResult
    11. => coarseGrainedExecutorBackend.statusUpdate(result)
    12. => driver ! StatusUpdate(executorId, taskId, result)

    ShuffleMapTask 和 ResultTask 生成的 result 不一样。ShuffleMapTask 生成的是 MapStatus，MapStatus 包含两项内容：一是该 task 所在的 BlockManager 的 BlockManagerId（实际是 executorId + host, port, nettyPort），二是 task 输出的每个 FileSegment 大小。ResultTask 生成的 result 的是 func 在 partition 上的执行结果。比如 count() 的 func 就是统计 partition 中 records 的个数。由于 ShuffleMapTask 需要将 FileSegment 写入磁盘，因此需要输出流 writers，这些 writers 是由 blockManger 里面的 shuffleBlockManager 产生和控制的。

    Driver 收到 task 的执行结果 result 后会进行一系列的操作：首先告诉 taskScheduler 这个 task 已经执行完，然后去分析 result。由于 result 可能是 indirectResult，需要先调用 blockManager.getRemoteBytes() 去 fech 实际的 result，这个过程下节会详解。得到实际的 result 后，需要分情况分析，如果是 ResultTask 的 result，那么可以使用 ResultHandler 对 result 进行 driver 端的计算（比如 count() 会对所有 ResultTask 的 result 作 sum），如果 result 是 ShuffleMapTask 的 MapStatus，那么需要将 MapStatus（ShuffleMapTask 输出的 FileSegment 的位置和大小信息）存放到 mapOutputTrackerMaster 中的 mapStatuses 数据结构中以便以后 reducer shuffle 的时候查询。如果 driver 收到的 task 是该 stage 中的最后一个 task，那么可以 submit 下一个 stage，如果该 stage 已经是最后一个 stage，那么告诉 dagScheduler job 已经完成。

    1. After driver receives StatusUpdate(result)
    2. => taskScheduler.statusUpdate(taskId, state, result.value)
    3. => if result is IndirectResult
    4.       serializedTaskResult = blockManager.getRemoteBytes(IndirectResult.blockId)
    5. => scheduler.handleSuccessfulTask(taskSetManager, tid, result)
    6. => taskSetManager.handleSuccessfulTask(tid, taskResult)
    7. => dagScheduler.taskEnded(result.value, result.accumUpdates)
    9. => dagScheduler.handleTaskCompletion(completion)
    10. => Accumulators.add(event.accumUpdates)
    12. // If the finished task is ResultTask
    13. => if (job.numFinished == job.numPartitions) 
    14.       listenerBus.post(SparkListenerJobEnd(job.jobId, JobSucceeded))
    15. => job.listener.taskSucceeded(outputId, result)
    16. =>    jobWaiter.taskSucceeded(index, result)
    17. =>    resultHandler(index, result)
    19. // if the finished task is ShuffleMapTask
    20. => stage.addOutputLoc(smt.partitionId, status)
    21. => if (all tasks in current stage have finished)
    22.       mapOutputTrackerMaster.registerMapOutputs(shuffleId, Array[MapStatus])
    23.       mapStatuses.put(shuffleId, Array[MapStatus]() ++ statuses)
    24. => submitStage(stage)

    上一节描述了 task 运行过程及 result 的处理过程，这一节描述 reducer（需要 shuffle 的 task ）是如何获取到输入数据的。关于 reducer 如何处理输入数据已经在上一章的 shuffle read 中解释了。

    问题：reducer 怎么知道要去哪里 fetch 数据？

    readMapStatus
    reducer 首先要知道 parent stage 中 ShuffleMapTask 输出的 FileSegments 在哪个节点。这个信息在 ShuffleMapTask 完成时已经送到了 driver 的 mapOutputTrackerMaster，并存放到了 mapStatuses: HashMap 里面，给定 stageId，可以获取该 stage 中 ShuffleMapTasks 生成的 FileSegments 信息 Array[MapStatus]，通过 Array(taskId) 就可以得到某个 task 输出的 FileSegments 位置（blockManagerId）及每个 FileSegment 大小。

    1. rdd.iterator()
    2. => rdd(e.g., ShuffledRDD/CoGroupedRDD).compute()
    3. => SparkEnv.get.shuffleFetcher.fetch(shuffledId, split.index, context, ser)
    4. => blockStoreShuffleFetcher.fetch(shuffleId, reduceId, context, serializer)
    6. => blocksByAddress: Seq[(BlockManagerId, Seq[(BlockId, Long)])] = compute(statuses)
    7. => basicBlockFetcherIterator = blockManager.getMultiple(blocksByAddress, serializer)
    8. => itr = basicBlockFetcherIterator.flatMap(unpackBlock)

    basicBlockFetcherIterator 收到获取数据的任务后，会生成一个个 fetchRequest，每个 fetchRequest 包含去某个节点获取若干个 FileSegments 的任务。图中展示了 reducer-2 需要从三个 worker node 上获取所需的白色 FileSegment (FS)。总的数据获取任务由 blocksByAddress 表示，要从第一个 node 获取 4 个，从第二个 node 获取 3 个，从第三个 node 获取 4 个。

    为了加快任务获取过程，显然要将总任务划分为子任务（fetchRequest），然后为每个任务分配一个线程去 fetch。Spark 为每个 reducer 启动 5 个并行 fetch 的线程（Hadoop 也是默认启动 5 个）。由于 fetch 来的数据会先被放到内存作缓冲，因此一次 fetch 的数据不能太多，Spark 设定不能超过 。注意这 48MB 的空间是由这 5 个 fetch 线程共享的，因此在划分子任务时，尽量使得 fetchRequest 不超过48MB / 5 = 9.6MB。如图在 node 1 中，Size(FS0-2) + Size(FS1-2) < 9.6MB 但是 Size(FS0-2) + Size(FS1-2) + Size(FS2-2) > 9.6MB，因此要在 t1-r2 和 t2-r2 处断开，所以图中有两个 fetchRequest 都是要去 node 1 fetch。那么会不会有 fetchRequest 超过 9.6MB？当然会有，如果某个 FileSegment 特别大，仍然需要一次性将这个 FileSegment fetch 过来。另外，如果 reducer 需要的某些 FileSegment 就在本节点上，那么直接进行 local read。最后，将 fetch 来的 FileSegment 进行 deserialize，将里面的 records 以 iterator 的形式提供给 rdd.compute()，整个 shuffle read 结束。

    下面再讨论一些细节问题：

    reducer 如何将 fetchRequest 信息发送到目标节点？目标节点如何处理 fetchRequest 信息，如何读取 FileSegment 并回送给 reducer？

    fetchrequest

    rdd.iterator() 碰到 ShuffleDependency 时会调用 BasicBlockFetcherIterator 去获取 FileSegments。BasicBlockFetcherIterator 使用 blockManager 中的 connectionManager 将 fetchRequest 发送给其他节点的 connectionManager。connectionManager 之间使用 NIO 模式通信。其他节点，比如 worker node 2 上的 connectionManager 收到消息后，会交给 blockManagerWorker 处理，blockManagerWorker 使用 blockManager 中的 diskStore 去本地磁盘上读取 fetchRequest 要求的 FileSegments，然后仍然通过 connectionManager 将 FileSegments 发送回去。如果使用了 FileConsolidation，diskStore 还需要 shuffleBlockManager 来提供 blockId 所在的具体位置。如果 FileSegment 不超过 spark.storage.memoryMapThreshold=8KB ，那么 diskStore 在读取 FileSegment 的时候会直接将 FileSegment 放到内存中，否则，会使用 RandomAccessFile 中 FileChannel 的内存映射方法来读取 FileSegment（这样可以将大的 FileSegment 加载到内存）。

    当 BasicBlockFetcherIterator 收到其他节点返回的 serialized FileSegments 后会将其放到 fetchResults: Queue 里面，并进行 deserialization，所以 fetchResults: Queue 就相当于在 Shuffle details 那一章提到的 softBuffer。如果 BasicBlockFetcherIterator 所需的某些 FileSegments 就在本地，会通过 diskStore 直接从本地文件读取，并放到 fetchResults 里面。最后 reducer 一边从 FileSegment 中边读取 records 一边处理。

    1. After the blockManager receives the fetch request
    3. => connectionManager.receiveMessage(bufferMessage)
    4. => handleMessage(connectionManagerId, message, connection)
    6. // invoke blockManagerWorker to read the block (FileSegment)
    7. => blockManagerWorker.onBlockMessageReceive()
    8. => blockManagerWorker.processBlockMessage(blockMessage)
    9. => buffer = blockManager.getLocalBytes(blockId)
    10. => buffer = diskStore.getBytes(blockId)
    11. => fileSegment = diskManager.getBlockLocation(blockId)
    12. => shuffleManager.getBlockLocation()
    13. => if(fileSegment < minMemoryMapBytes)
    14.      buffer = ByteBuffer.allocate(fileSegment)
    15.    else
    16.      channel.map(MapMode.READ_ONLY, segment.offset, segment.length)

    每个 reducer 都持有一个 BasicBlockFetcherIterator，一个 BasicBlockFetcherIterator 理论上可以持有 48MB 的 fetchResults。每当 fetchResults 中有一个 FileSegment 被读取完，就会一下子去 fetch 很多个 FileSegment，直到 48MB 被填满。

    1. BasicBlockFetcherIterator.next()
    2. => result = results.task()
    3. => while (!fetchRequests.isEmpty &&
    4.         (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size <= maxBytesInFlight)) {
    5.         sendRequest(fetchRequests.dequeue())
    6.       }

    Discussion

    这一章写了三天，也是我这个月来心情最不好的几天。Anyway，继续总结。

    架构部分其实没有什么好说的，就是设计时尽量功能独立，模块独立，松耦合。BlockManager 设计的不错，就是管的东西太多（数据块、内存、磁盘、通信）。

    如果想进一步了解 blockManager，可以参阅 Jerry Shao 写的 。