Spark DataFrame写入HDFS时排序失效问题求助（单文件需求）

这个问题我之前也碰到过，核心是Spark的执行计划顺序和partitionBy的shuffle特性导致的，我来给你拆解一下原因和可行的解决方案：

为什么你的原代码会失效？ #

你的代码顺序是coalesce(1).orderBy(desc("key")).drop(...).write.partitionBy(...)，执行逻辑如下：

1. 先把所有数据合并到1个分区；
2. 对整个数据集按key降序排序；
3. 但执行write.partitionBy("date")时，Spark会触发shuffle操作，把单分区的数据按date重新分配到不同的分区（每个date对应一个分区），这个shuffle过程会彻底打乱之前的全局排序，导致最终每个date目录下的文件内容乱序。

而如果不用coalesce(1)，orderBy会生成多个排序后的分区，写入时每个分区对应一个part文件，所以排序是对的，但文件数量多——这就是你碰到的矛盾点。

可行解决方案 #

方法一：分区内排序+单分区输出（适合大数据量） #

我们需要让相同date的数据在同一个分区内完成排序，这样写入时每个date的分区直接输出为单个文件，不会打乱顺序。同时注意你的期望输出是key升序，原代码的desc("key")是降序，需要调整：

import org.apache.spark.sql.functions._

datadf
  // 先按date分区，确保相同date的数据进入同一个分区
  .repartition(col("date"))
  // 在每个分区内按key升序排序（date字段用于保证分区内只处理同日期数据）
  .orderBy(col("date"), asc("key"))
  .drop(col("key"))
  .write
  .mode("overwrite")
  // 按date生成输出分区目录
  .partitionBy("date")
  // 强制每个输出分区只生成1个文件
  .option("spark.sql.shuffle.partitions", "1")
  .text("hdfs://path/")

如果不想修改全局shuffle分区数，你可以先统计date的唯一值数量，比如有2个不同日期，就用.repartition(2)替代.repartition(col("date"))，效果是一样的。

方法二：分组聚合排序（适合小数据量） #

如果你的数据量不大，可以用groupBy+sort_array实现分组内排序，这种方式不需要依赖分区操作，逻辑更直观：

import org.apache.spark.sql.functions._

datadf
  // 按date分组
  .groupBy("date")
  // 收集每个分组的key和value为结构体列表
  .agg(collect_list(struct("key", "value")).alias("data"))
  // 对每个分组的结构体列表按key升序排序
  .withColumn("sorted_data", sort_array(col("data"), asc = true))
  // 展开排序后的value列表为行
  .withColumn("value", explode(col("sorted_data.value")))
  // 清理无用字段
  .drop("data", "sorted_data", "key")
  // 合并为单个分区（保证每个date目录下只有1个文件）
  .coalesce(1)
  .write
  .mode("overwrite")
  .partitionBy("date")
  .text("hdfs://path/")

⚠️ 注意：这个方法不适合大数据量，因为collect_list会把每个date的所有数据加载到Executor内存中，数据量过大时会导致OOM。

验证结果 #

用上面的方法执行后，每个date目录下的part文件内容会和你的期望输出完全一致：

• /20180701/part-xxxxxxx.txt 按key升序排列a1到a10
• /20180702/part-xxxxxxx.txt 按key升序排列a11到a19

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