火花重新分区落入单个分区答案

【问题标题】：spark repartition fall into single partition火花重新分区落入单个分区
【发布时间】：2018-07-06 15:07:16
【问题描述】：

我正在学习 spark，当我在 pyspark shell 中使用以下表达式测试 repartition() 函数时，我观察到一个非常奇怪的结果：所有元素在 repartition() 函数之后落入同一个分区。在这里，我使用glom() 来了解rdd 内的分区。我期待repartition() 对元素进行洗牌并在分区之间随机分配它们。这只发生在我使用新的分区数

在我的测试过程中，如果我设置新的分区数 > 原始分区，也没有观察到洗牌。我在这里做错了吗？

In [1]: sc.parallelize(range(20), 8).glom().collect()
Out[1]:
[[0, 1],
 [2, 3],
 [4, 5],
 [6, 7, 8, 9],
 [10, 11],
 [12, 13],
 [14, 15],
 [16, 17, 18, 19]]

In [2]: sc.parallelize(range(20), 8).repartition(8).glom().collect()
Out[2]:
[[],
 [],
 [],
 [],
 [],
 [],
 [2, 3, 6, 7, 8, 9, 14, 15, 16, 17, 18, 19, 0, 1, 12, 13, 4, 5, 10, 11],
 []]

In [3]: sc.parallelize(range(20), 8).repartition(10).glom().collect()
Out[3]:
[[],
 [0, 1],
 [14, 15],
 [10, 11],
 [],
 [6, 7, 8, 9],
 [2, 3],
 [16, 17, 18, 19],
 [12, 13],
 [4, 5]]

我使用的是 spark 版本 2.1.1。

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

啊，我认为与底层分区器有关。我尝试了更大的数字，现在结果更有意义。

In [95]: [len(lst) for lst in sc.parallelize(range(1000), 8).glom().collect()]
Out[95]: [125, 125, 125, 125, 125, 125, 125, 125]

In [96]: [len(lst) for lst in sc.parallelize(range(1000), 8).repartition(10).glom().collect()]
Out[96]: [95, 95, 100, 105, 95, 95, 100, 105, 105, 105]

In [97]: [len(lst) for lst in sc.parallelize(range(1000), 8).repartition(5).glom().collect()]
Out[97]: [190, 195, 205, 210, 200]

【讨论】：

【解决方案2】：

恭喜！您刚刚重新发现 SPARK-21782 - 当 numPartitions 是 2 的幂时，重新分区会产生偏差：

目前，重新分区（shuffle-enabled coalesce）的算法如下：

对于每个初始分区索引，生成位置为 (new Random(index)).nextInt(numPartitions) 然后，对于初始分区索引中的第 k 个元素，将其放入新的分区位置 + k (modulo numPartitions)。

因此，基本上元素在 numPartitions 存储桶上的分布大致相等 - 从编号 position+1 的存储桶开始。

请注意，为每个初始分区索引创建一个新的 Random 实例，具有固定的种子索引，然后丢弃。因此，对于世界上任何 RDD 的每个索引，位置都是确定的。此外，当 bound 是 2 的幂时，nextInt(bound) 实现有一个特殊情况，这基本上是从初始种子中获取几个最高位，只有最小的加扰。

PySpark 让情况变得更糟，因为它 uses batched serializer 的默认批量大小等于 10，因此每个分区上的项目数量很少，所有项目都被洗牌到相同的输出。

好消息是already resolved in Spark 2.3 感谢Sergey Serebryakov。

【讨论】：