火花的合并功能是否尝试创建统一大小的分区？

Question

我想平衡 Spark 中 rdds/dataframes 的分区大小，以摆脱拖慢我工作速度的落后任务。我可以使用repartition(n_partition) 来做到这一点，它会创建大小相当一致的分区。但是，这涉及昂贵的洗牌。

我知道coalesce(n_desired_partitions) 是一种更便宜的替代方案，它避免了洗牌，而是在同一个执行程序上合并分区。但是，我不清楚这个函数是尝试创建大致统一大小的分区，还是简单地合并输入分区而不考虑它们的大小。

例如，假设以下我们在三个分区中具有 [1,12] 范围内的整数的 Rdd，如下所示：[(1,2,3,4,5,6,7,8),(9,10),(11,12)]。假设这些都在同一个执行程序上。

现在我打电话给rdd.coalesce(2)。权力合并的算法是否知道合并两个小分区（因为它们更小并且我们想要平衡的分区大小），而不是仅仅合并两个任意分区？

在别处讨论这个话题

根据this presentation（跳至 7:27），Netflix 大数据团队需要实现一个自定义的合并函数来平衡分区大小。另见SPARK-14042。

为什么这个问题不重复

关于分区和合并之间的区别有一个更普遍的问题here，但没有人解释支持合并的算法是否尝试平衡分区大小。

Answer 1

所以实际上重新分区没什么，它的定义如下所示

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
    coalesce(numPartitions, shuffle = true)
}

所以它只是简单地与 shuffle 合并，但是当调用 coalesce 时，它​​的 shuffle 将默认为 false，因此它不会对数据进行shuffle，直到不需要它。

例如，您有 2 个集群节点，每个节点有 2 个分区，现在您调用 rdd.coalesce(2)，这样它将合并节点的本地分区，或者如果您调用 coalesce(1)，那么它将需要 shuffle，因为其他 2 个分区将在另一个节点上，因此在您的情况下，它可能会加入本地节点分区，并且该节点的分区数量较少，因此您的分区大小不统一。

好的，根据您对问题的编辑，我也尝试执行以下操作

    val data = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10,11,12))
    data.getNumPartitions
    res2: Int = 4
   data.mapPartitionsWithIndex{case (a,b)=>println("partitionssss"+a);b.map(y=>println("dataaaaaaaaaaaa"+y))}.count

上述代码的输出将是

现在我将 4 分区合并为 2 并在该 rdd 上运行相同的代码，以检查如何优化 spark 合并数据，以便输出将是

现在您可以很容易地看到，即使在合并之前，spark 也将数据平均分配到 6-6 两个分区，所有分区中的元素数量并不相同。

 val coal=data.coalesce(2)
 coal.getNumPartitions
res4: Int = 2
coal.mapPartitionsWithIndex{case (a,b)=>println("partitionssss"+a);b.map(y=>println("dataaaaaaaaaaaa"+y))}.count