如何重分区Spark DataFrame为更小分区且不混同不同日期数据

首先明确前提：要把每个日期对应的单分区拆分为多个分区、同时保证同分区不混入其他日期数据，必须经过一次shuffle——原有同日期数据都在同一个分区内存储，要拆分到不同分区必然要做数据分发，不存在零shuffle的实现。下面的方案都只触发这一次必要shuffle，没有任何冗余开销，完全满足你的要求。

你之前两种写法不生效的原因非常直接：

• df.repartition(20, my_date_column)：Spark默认哈希分区器的逻辑是对分区键算哈希值，再对总分区数取模得到分区ID。由于my_date_column只有7个唯一值，哈希后最多映射到7个不同ID，剩下13个分区必然是空的。
• df.repartition(20, my_date_column, unique_id)：分区键变成「日期+唯一ID」的组合后，不同日期的组合键哈希取模完全可能命中同一个分区ID，自然会出现跨日期数据混杂。

方案1：按日期拆分后单独重分区（实现最简单，全语言通用） #

核心思路是先把每个日期的数据单独过滤出来，给每个日期的子数据集分配对应数量的分区，再把所有子数据集合并，全程只有一次shuffle。
比如7个日期要凑20个总分区，可以根据每个日期的数据量灵活分配分区数：数据量大的日期分3个分区，数据量小的分2个，加起来刚好20就行。如果每天数据量均匀，直接给每个日期分3个分区，总共有21个分区，多1个分区对性能几乎没有影响。
PySpark示例代码：

# 收集所有待处理的日期值
date_list = [row.my_date_column for row in df.select("my_date_column").distinct().collect()]
# 给每个日期分配分区数，示例为前6个日期分3个，最后1个分2个，总和3*6+2=20
partition_alloc = [3]*6 + [2]

result_df = None
for date, part_num in zip(date_list, partition_alloc):
    # 过滤出单个日期的数据，重分区到指定数量（因为只有一个日期，怎么分都不会混）
    sub_df = df.filter(f"my_date_column = '{date}'").repartition(part_num)
    result_df = sub_df if result_df is None else result_df.unionAll(sub_df)

这个方案的优势是逻辑完全可控，没有黑盒：

• 每个日期的数据只会在自己重分区后的几个分区内，unionAll不会打乱已有分区分布，绝对不会出现跨日期混存
• 仅在每个子数据集调用repartition时触发一次shuffle，没有额外冗余步骤
• 不会产生空分区，总分区数完全符合预期

方案2：自定义分区器（性能最优，适合Scala/Java场景） #

如果用Scala/Java开发，可以通过自定义分区器在一次shuffle中直接完成分区逻辑，不需要拆分-合并步骤，性能比方案1略好。
分区器逻辑非常简单：

1. 提前收集所有唯一日期，给每个日期分配一段连续、不重叠的分区ID区间，覆盖所有0到总分区数-1的ID
2. 计算分区时，先根据日期找到对应的分区区间，再根据唯一ID的哈希值在区间内做二次取模，得到最终分区ID
   Scala示例代码：

import org.apache.spark.Partitioner
// 收集所有日期值
val dates = df.select("my_date_column").distinct().as[String].collect()
// 给每个日期分配连续分区区间，示例每个日期分3个分区，总分区数21，可按需调整
val dateToRange = dates.zipWithIndex.map { case (date, idx) =>
  val start = idx * 3
  val end = start + 2
  date -> (start, end)
}.toMap
val totalParts = dates.length * 3

class DateBasedPartitioner extends Partitioner {
  override def numPartitions: Int = totalParts
  override def getPartition(key: Any): Int = {
    val (date, uniqueId) = key.asInstanceOf[(String, String)]
    val (start, end) = dateToRange(date)
    val rangeLen = end - start + 1
    // 算哈希时处理负数情况
    val subPos = (uniqueId.hashCode % rangeLen + rangeLen) % rangeLen
    start + subPos
  }
}

// 转PairRDD应用自定义分区器，再转回DataFrame
val resultRdd = df.rdd.map(row => (
  (row.getAs[String]("my_date_column"), row.getAs[String]("unique_id")),
  row
)).partitionBy(new DateBasedPartitioner()).values
val resultDf = spark.createDataFrame(resultRdd, df.schema)

这个方案同样完全满足要求，且shuffle的开销更低，但Python场景下自定义分区器需要做跨进程序列化，复杂度较高，优先用方案1即可。

避坑提醒 #

不要直接用默认的repartitionByRange实现这个需求。默认的范围分区是通过采样自动生成分区边界，很容易出现边界偏移，导致不同日期的数据落到同一个分区。除非你手动精准给定每个分区的切分边界，否则不推荐使用。

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