Spark Java中如何基于Dataset实现多排序结果分文件输出？

没问题，我来帮你搞定这个需求！用Spark Dataset实现分组排序并输出到独立文件其实很简单，下面是完整的步骤和代码示例，我会尽量讲清楚细节：

1. 定义强类型数据结构 #

因为Dataset是强类型的，我们先创建一个case class来映射你的数据，这样后续操作更清晰也更安全：

case class CityRecord(city: String, code: String)

2. 读取并预处理输入数据 #

注意你输入里的城市名有大小写差异（比如City1和city1），如果不统一会被当成不同分组，所以第一步要把城市名转成统一大小写（比如全小写）：

import org.apache.spark.sql.SparkSession

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("CitySortSplit")
  .master("local[*]") // 本地调试用，生产环境记得去掉这个配置
  .getOrCreate()

// 导入隐式转换，让我们能把RDD转成Dataset
import spark.implicits._

// 读取文本输入，每行按空格分割，统一城市名大小写
val inputDS = spark.read.textFile("/path/to/your/input.txt")
  .map(line => {
    val parts = line.split("\\s+") // 处理可能的多个空格分隔
    CityRecord(parts(0).toLowerCase(), parts(1))
  })

3. 排序+分组输出 #

接下来是核心操作：先按城市名、再按code排序，然后按城市名分区输出，这样每个城市的数据会自动存到独立的目录（对应你要的独立文件）：

inputDS
  // 先按city排序，再按code排序，确保组内顺序正确
  .orderBy($"city", $"code")
  // 按city字段分区，每个分区对应一个城市的数据
  .write
  .partitionBy("city")
  .mode("overwrite") // 如果输出目录已存在就覆盖
  .text("/path/to/your/output")

4. 优化输出文件数量（可选） #

默认情况下Spark可能会为每个分区生成多个小文件，如果希望每个城市只生成一个文件，可以在输出前对每个分区做合并：

inputDS
  .groupByKey(_.city)
  .mapGroups { (city, records) =>
    // 对每个城市的记录单独按code排序
    (city, records.toList.sortBy(_.code))
  }
  .flatMap { case (city, sortedList) => sortedList }
  .coalesce(1) // 合并成一个文件（数据量大时不建议，会影响性能）
  .write
  .partitionBy("city")
  .mode("overwrite")
  .text("/path/to/your/output")

输出结果说明 #

执行完后，你的输出目录下会生成两个子目录：city=city1和city=city2，每个目录里的文件就是对应城市的排序后数据，和你预期的File1、File2完全一致。如果想直接生成File1、File2这样的文件名，你可以自己用foreachPartition手动写入文件，但Spark官方更推荐用partitionBy的方式，因为它更符合分布式计算的设计，性能也更好。

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