含多表关联CASE WHEN的SQL转PySpark DataFrame报错求解

原有代码的错误点 #

• 字段选择遗漏：第一步select仅保留了dt、id、a.val1三个字段，后续逻辑用到的b.val1、b.val2、a.val2、a.val3等字段均未被保留，直接触发字段不存在的报错。
• 列引用方式错误：join生成新的DataFrame后，不能直接使用原来的a、b单表的列对象（比如b.val2、a.val2），需要基于join后的DataFrame引用字段，否则会出现列匹配异常。
• 分组字段引用错误：group by阶段仍然使用a.val1作为字段名，但join后如果没有主动保留前缀，该字段名实际已经变化，会触发找不到字段的报错。

正确实现代码 #

首先导入依赖的函数：

from pyspark.sql import functions as F

写法1（完全对齐原SQL逻辑，简洁版） #

# 给表起别名避免字段冲突，引用更清晰
c_alias = c.alias("c")
a_alias = a.alias("a")
b_alias = b.alias("b")

# 多表关联
joined = c_alias.join(a_alias, F.col("c.e_id") == F.col("a.e_id"), "inner") \
                .join(b_alias, F.col("c.m_id") == F.col("b.m_id"), "left_outer")

# 逻辑计算+分组统计
df = joined.select(
    F.col("c.dt"),
    F.col("c.id"),
    F.col("a.val1"),
    F.when(F.col("b.val1") == False, True).otherwise(False).alias("inf"),
    F.when(F.col("b.val1") == False, F.coalesce(F.col("b.val2"), F.col("a.val2"))).otherwise(F.col("a.val2")).alias("mob"),
    F.when(F.col("b.val1") == False, F.coalesce(F.col("b.val3"), F.col("a.val3"))).otherwise(F.col("a.val3")).alias("cli"),
    F.col("a.val3").alias("fam")
).groupBy("dt", "id", "val1", "inf", "mob", "cli", "fam") \
 .count() \
 .withColumnRenamed("count", "unique_val")

写法2（分步计算，可读性更高） #

# 给表起别名避免字段冲突，引用更清晰
c_alias = c.alias("c")
a_alias = a.alias("a")
b_alias = b.alias("b")

# 多表关联
joined = c_alias.join(a_alias, F.col("c.e_id") == F.col("a.e_id"), "inner") \
                .join(b_alias, F.col("c.m_id") == F.col("b.m_id"), "left_outer")

# 分步计算衍生字段
df = joined.withColumn("inf", F.when(F.col("b.val1") == False, True).otherwise(False)) \
           .withColumn("mob", F.when(F.col("b.val1") == False, F.coalesce(F.col("b.val2"), F.col("a.val2"))).otherwise(F.col("a.val2"))) \
           .withColumn("cli", F.when(F.col("b.val1") == False, F.coalesce(F.col("b.val3"), F.col("a.val3"))).otherwise(F.col("a.val3"))) \
           .select(
               F.col("c.dt"),
               F.col("c.id"),
               F.col("a.val1"),
               "inf",
               "mob",
               "cli",
               F.col("a.val3").alias("fam")
           ).groupBy("dt", "id", "val1", "inf", "mob", "cli", "fam") \
            .count() \
            .withColumnRenamed("count", "unique_val")

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