如何在PySpark中实现类Spark SQL的Update操作及改写指定SQL代码

1. 用 PySpark DataFrame API 实现类似 SQL UPDATE SET 的逻辑 #

Spark DataFrame 是**不可变（immutable）**的，无法直接对原有数据集执行修改操作，只能通过生成新的 DataFrame 来模拟 UPDATE SET 的效果，核心是使用 withColumn 方法结合内置函数实现列值修改：

全量更新列值 #

如果需要对所有行的指定列进行统一修改：

from pyspark.sql.functions import lit, current_timestamp

# 假设original_df是你的原始DataFrame
updated_df = original_df.withColumn("status", lit("success")) \
                        .withColumn("date", current_timestamp())

条件更新列值 #

如果需要仅对满足特定条件的行修改列值，使用 when/otherwise 函数实现分支逻辑：

from pyspark.sql.functions import when, lit, current_timestamp

# 示例：仅当"id"大于100的行，更新status和date
updated_df = original_df.withColumn(
    "status",
    when(original_df.id > 100, lit("success")).otherwise(original_df.status)
).withColumn(
    "date",
    when(original_df.id > 100, current_timestamp()).otherwise(original_df.date)
)

2. 替代 Spark SQL UPDATE 语句的纯 PySpark 实现 #

你提供的 Spark SQL 语句是对指定表执行全量更新，纯 PySpark 实现分两种场景：

场景1：普通数据源（非ACID表） #

普通数据源（如Parquet、CSV等）不支持原地修改，需要先读取数据、修改后覆盖原数据源：

# 1. 读取原表/数据源（fp为表名或存储路径）
original_df = spark.read.table(fp)  # 读取表；如果是路径用spark.read.load(fp)

# 2. 执行全量更新逻辑
updated_df = original_df.withColumn("status", lit("success")) \
                        .withColumn("date", current_timestamp())

# 3. 覆盖写回原表/路径
updated_df.write.mode("overwrite").saveAsTable(fp)  # 写回Hive表
# 若为路径存储：updated_df.write.mode("overwrite").save(fp)

场景2：ACID兼容数据源（如Delta Lake） #

如果使用 Delta Lake 支持ACID事务，可以直接执行原地更新，更贴近 SQL UPDATE 的语义：

from delta.tables import DeltaTable

# 加载Delta表（fp为表名或存储路径）
delta_table = DeltaTable.forName(spark, fp)  # 或DeltaTable.forPath(spark, fp)

# 执行全量更新，对应SQL的"UPDATE ... SET ..."无WHERE条件
delta_table.update(
    condition="1=1",  # 1=1表示匹配所有行
    set={
        "status": "'success'",
        "date": "current_timestamp()"
    }
)

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