程序员社区

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块的数据用来迭代计算。

赞(0) 打赏
未经允许不得转载:IDEA激活码 » Spark任务输出追踪器MapOutputTracker详解

相关推荐

  • 暂无文章

一个分享Java & Python知识的社区