求通用Spark Scala方法：无需显式列名合并同结构DataFrame（支持列表连接键）

下面是一个满足需求的Scala方法，专门用于合并结构完全相同的两个DataFrame，支持通过列表指定连接键，无需手动枚举所有列，具备通用性：

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StructType

def mergeDataFrames(dfA: DataFrame, dfB: DataFrame, joinKeys: Seq[String]): DataFrame = {
  // 校验两个DataFrame的结构是否完全一致
  require(dfA.schema.equals(dfB.schema), "两个DataFrame的结构必须完全相同")
  
  // 获取所有非连接键的列名
  val nonJoinColumns = dfA.columns.filterNot(joinKeys.contains)
  
  // 执行全外连接
  val joinedDf = dfA.join(dfB, joinKeys, "full_outer")
  
  // 处理非连接键列：合并两个DataFrame中的值，优先取dfA的非空值，若无则取dfB的
  val mergedColumns = joinKeys.map(col) ++ nonJoinColumns.map(c => 
    coalesce(dfA(c), dfB(c)).alias(c)
  )
  
  // 生成最终合并后的DataFrame
  joinedDf.select(mergedColumns: _*)
}

关键说明 #

• 结构校验：通过require语句确保输入的两个DataFrame schema完全匹配，避免后续处理出错
• 连接逻辑：使用full_outer连接保留两个DataFrame的所有记录，基于指定的连接键匹配
• 列合并规则：对于非连接键列，用coalesce函数优先保留第一个DataFrame的非空值，若第一个为空则取第二个的；如果需要优先保留第二个DataFrame的值，只需调换coalesce的参数顺序即可
• 通用性：自动遍历所有非连接键列，无需手动写出每一列的处理逻辑，适配任意结构的同schema DataFrame

示例用法 #

假设我们有两个结构相同的用户数据DataFrame：

// 初始化SparkSession
val spark = SparkSession.builder().appName("MergeDFExample").master("local[*]").getOrCreate()
import spark.implicits._

// 创建测试DataFrame A
val dfA = Seq(
  (1, "Alice", 28, "New York"),
  (2, "Bob", 32, "London")
).toDF("id", "name", "age", "city")

// 创建测试DataFrame B
val dfB = Seq(
  (2, "Bob", 33, "London"), // age字段更新
  (3, "Charlie", 25, "Paris") // 新增记录
).toDF("id", "name", "age", "city")

// 调用合并方法，指定"id"和"name"为连接键
val mergedDf = mergeDataFrames(dfA, dfB, Seq("id", "name"))

// 查看结果
mergedDf.show()

执行后输出结果：

+---+-------+---+--------+
| id|   name|age|    city|
+---+-------+---+--------+
|  1|  Alice| 28|New York|
|  2|    Bob| 32|  London| // 优先保留dfA的age值，若要取dfB的33，调换coalesce参数即可
|  3|Charlie| 25|   Paris|
+---+-------+---+--------+

内容的提问来源于stack exchange，提问作者K.T.A