Spark同名字段Dataset连接后映射为Case Class元组的实现方法

核心思路是把两个Dataset的字段分别封装成结构体（Struct），规避同名字段冲突，再将结构体映射到对应的Case Class，最终组合成目标元组。

方法一：连接前封装结构体 #

先将每个Dataset的所有字段打包为独立Struct，再执行连接操作，从根源避免字段冲突：

// 将dataset1的所有字段封装到名为ds1的Struct中
val ds1WithStruct = dataset1.select(struct(dataset1.columns.map(col): _*).as("ds1"))
// 将dataset2的所有字段封装到名为ds2的Struct中
val ds2WithStruct = dataset2.select(struct(dataset2.columns.map(col): _*).as("ds2"))

// 基于Struct内的key字段执行内连接
val joined = ds1WithStruct.join(
  ds2WithStruct,
  ds1WithStruct("ds1.key") === ds2WithStruct("ds2.key"),
  "inner"
)

// 直接将Struct映射到对应Case Class，组合成元组
val result = joined.as[(CaseClass1, CaseClass2)]

方法二：连接后封装结构体 #

如果已完成连接操作，可在连接结果中对两个Dataset的字段分别打包：

val joined = dataset1.as("ds1")
  .join(dataset2.as("ds2"), dataset1("key") === dataset2("key"), "inner")

// 提取ds1、ds2各自的所有字段
val ds1Cols = dataset1.columns.map(col("ds1." + _))
val ds2Cols = dataset2.columns.map(col("ds2." + _))

// 封装Struct并映射为元组类型
val result = joined
  .select(struct(ds1Cols: _*).as("ds1"), struct(ds2Cols: _*).as("ds2"))
  .as[(CaseClass1, CaseClass2)]

关键说明 #

• struct()函数可将多个字段打包为单个结构体列，同名字段会被隔离在不同Struct下，彻底解决冲突问题。
• 映射元组时，Spark会自动将ds1结构体匹配到CaseClass1、ds2结构体匹配到CaseClass2，需保证Case Class的字段名、类型与原Dataset完全一致。

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