Coalesce可以增加Spark DataFrame的分区吗

Question

我试图理解coalesce（）和re分区（）之间的区别。

如果我正确理解了这个答案，coalesce（）只能减少dataframe的分区数量，如果我们尝试增加分区数量，那么分区数量将保持不变。

但当我试图执行下面的代码时，我发现了两件事

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

当我执行以下代码时：

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

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

如果我们使用以下代码片段：

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。为了更好地理解我们为什么会得到这个结果，我们在双重合并上执行解释（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]

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

如果我们想要应用合并（4），我们需要在两个合并之间执行转换，例如复杂的选择：

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

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

println(separatedCoalesce.rdd.getNumPartitions)

注意：简单选择为。选择（col（“MyCol”）将不起作用，因为spark将在优化阶段放弃选择

此代码打印4。如果我们看一下分离合并的物理计划：

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

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

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

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

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

谢谢，