Spark源码学习1.4——MapOutputTracker.scala

相关类:MapOutputTrackerMessage,GetMapOutputStatuses extends MapPutputTrackerMessage,StopMapOutputTracker extends MapOutputTrackerMessage,MapOutputTrackerMasterActor,MapOutputTrackerMaster。

首先重写MapOutputTrackerMasterActor的receiveWithLogging:判断MapOutputTracker需要处理的类型是GetMapOutputStatuses(),获取MapOutputTracker的状态;或者是StopMapOutputTracker,停止MapOutputTracker。

一、MapOutputTracker是一个abstract抽象类。获取的Map out的信息根据master和worker有不同的用途:master上,用来记录ShuffleMapTasks所需的map out的源;worker上,仅仅作为cache用来执行shuffle计算。

1、askTracker():检查MapOutputTracker的连接是否正常。

2、sendTracker():检查MapOutPutTracker是否正常工作(发送任意信息返回true)。

3、getServerStatuses():根据参数shuffle id来获取shuffle对应的map out所在的位置及信息。如果没有直接的对应shuffle id的信息,则需要从所有的map中匹配对应shuffle id的map out。

4、getEpoch()和updateEpoch():获取和更新epoch的值。epoch的值是与master同步的,保证map outs是最新的有用的。

二、MapOutPutTrackerMaster针对master的MapOutPutTracker,按照前文的意思,它的作用是为每个shuffle准备其所需要的所有map out,可以加速map outs传送给shuffle的速度。在存储map out的HashMap中,HashMap是基于时间戳的,因此map outs被减少只能因为它被注销掉或者生命周期耗尽。

1、registerShuffle():在map out的集合mapStatuses中注册新的Shuffle,参数为Shuffle id和map的个数。

2、registerMapOutPut():根据Shuffle id在mapStatuses中为Shuffle添加map out的状态(存储的map out其实就是map out的状态)。

3、registerMapOutPuts():同时添加多个map out。

4、unregisterMapOutPut():在mapStatuses中注销给定Shuffle的map out。

5、重写unrigesterShuffle():移除mapStatuses中的给定Shuffle的map out。

6、containShuffle():判断是否存在给定的Shuffle。

7、incrementEpoch():同步epoch加一。

8、getSerializedMapOutputStatuses():给定Shuffle id,返回其map out集合。首先是对epoch进行锁状态下的同步,保证获取资源的正确性;其次,根据Shuffle id获取指定位置的statuses,如果指定位置没有对应Shuffle id的statuses,那么获取这个位置的statuses快照返回,作为参考;最后,如果操作的epoch与锁状态下的epoch是一致的,将获取到的statuses存入缓存。

9、stop():停止MapOutPutTracker,清除mapStatuses,清空缓存。

10、cleanup():在指定时间清除mapStatuses和cachedSerializedStatuses。

三、MapOutPutTracker对象。它通过serializedMapStatuses将map out流通过gzip的压缩方式压缩(压缩是可行的,因为很多map out基于同样的hostname),这样方便数据流传递给reduce进行操作。

上一篇:Spark源码学习2


下一篇:delphi 获取文件的最新修改时间 http://www.delphitop.com/html/wenjian/64.html