PySpark如何从第二张表不同列取值填充第一张表的缺失值

• 你原有的SQL使用三次左连接+union会生成大量重复行，而且额外新增了id列，没有直接替换原表的original_val字段
• 没有保留非person行原本已经存在的original_val值
• Table2的键值对分散在三列中，更合理的处理逻辑是先把Table2转成长表结构，合并为统一的「姓名-对应值」映射表后，再关联回填Table1的缺失值

方案1：PySpark SQL 实现 #

WITH t2_mapping AS (
    -- 合并Table2三对键值列，生成统一映射表
    SELECT key AS person_name, value_1 AS val FROM Table2 WHERE key IS NOT NULL AND TRIM(key) != ''
    UNION ALL
    SELECT key2 AS person_name, value_2 AS val FROM Table2 WHERE key2 IS NOT NULL AND TRIM(key2) != ''
    UNION ALL
    SELECT key3 AS person_name, value_3 AS val FROM Table2 WHERE key3 IS NOT NULL AND TRIM(key3) != ''
)
SELECT 
    t1.name,
    t1.Value,
    -- 优先保留原有值，为空时使用关联到的映射值
    COALESCE(t1.original_val, t2.val) AS original_val
FROM Table1 t1
LEFT JOIN t2_mapping t2
ON t1.Value = t2.person_name

方案2：PySpark DataFrame API 实现 #

from pyspark.sql.functions import col, coalesce

# 读取Table2并生成统一映射表
t2 = spark.table("Table2")
t2_mapping = t2.select(col("key").alias("person_name"), col("value_1").alias("val")).filter("trim(key) != ''") \
    .unionAll(t2.select(col("key2").alias("person_name"), col("value_2").alias("val")).filter("trim(key2) != ''")) \
    .unionAll(t2.select(col("key3").alias("person_name"), col("value_3").alias("val")).filter("trim(key3) != ''"))

# 关联回填Table1的缺失值
t1 = spark.table("Table1")
result = t1.join(t2_mapping, t1.Value == t2_mapping.person_name, "left") \
    .select(
        t1.name,
        t1.Value,
        coalesce(t1.original_val, t2_mapping.val).alias("original_val")
    )

# 输出结果
result.show()

两种方案执行后都会得到你期望的输出，person行的original_val会正确填充对应值，非person行的原有值完全保留。

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