PySpark中如何基于跨DataFrame匹配条件更新全列值？

需求分析 #

当df_2的ADDRESS_CODE值存在于df_1的ZIP_CODE列中时，将df_1所有行的UPDATED_MESSAGE列填充为df_2的CODE值（即INDIA_WON）。

实现方式一：先验证存在性再批量更新 #

这种方式先确认匹配关系是否存在，再决定是否更新，适合df_2数据量较小的场景（比如本例只有一行）：

from pyspark.sql.functions import lit

# 提取df_2中的关键值
target_address = df_2.select("ADDRESS_CODE").first()[0]
target_code = df_2.select("CODE").first()[0]

# 检查该地址码是否在df_1的ZIP_CODE中存在
is_exist = df_1.filter(df_1.ZIP_CODE == target_address).count() > 0

# 根据存在性更新列
if is_exist:
    df_1_updated = df_1.withColumn("UPDATED_MESSAGE", lit(target_code))
else:
    df_1_updated = df_1  # 无匹配则保持原数据

# 查看结果
df_1_updated.show()

执行后结果：

+---+---------------+---------+
| ID|UPDATED_MESSAGE| ZIP_CODE|
+---+---------------+---------+
|  1|      INDIA_WON|5647-0394|
|  2|      INDIA_WON|6748-9384|
|  3|      INDIA_WON|9485-9484|
+---+---------------+---------+

实现方式二：纯DataFrame API关联判断 #

如果需要避免Python端的条件判断，可直接用Spark内置的exists函数实现：

from pyspark.sql.functions import when, exists

# 获取df_2的CODE值（因df_2只有一行，直接提取）
target_code = df_2.select("CODE").first()[0]

# 构造存在性条件：检查df_2中是否有ADDRESS_CODE匹配df_1的ZIP_CODE
match_condition = exists(df_2, df_2.ADDRESS_CODE == df_1.ZIP_CODE)

# 更新UPDATED_MESSAGE列
df_1_updated = df_1.withColumn(
    "UPDATED_MESSAGE",
    when(match_condition, lit(target_code)).otherwise(df_1.UPDATED_MESSAGE)
)

df_1_updated.show()

说明 #

• 两种方式都能实现需求，第一种更直观，适合小数据量的df_2；第二种纯DataFrame操作，更符合Spark的分布式处理逻辑。
• 若df_2有多行，需要先对df_2进行去重或聚合，确保获取唯一的CODE值（根据实际业务逻辑调整）。

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