Spark是否识别DataFrame分区键？Parquet分区与Shuffle相关技术问询

咱先逐个拆解你遇到的问题，结合你给出的代码场景来解释：

问题1：Spark是否知晓df2已按numerocarte列分区？若知晓，窗口函数是否不会触发Shuffle？反之是否会触发？ #

Spark默认是不知道df2已经按numerocarte列分区的。原因很简单：你用repartition(42, col("numerocarte"))写入Parquet时，只是通过哈希分区的逻辑把相同哈希值的numerocarte数据放到了同一个物理分区文件里，但Parquet的元数据不会记录这个“按numerocarte分区”的逻辑信息，Spark的查询优化器也不会自动推断出这个分区规则。

所以当你执行窗口函数partitionBy(col("numerocarte"))时，Spark会默认认为数据是分散在各个分区的，为了保证同一个numerocarte的所有数据都能在同一个分区内完成聚合计算，就会触发Shuffle操作——这就是为什么你的分区数从42变成了默认的200个（这个数由spark.sql.shuffle.partitions配置控制）。如果Spark知道数据已经按该列分区，那窗口函数就不会触发Shuffle，直接在每个分区内处理即可。

问题2：若Spark不知晓，如何告知数据已按指定列分区？ #

有几种靠谱的方式可以让Spark明确知道数据的分区规则，避免不必要的Shuffle：

方法1：显式设置RDD分区器（推荐） #

因为DataFrame的分区信息其实是底层RDD的属性，我们可以手动给RDD指定基于numerocarte的哈希分区器，再转回DataFrame：

import org.apache.spark.HashPartitioner

// 将DataFrame转为RDD，用numerocarte作为分区key
val keyedRdd = df2.rdd.keyBy(row => row.getAs[String]("numerocarte"))
// 使用和之前一致的42个分区创建哈希分区器
val partitioner = new HashPartitioner(42)
// 重新分区后移除多余的key
val partitionedRdd = keyedRdd.partitionBy(partitioner).map(_._2)
// 转回DataFrame，保留原schema
val df3 = spark.createDataFrame(partitionedRdd, df2.schema)

这样处理后，Spark就能识别到df3的分区规则，后续执行窗口函数时就不会触发Shuffle了。

方法2：读取后重新repartition（简单但有多余Shuffle） #

如果你不介意一次额外的Shuffle操作，可以直接在读取后执行：

val df3 = df2.repartition(42, col("numerocarte"))

这会让Spark明确记录下数据按numerocarte分区的规则，后续操作就不会再Shuffle了。不过因为你的数据已经是分区好的，这个操作其实是多余的，所以优先选方法1。

方法3：开启自适应查询优化（AQE） #

如果你的Spark版本在2.3及以上，可以开启AQE：

spark.conf.set("spark.sql.adaptive.enabled", "true")

AQE会自动检测数据的分区情况，可能避免不必要的Shuffle，但这个不是百分百保证的，显式设置分区器更可靠。

问题3：如何查看DataFrame的分区键？ #

Spark DataFrame本身没有直接显示“分区键”的API，因为分区逻辑是底层RDD的属性，我们可以通过以下几种方式间接查看：

查看RDD的分区器 #

通过RDD的partitioner属性可以判断分区类型和数量：

df2.rdd.partitioner match {
  case Some(p: HashPartitioner) => println(s"哈希分区，分区数：${p.numPartitions}")
  case _ => println("无明确分区器或非哈希分区")
}

不过这个方法只能知道分区器类型，无法直接看到分区列，需要结合你的业务逻辑推断（比如你之前是按numerocarte分区的，那就能对应上）。

查看分区内的数据样本 #

可以查看每个分区里的numerocarte值，判断分区是否按该列划分：

// 收集每个分区内的distinct numerocarte值
val partitionKeys = df2.rdd.glom()
  .map(rows => rows.map(_.getAs[String]("numerocarte")).distinct)
  .collect()

// 打印每个分区的结果
partitionKeys.zipWithIndex.foreach { case (keys, idx) =>
  println(s"分区${idx}的numerocarte值：${keys.mkString(", ")}")
}

查看物理执行计划 #

执行df2.explain()，查看输出中的Exchange操作：如果没有Exchange说明当前没有Shuffle，结合你之前的写入逻辑可以推断分区情况，但同样无法直接看到分区列。

问题4：两张按同一列重分区的表进行Join时，Spark是否会利用该分区信息？ #

分两种情况：

• 如果两张表都被Spark明确识别为按同一列、同一分区器分区（比如都用方法1设置了相同的哈希分区器，或者都执行了repartition(n, col("numerocarte"))），那么Spark会使用分区连接（Partitioned Join），直接在每个对应的分区内完成Join操作，完全不需要触发Shuffle。
• 如果只是物理上分区但Spark不知道分区规则（比如你直接读取Parquet文件，没有显式设置分区器），那么Spark还是会触发Shuffle，先把两张表按Join列重新分区后再执行Join。

举个例子：如果你两张表都是用repartition(42, col("numerocarte"))处理过的，并且在Join前没有被其他操作打乱分区，Spark的优化器会自动识别到它们的分区一致，直接走分区连接，性能会提升很多。

内容的提问来源于stack exchange，提问作者astro_asz