union() 操作中的任务执行顺序

Question

我有一个包含 80 个 rdds 的列表，我想处理然后最终加入。 “过程”部分包括为每个 rdd 做一个 map 和一个 reduce 键。 然后我通过联合来“加入”他们。 这是我的代码草图：

rdds0.foreach(_.persist()) //rdds0 are persisted

//trigger a map and a shuffle for each rdd
vals rdds = rdds0.map(rdd => rdd.map(f1).reduceByKey(f2))

//action on the union of the rdds
sparkContext.union(rdds).collect()

但是，生成的 DAG 存在问题。 确实spark生成的DAG是这样的：

• 80 个阶段，每个 RDD 的每个“地图”一个阶段

• 联合的最后阶段，以 80 个 reduceByKey 并行开始

我对粗体部分有疑问。 AFAIK，这意味着对于最后一个任务，Spark 将并行调度 80 个 reducebykey，其中每个都占用大量内存。 一旦为此 RDD 完成了 map 阶段，就能够为每个 rdd 单独执行 reduceByKey() 似乎更有效。 相反，在 所有 map 阶段完成之前不能执行 reduceByKey，然后它们都被同时调度。

有没有办法强制 Spark 以某种方式尽快执行 redueByKey() 操作，而不是等待所有地图任务？ 我认为这是 union() 创建 PartitionerAwareUnionRDD 而不是 UnionRDD() 的问题，但似乎两种 RDD 类型都会生成相同的 DAG。

Answer 1

reduceByKey 是一个广泛的转换 - 这意味着它具有：

• “map-side”组件 - 在 shuffle 之前发生的操作的一部分 - 包含在 DAG 的第一阶段。
• “reduce-side”组件 - 洗牌后发生的操作的一部分 - 包含在 DAG 的第二阶段。

“reduce-side”组件的结果直接通过管道传送到联合。在这种情况下，确实没有什么可优化的。