Coalesce 可以增加 Spark DataFrame 的分区吗

Question

我想了解coalesce() 和repartition() 之间的区别。

如果我正确理解this answer，coalesce() 只能减少数据帧的分区数，如果我们尝试增加分区数，则分区数保持不变。

但是当我尝试执行下面的代码时，我观察到了两件事

1. 对于具有合并的分区数的 Dataframe 可以增加
2. 对于 Rdd，如果 shuffle = false 则分区数不能随着合并而增加。

这是否意味着通过合并数据帧分区可以增加？

对数据框应用合并

当我执行以下代码时：

val h1b1Df = spark.read.csv("/FileStore/tables/h1b_data.csv")
println("Original dataframe partitions = " + h1b1Df.rdd.getNumPartitions)

val coalescedDf = h1b1Df.coalesce(2)
println("Coalesced dataframe partitions = " + coalescedDf.rdd.getNumPartitions

val coalescedDf1 = coalescedDf.coalesce(6) 
println("Coalesced dataframe with increased partitions = " + coalescedDf1.rdd.getNumPartitions) 

我得到以下输出

Original dataframe partitions =  8
Coalesced dataframe partitions = 2
Coalesced dataframe with increased partitions = 6

对 RDD 应用合并

当我执行以下代码时：

val inpRdd = h1b1Df.rdd
println("Original rdd partitions = " + inpRdd.getNumPartitions)

val coalescedRdd = inpRdd.coalesce(4)
println("Coalesced rdd partitions = " + coalescedRdd.getNumPartitions)

val coalescedRdd1 = coalescedRdd.coalesce(6, false)
println("Coalesced rdd with increased partitions = " + coalescedRdd1.getNumPartitions)

我得到以下输出：

Original rdd partitions =  8
Coalesced rdd partitions = 4
Coalesced rdd with increased partitions = 4

Answer 1

Coalesce 可用于通过设置 shuffle=true 来增加分区，这等于重新分区。 当您使用合并 shuffle=false 来增加时，不会发生数据移动。所以一个分区的数据不能移动到另一个分区。而 reduce 只是合并最近的分区。

谢谢，

Answer 2

coalesce for dataframe 不能增加超过集群中核心总数的分区。

 val h1b1Df = spark.read.csv("/FileStore/tables/h1b_data.csv")
 h1b1Df.rdd.getNumPartitions        // prints 8

 val cloasedDf = h1b1Df.coalesce(21)  
 cloasedDf.rdd.getNumPartitions     // prints 8

 val cloasedDf1 = cloasedDf.coalesce(2) // prints 2
 cloasedDf1.rdd.getNumPartitions

 val cloasedDf2 = cloasedDf.coalesce(7) // prints 7
 cloasedDf2.rdd.getNumPartitions

Answer 3

如果您在同一个数据帧上应用多个合并而不在这些合并之间执行任何转换，Spark 将通过仅应用最新的合并来优化您的合并，给您的印象是您可以使用合并增加分区数量强>

说明

如果我们取下面的代码sn-p：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
    .master("local[8]") // use 8 threads
    .appName("test-app")
    .getOrCreate()

import spark.implicits._

val input = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).toDF("MyCol")
val simpleCoalesce = input.coalesce(4)
val doubleCoalesce = simpleCoalesce.coalesce(8)

println(doubleCoalesce.rdd.getNumPartitions)

它将打印8。为了更好地理解为什么我们会得到这个结果，我们在doubleCoalesce 上执行explain(true)：

doubleCoalesce.explain(true)

我们得到以下步骤来构建最终的spark执行计划来解决doubleCoalesce：

== Parsed Logical Plan ==
Repartition 8, false
+- Repartition 4, false
   +- Project [value#1 AS MyCol#4]
      +- LocalRelation [value#1]

...

== Physical Plan ==
Coalesce 8
+- LocalTableScan [MyCol#4]

我们可以看到，在解析的逻辑计划（即对代码 sn-p 的简单解析）和物理计划（将应用于生成最终数据帧的计划）之间，仅保留最新的合并，所以coalesce(4) 转换实际上从未应用过。因此，只应用了coalesce(8)，我们得到了一个包含 8 个分区的数据帧。

如果我们想应用coalesce(4)，我们需要在两个合并之间进行转换，例如一个复杂的select：

import org.apache.spark.sql.functions.col

val separatedCoalesce = simpleCoalesce
  .select((col("MyCol") + 0).as("MyCol"))
  .coalesce(8)

println(separatedCoalesce.rdd.getNumPartitions)

注意：简单的 select as .select(col("MyCol")) 将不起作用，因为 spark 将在优化阶段放弃选择

此代码打印4。如果我们看separatedCoalesce的实物图：

== Physical Plan ==
Coalesce 8
+- *(1) Project [(MyCol#4 + 0) AS MyCol#9]
   +- Coalesce 4
      +- LocalTableScan [MyCol#4]

我们可以看到，这里应用了coalesce(4)，所以最后我们得到一个只有4个分区的数据帧，尽管我们应用了coalesce(8)

结论

Spark 优化可能很棘手，会让您认为没有发生的事情发生了。所以请记住，spark 并不完全执行编写的代码，而是执行它的优化版本。