无Shuffle的Spark作业中如何强制启用AQE合并小分区？

一、低成本触发Shuffle以启用AQE合并的方案 #

你提到的「与1行DataFrame Join」的方式存在隐患：如果关联键是固定值，所有数据会被shuffle到同一个分区，反而会产生超大分区，违背合并小分区的初衷。更优的低成本方案如下：

推荐方案：使用repartition()无参调用 #

直接执行：

val mergedDf = df.repartition()

• 触发轻量Shuffle：基于行哈希值将数据均匀分配到spark.sql.shuffle.partitions（默认200）个分区
• 配合AQE自动合并：开启spark.sql.adaptive.enabled和spark.sql.adaptive.coalescePartitions.enabled后，AQE会自动将小分区合并到spark.sql.adaptive.advisoryPartitionSizeInBytes（默认64MB）指定的大小
• 优势：代码简洁，无额外计算逻辑，Shuffle成本极低，且数据分布均匀，避免单分区过载

替代方案：带随机键的Join（不推荐，仅作参考） #

如果因特殊场景无法使用repartition()，可以用随机键实现均匀Shuffle：

val dummyDf = spark.range(spark.sql.shuffle.partitions).selectExpr("id as rand_key")
val mergedDf = df.withColumn("rand_key", hash(lit(1)) % spark.sql.shuffle.partitions)
  .join(dummyDf, "rand_key")
  .drop("rand_key")

但此方案比repartition()多了列操作和关联逻辑，成本略高，优先级低于前者。

二、无需AQE的替代方案 #

如果不想依赖AQE，可直接通过以下方式合并小分区：

• coalesce(numPartitions)（无Shuffle）：
  手动计算目标分区数（例如math.ceil(totalDataSize / 64MB).toInt），执行：

  val mergedDf = df.coalesce(targetNumPartitions)
  

  优势：无Shuffle开销，速度快；缺点：仅合并相邻分区，若小分区分散，合并效果可能不佳，且需提前预估数据总大小。

• 手动指定repartition(numPartitions)：
  直接设置目标分区数，触发Shuffle并均匀分配数据：

  val mergedDf = df.repartition(targetNumPartitions)
  

  优势：数据分布均匀，合并效果稳定；缺点：需手动预估合适的分区数，灵活性不如AQE。

• 调整读取阶段参数：
  从根源减少小分区生成，例如：

  • 调整spark.sql.files.maxPartitionBytes（默认128MB），让Spark读取时自动合并小文件为指定大小的分区
  • 针对Parquet数据源，开启spark.sql.parquet.readPartitioning，优化读取时的分区策略
    此方案无需后续合并操作，是最理想的无额外成本方案。

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