将含多内连接与WHERE的SQL UPDATE转为PySpark时遇错求助

核心错误点 #

• 列引用不匹配：定义join_on时直接使用原始DataFrame（df_2/df_3/df_4），但后续JOIN操作已给表设置别名（a/b/c/d），Spark无法识别这些列的归属；同时将WHERE条件错误混入JOIN条件，导致解析时误以为这些列属于df_1（别名a）。
• JOIN条件滥用：三次JOIN共用同一个join_on，但每个JOIN仅需对应两张表的关联条件，叠加所有条件会导致逻辑混乱和解析错误。
• 逻辑偏离原始SQL：原始SQL是更新companyc1.sales为500，你的PySpark代码却在选取df_2的列，完全不符合需求。

修正后的PySpark代码 #

from pyspark.sql import functions as F

# 拆分定义各表的JOIN条件
join_1_2 = (F.col("a.C1_PROFIT") == F.col("b.C2_PROFIT"))
join_1_3 = (F.col("a.C1_REVENUE") == F.col("c.C3_REVENUE_BREAK"))
join_1_4 = (F.col("a.C1_LOSS") == F.col("d.C4_TOTAL_LOSS"))

# 对齐原始SQL的INNER JOIN逻辑，最后添加WHERE过滤条件
df = (df_1.alias("a")
      .join(df_2.alias("b"), join_1_2, "inner")
      .join(df_3.alias("c"), join_1_3, "inner")
      .join(df_4.alias("d"), join_1_4, "inner")
      # 满足WHERE条件时更新sales为500，否则保留原值
      .withColumn("sales", F.when(F.col("d.TOTAL_YEAR_PROFIT") > F.col("c.TOTAL_GROWTH"), "500").otherwise(F.col("a.sales")))
      # 按需选择输出列，这里以保留companyc1所有列+更新后的sales为例
      .select("a.*", "sales")
)

额外说明 #

• 原始SQL使用INNER JOIN，你之前写的是left，若要严格对齐原始逻辑，必须用inner；若需保留LEFT JOIN逻辑，将"inner"改为"left"即可，注意空值处理。
• 统一用F.col()引用列，避免直接用DataFrame对象引用，逻辑更清晰且不易出错。

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