Spark中sort()导致前置transformations重复执行两次的问题咨询

现象合理性判断 #

这个现象是正常的，属于Spark全局排序的固有机制，并非异常问题。

深层原因 #

Spark的sort()（Dataset API对应orderBy）实现全局排序时，需要完成两个核心步骤，这两个步骤都会触发全量数据的处理：

1. 全局边界计算：为了将数据划分到有序的输出分区，Spark需要先扫描全量数据，统计排序键的分布，计算出每个输出分区的边界阈值（比如基于排序键的分位数）；
2. 实际排序shuffle：根据第一步得到的边界阈值，将数据重新shuffle到对应的分区，同时在分区内完成排序。

这两次全量扫描都会回溯执行sort()之前的所有转换逻辑（包括你代码中的mapPartitions），所以才会出现mapPartitions被执行两次的情况。你提到确认不是采样导致的全量计算，这也符合逻辑——当Spark判断数据量较小，或者需要精确的分区边界时，会选择全量扫描而非采样来计算阈值。

为什么cache/persist能解决重复执行问题 #

当你在sort()前对ReflexivLongSubKmerDS调用persist()或cache()后，第一次扫描生成的数据会被持久化到内存或磁盘中。第二次扫描时，Spark会直接读取持久化的结果，无需重新执行前面的mapPartitions等转换操作，因此mapPartitions只会执行一次。

总结 #

这种两次扫描的逻辑是Spark全局排序的设计使然，不是bug。如果你的转换操作计算成本较高，提前持久化是避免重复计算、提升性能的合理方案。

内容的提问来源于stack exchange，提问作者LIREN HUANG