如何在Scala中基于共同列合并Spark DataFrame序列？

针对你这种持有多个带共同列的DataFrame、需要合并的场景，用Scala集合的reduce方法是最贴合函数式风格的优雅方案，完全不需要foreach或者手动写递归（当然递归也能实现，后面也会补充）。

核心思路 #

利用Seq的reduce方法，从第一个DataFrame开始，依次将当前合并结果与下一个DataFrame按照共同列做连接，最终得到所有列合并后的目标DataFrame。

完整实现代码 #

结合你给出的示例，完整可运行的代码如下：

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

val conf = new SparkConf().setMaster("local[*]")
val spark = SparkSession.builder()
  .appName("Feature Generator tests")
  .config(conf)
  .config("spark.sql.warehouse.dir", "/tmp/hive")
  .enableHiveSupport()
  .getOrCreate()

// 创建基础DataFrame
val df = spark.range(0, 1000).toDF().withColumn("Product", concat(lit("product"), col("id")))
// 生成包含多个DataFrame的Seq
val dataFrames = Seq(1,2,3).map(s => df.withColumn("_" + s.toString, lit(s)))

// 定义用于连接的共同列
val joinKeys = Seq("id", "Product")

// 用reduce合并所有DataFrame
val mergedDF = dataFrames.reduce((df1, df2) => df1.join(df2, joinKeys, "inner"))

// 验证结果列
mergedDF.columns // 输出: Array(id, Product, _1, _2, _3)

为什么选reduce？ #

• 纯函数式实现：不需要定义可变变量（比如用foreach时可能需要var来存中间结果），代码更简洁安全。
• 语义清晰：reduce的作用就是将集合元素依次合并，完美匹配我们“逐个连接DataFrame”的需求。
• 适配任意数量：不管Seq里有1个还是N个DataFrame，都能自动处理（当只有1个元素时直接返回该DataFrame）。

备选：递归实现 #

如果你更倾向于递归的写法，也可以写一个轻量的递归函数：

import org.apache.spark.sql.DataFrame

def mergeDFs(dfs: Seq[DataFrame], joinKeys: Seq[String]): DataFrame = dfs match {
  case head :: Nil => head // 只剩一个DataFrame时直接返回
  case head :: tail => head.join(mergeDFs(tail, joinKeys), joinKeys, "inner") // 递归合并剩余的DataFrame
  case Nil => throw new IllegalArgumentException("不能传入空的DataFrame序列")
}

// 调用递归函数
val mergedDF = mergeDFs(dataFrames, joinKeys)

注意事项 #

• 连接类型：示例中用的是inner连接，如果你需要保留所有行（比如部分DataFrame可能缺失部分共同列的值），可以换成outer或者left连接，根据业务需求调整。
• 列名冲突：确保各个DataFrame的额外列没有重名，否则连接后会出现重复列名问题（如果有这种情况，可以在连接前先重命名冲突列）。

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