Spark任务输出追踪器MapOutputTracker详解

一.什么是shuffle

MapOutputTrancker用于跟踪map任务的输出状态，此状态便于reduce任务定位到map输出结果所在的节点地址，进而获取中间输出结果，每个map任务或者reduce任务，都会有其唯一的标识，分别为mapid和reduceid，每个reduce任务的输入可能是多个map任务的输出，因为reduce可能会到多个map任务所在的节点上去拉取Block，这一过程叫做shuffle，每次shuffle的过程都有其唯一的标识shuffleid.

二.MapOutputTrancker的创建方式

在Driver端和Executor端启动的同时，都会创建MapOutputTracker的实例，不同的是Driver端创建的是MapOutputTrackerMaster，Executor端创建的是MapOutputTranckerWoker。

• Driver端启动时会创建MapOutputTrackerMaster，之后创建MapOutputTrackerMasterEndpoint,并且注册到Dispatcher中，端点名称为MapOutputTrancker。

• Executor端创建MapOutputTranckerWorker，不仅会和Driver端一样，注册端点信息等，而且会从远端Driver获取之前在NettyRpcEnv的Dispatcher中注册好的MapOutputTrackerMasterEndpoint的引用。

三.MapOutputTrancker的属性

• trackerEndpoint：持有Driver端上MapOutputTrackerMasterEndpoint的引用Ref
• mapStatuses：用于维护各个map任务输出的状态，类型为Map[Int，Array[MapStatus]],key为shuffleid，Array存储着各个map任务对于的状态信息mapStatus。由于各个MapOutputTranckerWoker会不断向MapOutputTranckerMaster汇报本节点的Executor运行的map任务状态信息，因此MapOutputTranckerMaster中的mapStatuses中维护的信息是最新最全的。而MapOutputTrackerWorker的mapStatuses对于本节点上的map任务状态是及时更新的，对于其他节点的map任务状态则是一个缓冲，如果后续在获取mapStatus时，无法命中缓存，则向Drievr端的MapOutputTranckerMaster获取最新的任务状态信息。
• fetching：shuffle拉取的集合，用来记录当前Executor正在从哪些Map输出的位置拉取数据。

四.获取mapStatus的流程

1.首先 从当前Executor中的MapOutputTracker的mapStatuses缓存中，获取MapStatus数组，如果没有则向远端Driver中的MapOutputTranckerMaster去获取任务状态信息。

2.然后 判断fetching中是否已经存在要获取的shuffleid，如果有，这就说明有其他线程对此shuffleid的数据进行远程拉取了，这样就等待其他线程拉取完毕，直到fetching中不存在要取的shuffleid，这时就从mapStatuses中再次获取mapStatus集合。

3.如果还获取不到，则说明其他线程拉取失败了，则需要自己去拉取数据，首先将shuufleid加入fetching集合中，表示当前shuffleid的任务状态信息，已经有线程在拉取了，之后会调用ackTracker方法，向MapOutputTrackerMasterEndpoint发送消息去获取map任务的状态信息。

4.之后 MapOutputTrackerMaster接受到该消息之后，将请求包装成MapOutputMessage消息，放入到消息队列，异步的去处理该消息。

• 首先会getSerializedMapOutputStatuses方法，查询本地记录shuffle对应的Map输出状态。

• 在获取的过程中需要为每个shuffleId分配一个分段锁，因为这里支持并发调用，同一时间有多个线程需要获取同一个shuffleId对应的输出，所以需要保证Map元数据信息只序列化或者广播一次。所以在获取锁之前和得到锁之后都需要再次查询一下缓存，可能有其他线程已经缓存了MapStatus。

• 如果缓存还是为空，则需要将MapStatus序列化或者包装为Broadcast。对于序列化还是广播，通过比较序列化后的结果大小是否超出spark.shuffle.mapOutput.minSizeForBroadcast，默认值为512K。

• 序列化完成后，将此结果进行缓存，并向MapOutputTrackerWorker返回结果。

• MapOutputTrackerWorker的askTracker接收到返回的结果后结束阻塞，将数据反序列化成mapStatus集合缓存下来，然后将shuffleid从fetching中移除，唤醒哪些在fetching锁上等待的线程，使这些线程可以获取自己需要的MapStatus数组。

5.最后 返回任务状态信息mapStatus数组。

6.注意 MapOutputTrancker中会有线程池，区别于Dispatcher中的线程池，同时还有MessageLoop，和Dispatcher中非常相似。

五.ShuffleReader如何使用mapStatus

1.在ShuffleRDD的compute方法中，会获取BlockStoreShuffleReader，然后在BlockStoreShuffleReader中，会调用mapOutputTracker.getMapSizesByExecutorId方法获取一组二元组序列Seq[(BlockManagerId, Seq[(BlockId, Long)])]，第一项代表了BlockManagerId，第二项描述了存储于该BlockManager上的一组shuffle blocks。

2.getMapSizesByExecutorId会调用getStatuses方法获取MapStatus集合，然后最后返回MapStatus集合。

3.最后根据执行的分区范围[startPartition, endPartition]将返回的结果Array[MapStatus]转换成Seq[(BlockManagerId, Seq[(BlockId, Long)])]。

4.利用这个Seq[(BlockManagerId, Seq[(BlockId, Long)])]，去指定的BlockManager中去拉取对应的Block块的数据用来迭代计算。