Spark Shuffle Tracking 原理分析(1)

这篇具有很好参考价值的文章主要介绍了Spark Shuffle Tracking 原理分析(1)。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

如果开启了 ESS，那么 Executor 计算完后，把 shuffle 数据交给 ESS, Executor 没有任务时，可以安全退出，下游任务从 ESS 拉取 shuffle 数据。

1. 背景

如果 Executor 执行了上游的 Shuffle Map Task 并且把 shuffle 数据些到本地。并且现在 Executor 没有 Task 运行，那么此 Executor 是否能销毁？

现状是如果 Executor 没有 active 的 shuffle 数据，则可以被销毁。
active shuffle 的定义：如果 Shuffle Map Stage 的 task 把 shuffle 数据输出到本地。如果依赖此 shuffle 的Stage 没有计算完毕，则称此 shuffle 为 active shuffle。因为依赖此 shuffle 的 Task 可能从 Driver 端获取了 MapStatus，但是还没有拉取完 shuffle 数据。

为了达到此目的，需要跟踪每个 Stage 和每个 Task 的运行信息。并且启动定时任务，定时扫描每个 Executor，判断是否有任务运行，是否有 active 的 shuffle，如果没有则可以退出。

退出有两种，如果开启了 decommission，则到期的 executors 进入 decommission 模式，否则执行 killExecutors。

参数配置

spark.dynamicAllocation.shuffleTracking.enabled: 默认 true，是否开启 shuffle tracking。
spark.dynamicAllocation.shuffleTracking.timeout: 默认 Long.MaxValue，

2. 设计

ExecutorMonitor 为每个 Executor 创建一个 Tracker, 用于跟踪此 Executor 的状态。

private val executors = new ConcurrentHashMap[String, Tracker]()

定时任务间隔时间查找 timeout 的 executor，然后处理。

timedOutExecutors 方法的主要逻辑，就是遍历 executors。如果 executor 没有 active 的 shuffle 并且当前时间大于 executor 的超时时间 timeoutAt，则此 executor 可以被安全释放。

为什么 executor 有 active shuffle 数据就不能 kill?
Spark Shuffle Tracking 原理分析(1),2024年程序员学习,spark,大数据,分布式

Shuffle 的过程：

MapTask 把 shuffle 写到本地，并且把状态汇报给 Driver.
Reduce Task 从 Driver 获取 shuffle status，并从 shuffle status 获取每个 shuffle 数据的地址。
连接对应的 executor 获取 shuffle 数据。

如果在 reduce 获取完 shuffle status 后，MapTask 所在的 Executor 被 kill 掉，Reduce Task 就无法获取 shuffle 数据。

如果执行 decommission 逻辑，把 MapTask 的 shuffle 数据长传到 bos 等分布式存储是否可以？

也是不可以的，因为 reduce 可能已经把 shuffle status 拿走，获取的 shuffle status 没有记录 shuffle 数据在分布式存储上。

参考： ExecutorMonitor，ExecutorAllocationManager

Executor 状态的更新

ExecutorMonitor 实现了 SparkListner 接口，当 Job, Stage, Task 等 start 和 end 时，都会执行回调。

以 hasActiveShuffle 为例
每个 executor 用一个集合 shuffleIds 存储其上拥有的 shuffle 数据。当其为空时，说明没有 shuffle 数据。

在 onTaskEnd 和 onBlockUpdated 时调用 addShuffle 向 shuffleIds 添加数据。

在以下时机删除 shuffleIds 里的数据。

依赖 driver 端的 ContextCleaner，当 ShuffleRDD 仅有 weakReference 时触发。
rdd.cleanShuffleDependencies 方法，但是此方法仅在 org.apache.spark.ml.recommendation.ALS 使用。

timeoutAt 的计算逻辑

总结：timeoutAt 根据 idle 的时间，spark.dynamicAllocation.cachedExecutorIdleTimeout 和 spark.dynamicAllocation.shuffleTracking.timeout 这 3 个值中最大的值。

详细计算逻辑：
timeoutAt 在一些事件发生时触发计算，如 onBlockUpdated, onUnpersistRDD, updateRunningTasks, removeShuffle, updateActiveShuffles
timeoutAt 的计算逻辑：
当执行器有计算任务时为 Long.MaxValue。
否则为 max(_cacheTimeout, _shuffleTimeout, idleTimeoutNs)
_cacheTimeout: 如果没有 cache 数据，为0，否则为参数 spark.dynamicAllocation.cachedExecutorIdleTimeout 的值（默认 Long.MaxValue）。

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
Spark Shuffle Tracking 原理分析(1),2024年程序员学习,spark,大数据,分布式