为啥 sortBy() 不能在 Spark 中对数据进行均匀排序？

Posted 2023-04-15

【中文标题】为啥 sortBy() 不能在 Spark 中对数据进行均匀排序？

【英文标题】：Why sortBy() cannot sort the data evenly in Spark?为什么 sortBy() 不能在 Spark 中对数据进行均匀排序？

【发布时间】：2016-12-27 15:05:44

【问题描述】：

我正在编写一个 pyspark 脚本来读取一个大的二维数组，所以我尝试首先生成一个索引数组并使用 read 方法映射以读取相应的数组。例如，如果我有一个包含 10 行的数组，那么我希望这 10 行均匀分区，因为每个分区有 2 行。我用 sortBy() 尝试过这种方式：

rdd = sc.range(0, 10, 1).sortBy(lambda x: x, numPartitions = 5)
rdd.glom().collect()

但是，结果显示为：

[[0, 1, 2], [3, 4], [5, 6], [7, 8], [9]]

这表明 sortBy() 没有像我预期的那样工作，所以第一个分区有 3 个数字，而最后一个分区只有 1 个数字。当我用另一种读取方法映射每个分区时，分区的大小是不同的，有时会导致散乱。

我尝试了另一种 RDD 生成方式：

rdd = sc.parallelize(range(0, 10, 1), 5)
rdd.glom().collect()

它会返回我想要的结果。

[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]

谁能帮忙解释一下为什么使用 sortBy() 的第一个方法不能返回均匀排序的结果？

【参考方案1】：

因为它不是设计来的。在一般情况下，不可能对数据进行分区（包括范围分区）以实现相等大小的分区。请记住，根据分区程序的约定，特定值的所有记录都必须驻留在单个分区上。即使在可以实现均匀分布的情况下，确定准确的分区边界也会非常昂贵。

因为 Spark 采样数据的目标是获得近似统一大小的范围，这种行为对于典型的 Spark 应用程序来说已经足够好了。

SparkContext.parallelize 根本不使用分区器。相反，它根据特定输入的语义计算拆分，因此可以创建大小相等的拆分。

如果您有关于数据分布的先验知识，您始终可以设计自定义分区函数，从而产生所需的输出。例如：

import bisect
from functools import partial

partition_func = partial(bisect.bisect, [2, 4, 6, 8])

(sc.range(0, 10)
    .map(lambda x: (x, None))
    .repartitionAndSortWithinPartitions(5, partition_func)
    .keys())

对于 CPython 中相对较短（最多 1

(sc.range(0, 10, 1)
    .map(lambda x: (x, None))
    .partitionBy(10)
    .keys()
    .glom()
    .collect())

[[0], [1], [2], [3], [4], [5], [6], [7], [8], [9]]

但这只是一个实现细节（hash(x) 其中isinstance(x, int) 等于x）。