如何在PySpark中实现T-SQL的UPDATE...SET更新逻辑？

原T-SQL逻辑说明：

当临时表#TERT（A）的NO与表KMLK（B）的NO匹配，且A的TARIH为空字符串、B的DURUM为1时，将A的TARIH字段更新为B的DTARIH。

PySpark没有直接支持T-SQL的UPDATE JOIN语法，需通过关联数据+条件赋值实现，以下是两种常用方案：

方法1：DataFrame API实现 #

假设你已拥有对应的数据框：

• tert_df：对应T-SQL中的临时表#TERT
• kmlk_df：对应T-SQL中的表KMLK

from pyspark.sql import functions as F

# 提前筛选KMLK中符合条件的记录，减少关联数据量
filtered_kmlk = kmlk_df.filter(F.col("DURUM") == 1).select("NO", "DTARIH")

# 左关联两个表，保留原TERT的所有行
joined_df = tert_df.join(filtered_kmlk, on="NO", how="left")

# 条件更新TARIH字段，符合要求则替换为DTARIH，否则保留原值
updated_tert_df = joined_df.withColumn(
    "TARIH",
    F.when(
        (F.col("TARIH") == "") & (F.col("DTARIH").isNotNull()),
        F.col("DTARIH")
    ).otherwise(F.col("TARIH"))
).drop("DTARIH")  # 移除临时关联字段

# 若需覆盖原数据框，直接重新赋值即可
tert_df = updated_tert_df

方法2：Spark SQL实现 #

如果更习惯SQL语法，可通过创建临时视图完成转换：

# 创建临时视图，映射原表
tert_df.createOrReplaceTempView("TERT")
kmlk_df.createOrReplaceTempView("KMLK")

# 执行SQL查询，通过子查询+CASE WHEN完成字段更新
updated_tert_df = spark.sql("""
    SELECT 
        t.NO,
        CASE 
            WHEN t.TARIH = '' AND k.DTARIH IS NOT NULL THEN k.DTARIH
            ELSE t.TARIH
        END AS TARIH,
        -- 需显式列出TERT表的其他所有字段，保证结构完整
        t.字段1,
        t.字段2
    FROM TERT t
    LEFT JOIN (
        SELECT NO, DTARIH 
        FROM KMLK 
        WHERE DURUM = 1
    ) k ON t.NO = k.NO
""")

注意事项 #

• 两种方案均使用左关联（LEFT JOIN），确保原#TERT中的所有行都被保留，仅更新符合条件的记录。
• 提前筛选KMLK的条件DURUM=1，能有效降低关联运算的数据量，提升执行效率。

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