Spark如何通过Join合并两个DataFrame的同名列？

现有实现正确性说明 #

你的现有实现逻辑是正确的，完全可以达到预期的合并效果，缺点是当列数较多时需要重复编写大量冗余代码，维护成本高，且没有用到Spark API的动态表达式能力。

更高效的实现方案 #

可以通过动态构造查询表达式的方式批量处理所有待合并列，无需为每一列单独编写增删改逻辑，通用实现如下：

步骤1：定义公共参数

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

// 定义关联键，支持1-2个或更多关联键
val joinKeys = Seq("key")
// 自动提取所有非关联的待合并列，无需手动枚举
val mergeCols = myDF.columns.filterNot(joinKeys.contains).toSeq

场景1：合并规则为两列求和

和你第一个示例的逻辑完全等价：

val newDF = myDF.as("left")
  .join(dbDF.as("right"), joinKeys, "inner")
  .select(
    // 先保留所有关联键
    joinKeys.map(col) ++
    // 批量为每个待合并列生成求和表达式
    mergeCols.map(c => (col(s"left.$c") + col(s"right.$c")).as(c)) : _*
  )

场景2：合并规则为取非空值

和你第二个外连接取非空的逻辑完全等价：

val newDF = myDF.as("left")
  .join(dbDF.as("right"), joinKeys, "outer")
  .select(
    joinKeys.map(col) ++
    mergeCols.map(c => coalesce(col(s"left.$c"), col(s"right.$c")).as(c)) : _*
  )

扩展：支持自定义不同列的合并规则

如果不同列需要用不同的合并逻辑，可以通过规则映射实现：

// 自定义每列的合并规则
val mergeRuleMap = Map(
  "amount" -> (col("left.amount") + col("right.amount")),
  "status" -> coalesce(col("left.status"), col("right.status")),
  "update_time" -> greatest(col("left.update_time"), col("right.update_time"))
)

val newDF = myDF.as("left")
  .join(dbDF.as("right"), joinKeys, "outer")
  .select(
    joinKeys.map(col) ++
    mergeRuleMap.map { case (colName, expr) => expr.as(colName) }.toSeq : _*
  )

性能优化建议 #

如果其中一个DataFrame的数据量较小（小于10MB默认阈值），可以使用广播join提升性能：

import org.apache.spark.sql.functions.broadcast

val newDF = myDF.as("left")
  .join(broadcast(dbDF.as("right")), joinKeys, "inner")
  // 后续select逻辑和上述场景一致

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