Spark：repartition和repartitionByRange有什么区别？

我认为最好通过一些实验来研究差异。

测试数据帧

对于这个实验，我使用了以下两个数据帧（我在 Scala 中展示了代码，但概念与 Python API 相同）：

// Dataframe with one column "value" containing the values ranging from 0 to 1000000
val df = Seq(0 to 1000000: _*).toDF("value")

// Dataframe with one column "value" containing 1000000 the number 0 in addition to the numbers 5000, 10000 and 100000
val df2 = Seq((0 to 1000000).map(_ => 0) :+ 5000 :+ 10000 :+ 100000: _*).toDF("value")

理论

• repartition应用HashPartitioner何时提供一列或多列，以及RoundRobinPartitioner在提供的分区数量上均匀分布数据。如果提供了一列（或更多），这些值将被散列并用于通过计算类似的东西来确定分区号partition = hash(columns) % numberOfPartitions。

• repartitionByRange将根据列值的范围对数据进行分区。这通常用于连续（非离散）值，例如任何类型的数字。请注意，由于性能原因，此方法使用采样来估计范围。因此，输出可能不一致，因为采样可能返回不同的值。样本大小可以由 config 控制spark.sql.execution.rangeExchange.sampleSizePerPartition。

还值得一提的是，对于这两种方法，如果没有numPartitions给出，默认情况下它会将 Dataframe 数据分区到spark.sql.shuffle.partitions您的 Spark 会话中配置，并且可以通过自适应查询执行（自 Spark 3.x 起可用）合并。

测试设置

基于给定的 Testdata 我总是应用相同的代码：

val testDf = df
// here I will insert the partition logic
    .withColumn("partition", spark_partition_id()) // applying SQL built-in function to determin actual partition
    .groupBy(col("partition"))
    .agg(
      count(col("value")).as("count"),
      min(col("value")).as("min_value"),
      max(col("value")).as("max_value"))
    .orderBy(col("partition"))

testDf.show(false)

检测结果

df.repartition(4, col("value"))

正如预期的那样，我们得到了 4 个分区，因为 的值df范围从 0 到 1000000，我们看到它们的散列值将产生一个分布良好的 Dataframe。

+---------+------+---------+---------+
|partition|count |min_value|max_value|
+---------+------+---------+---------+
|0        |249911|12       |1000000  |
|1        |250076|6        |999994   |
|2        |250334|2        |999999   |
|3        |249680|0        |999998   |
+---------+------+---------+---------+

df.repartitionByRange(4, col("value"))

同样在这种情况下，我们得到 4 个分区，但这次最小值和最大值清楚地显示了分区内的值范围。它几乎均匀分布，每个分区有 250000 个值。

+---------+------+---------+---------+
|partition|count |min_value|max_value|
+---------+------+---------+---------+
|0        |244803|0        |244802   |
|1        |255376|244803   |500178   |
|2        |249777|500179   |749955   |
|3        |250045|749956   |1000000  |
+---------+------+---------+---------+

df2.repartition(4, col("value"))

现在，我们正在使用另一个 Dataframe df2。这里，散列算法对只有 0、5000、10000 或 100000 的值进行散列。当然，值 0 的散列将始终相同，因此所有零最终都在同一个分区中（在这种情况下，分区 3 ）。其他两个分区只包含一个值。

+---------+-------+---------+---------+
|partition|count  |min_value|max_value|
+---------+-------+---------+---------+
|0        |1      |100000   |100000   |
|1        |1      |10000    |10000    |
|2        |1      |5000     |5000     |
|3        |1000001|0        |0        |
+---------+-------+---------+---------+

df2.repartition(4)

如果不使用“value”列的内容，该repartition方法将在 RoundRobin 的基础上分发消息。所有分区的数据量几乎相同。

+---------+------+---------+---------+
|partition|count |min_value|max_value|
+---------+------+---------+---------+
|0        |250002|0        |5000     |
|1        |250002|0        |10000    |
|2        |249998|0        |100000   |
|3        |250002|0        |0        |
+---------+------+---------+---------+

df2.repartitionByRange(4, col("value"))

这种情况表明数据帧df2没有很好地定义用于按范围重新分区，因为几乎所有值都是 0。因此，我们甚至最终只有两个分区，而分区 0 包含所有零。

+---------+-------+---------+---------+
|partition|count  |min_value|max_value|
+---------+-------+---------+---------+
|0        |1000001|0        |0        |
|1        |3      |5000     |100000   |
+---------+-------+---------+---------+