如何使用Java用另一Spark DataFrame的行替换匹配ID的行

在Spark里，DataFrame是不可变的，没法直接修改原DataFrame的行，正确思路是通过组合现有DataFrame生成新的结果集，下面给两种实用方法：

方法一：过滤不匹配行 + 合并更新行 #

核心逻辑是先移除df1中需要被替换的行，再把df2的更新行合并进去：

1. 用left_anti join筛选出df1中ID不在df2里的行（这种方式比手动过滤ID集合更高效，适合大数据量场景）
2. 用unionByName合并过滤后的df1和df2（保证列名匹配，避免列顺序问题）

Python代码示例 #

# 假设已初始化SparkSession并创建df1、df2
# 步骤1：过滤df1中无需替换的行
df1_filtered = df1.join(df2, on="ID", how="left_anti")

# 步骤2：合并过滤后的df1与df2的更新行
result_df = df1_filtered.unionByName(df2)

# 查看结果
result_df.show()

执行结果 #

+---+-----+---+-----+
| ID|    B|  C|    D|
+---+-----+---+-----+
|  2|  2.0|2.0|  2.0|
|  3|  3.0|3.0|  3.0|
|  4|  4.0|4.0|  4.0|
|  1|100.0|1.0|100.0|
+---+-----+---+-----+

方法二：外连接+Coalesce优先取更新值 #

如果df2仅更新部分列，这种方法更灵活：

1. 以ID为键做全外连接
2. 对每个列，优先使用df2的更新值，df2没有的话保留df1的原值
3. 选择目标列生成最终结果

Python代码示例 #

from pyspark.sql.functions import coalesce

# 全外连接df1和df2，给同名列加后缀区分
joined_df = df1.join(df2, on="ID", how="outer", suffixes=("_df1", "_df2"))

# 对每一列取优先值，生成结果集
result_df = joined_df.select(
    "ID",
    coalesce(joined_df.B_df2, joined_df.B_df1).alias("B"),
    coalesce(joined_df.C_df2, joined_df.C_df1).alias("C"),
    coalesce(joined_df.D_df2, joined_df.D_df1).alias("D")
)

# 查看结果
result_df.show()

说明 #

这种方法适配df2仅更新部分列的场景，比如若df2只有ID和B列，依然能保留df1的C、D列原值，同时替换B列的匹配行数据。

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