如何将多个Case Class放入单个RDD/DataSet并高效分路径导出？

问题根源 #

你之前用父特质P转DataSet[P]失败的核心原因是：Spark的DataSet依赖具体、带固定Schema的类型，特质是抽象类型，Spark无法为其生成对应的编码器（Encoder），也就无法推断统一的Schema。此外，即便转换成功，partitionBy导出的会是合并Schema的文件，包含大量空值，不符合你要专属格式的需求。

最优简单方案：单次读取+分支处理 #

核心思路是只读取一次原始数据，通过过滤、字段选择拆分出对应类型的数据，转成目标Case Class后分别导出。既保证效率，又能得到各类型的专属格式文件。

方法1：用DataFrame+Case Class转换

import org.apache.spark.sql.{SparkSession, functions => F}
import org.apache.spark.sql.types.{IntegerType, StringType, BooleanType, StructType, StructField}

case class A(id: Int, age: Int, name: String)
case class B(id: Int, age: Int, married: Boolean)
case class C(id: Int, age: Int, gender: String)

val spark = SparkSession.builder().getOrCreate()

// 1. 读取原始数据（假设原始数据包含所有可能字段）
val rawDF = spark.read.schema(StructType(Seq(
  StructField("id", IntegerType),
  StructField("age", IntegerType),
  StructField("name", StringType, nullable = true),
  StructField("married", BooleanType, nullable = true),
  StructField("gender", StringType, nullable = true)
))).load("原始数据路径")

// 2. 拆分并导出各类型数据
// 导出A类型到指定路径
rawDF.filter(F.col("name").isNotNull)
  .select("id", "age", "name")
  .as[A]
  .write.save("hdfs://some/a")

// 导出B类型到指定路径
rawDF.filter(F.col("married").isNotNull)
  .select("id", "age", "married")
  .as[B]
  .write.save("hdfs://some/b")

// 导出C类型到指定路径
rawDF.filter(F.col("gender").isNotNull)
  .select("id", "age", "gender")
  .as[C]
  .write.save("hdfs://some/c")

方法2：用通用Case Class中转

如果习惯用强类型DataSet操作，可以先定义包含所有字段的临时Case Class，再拆分：

case class RawData(id: Int, age: Int, name: Option[String], married: Option[Boolean], gender: Option[String])

val rawDS = spark.read.load("原始数据路径").as[RawData]

// 导出A
rawDS.filter(_.name.isDefined)
  .map(r => A(r.id, r.age, r.name.get))
  .write.save("hdfs://some/a")

// 导出B
rawDS.filter(_.married.isDefined)
  .map(r => B(r.id, r.age, r.married.get))
  .write.save("hdfs://some/b")

// 导出C
rawDS.filter(_.gender.isDefined)
  .map(r => C(r.id, r.age, r.gender.get))
  .write.save("hdfs://some/c")

方案优势 #

• 效率高：原始数据仅读取、解析一次，后续过滤、转换均在分布式内存中完成，避免重复扫描数据
• 格式纯净：导出的文件完全对应A/B/C的专属Schema，无多余空值字段
• 实现简单：全部使用Spark原生API，无需自定义复杂方法或编码器

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