在 Spark 中使用 reduceByKey 时有有效的分区方法吗？

Question

当我使用reduceByKey 或aggregateByKey 时，我遇到了分区问题。

例如）reduceBykey(_+_).map(code)

特别是如果输入数据有偏差，使用上述方法时，分区问题会变得更加严重。

因此，作为解决方案，我使用repartition 方法。

例如，http://dev.sortable.com/spark-repartition/ 类似。

这对分区分布有好处，但是repartition也很贵。

有没有办法明智地解决分区问题？

Answer 1

你必须区分两个不同的问题：

数据偏斜

如果数据分布高度偏斜（让我们假设只有一个唯一键的最坏情况），那么根据定义，输出将偏斜，更改分区器无济于事。

有一些技术可以用来部分解决问题，但整体分区并不是这里的核心问题。

分区偏差

即使数据是均匀分布的，选择不当的分区函数也会导致数据分布偏斜。例如：

val rdd = sc.parallelize(Seq((5, None), (10, None), (15, None), (20, None)), 5)
rdd
  .partitionBy(new org.apache.spark.HashPartitioner(5))
  .glom.map(_.size).collect

Array[Int] = Array(4, 0, 0, 0, 0)

正如您所看到的，尽管密钥分布没有倾斜，但倾斜是由 hashCode 的数据规律和不良属性引起的。

如果这样选择不同的Partitioner:

rdd
  .partitionBy(new org.apache.spark.RangePartitioner(5, rdd))
  .glom.map(_.size).collect

Array[Int] = Array(1, 1, 1, 1, 0)

或调整现有的属性：

rdd
  .partitionBy(new org.apache.spark.HashPartitioner(7))
  .glom.map(_.size).collect

Array[Int] = Array(0, 1, 0, 1, 0, 1, 1)

可以解决问题。

Answer 2

你是对的，

重新分区运行起来确实很昂贵。由于洗牌和其他小步骤。按照您的示例创建一个示例，如下所示：

rdd.map(x => (x, x * x)).repartition(8).reduceByKey(_+_)

在此处查看 DAG：

这一步将在 DAG 上创建一个 map、一个 repartition 和一个 reduce。

但是，如果您使用reduceByKey 中的重新分区，则可以“免费”进行重新分区。

Repratition 的主要部分是 Shuffle，reduceByKey 的主要部分也是 shuffle。你可以看到在 Scala lib 中，reduceByKey 有一个 numPartition parameter。

因此您可以为此更改代码：

rdd.map(x => (x, x * x)).reduceByKey(_+_, 8)

您可以在reduceByKey 中看到与重新分区相同的代码，它要快得多。因为你少洗牌了。