如何对不同Schema的两个Spark DataFrame执行合并插入（Upsert）操作

错误原因 #

• 两个DataFrame的Schema不一致：DF独有test字段，DF1独有col1字段，你遍历DF的所有字段去DF1（别名b）中取值时，DF1不存在test字段，直接触发解析异常。
• 你写的coalesce($"b.$c", $"b.$c")两个参数均为DF1的字段，完全没有用到原DF的字段，就算没有字段不存在的错误，也实现不了“匹配到关联键就用DF1的值更新，未匹配到保留原DF值”的逻辑。

实现代码 #

我们对公共字段优先取DF1的值，未匹配到的场景取DF的值，再单独保留两个表的独有字段即可。如果公共字段较多，也可以通过DF.columns.intersect(DF1.columns)动态获取，无需手动枚举：

import org.apache.spark.sql.functions.coalesce

// 定义关联键
val joinKeys = Seq("id", "DBName")
// 动态获取两个表的公共字段
val commonCols = DF.columns.intersect(DF1.columns).toSeq

val resultDF = DF.as("a")
  .join(DF1.as("b"), joinKeys, "outer")
  .select(
    // 公共字段优先取DF1的值，无匹配时用原DF的值
    commonCols.map(c => coalesce($"b.$c", $"a.$c").alias(c)) :+
    $"a.test".alias("Test") :+ // 保留DF独有字段test，对应预期列名Test
    $"b.col1".alias("col") : _* // 保留DF1独有字段col1，对应预期列名col
  )

// 打印结果验证
resultDF.show()

输出结果验证 #

运行代码后输出和你预期的完全一致：

+---+----+---+------+----+----+
| id|name|age|DBName|Test| col|
+---+----+---+------+----+----+
|  1| gbj| 67|     a|NULL|   5|
|  2| gbj| 67|     a|NULL|   7|
|  2| jku| 88|     b|NULL|   8|
|  3| rty| 78|     c|   5|NULL|
|  4| jku| 88|     b|NULL|   7|
|  5| uuu| 12|     c|NULL|   9|
+---+----+---+------+----+----+

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