如何在Azure Synapse无服务器SQL池中通过PySpark更新数据行？

Azure Synapse无服务器SQL池本身是基于文件存储（如ADLS Gen2）的查询层，不支持原生的UPDATE/DELETE等DML操作，因为它没有事务性存储引擎。以下是几种可行的替代更新方法：

1. 全量覆盖更新（适合小数据集或允许全量替换的场景） #

直接处理完数据后覆盖原存储路径下的文件，无服务器SQL池的表会自动读取更新后的文件内容。

步骤示例：

• 读取原存储路径的数据到PySpark DataFrame
• 执行数据更新逻辑（比如修改特定行的字段值）
• 将更新后的DataFrame以overwrite模式写入原存储路径

代码示例：

from pyspark.sql.functions import col, when

# 读取原数据（对应无服务器SQL表的存储路径）
df = spark.read.format("parquet").load("abfss://container@yourstorage.dfs.core.windows.net/data/original_table")

# 执行更新：比如将id=123的行status字段改为"active"
updated_df = df.withColumn(
    "status",
    when(col("id") == 123, "active").otherwise(col("status"))
)

# 覆盖写入原路径，注意保持文件格式与原表一致
updated_df.write.format("parquet").mode("overwrite").save("abfss://container@yourstorage.dfs.core.windows.net/data/original_table")

2. 增量合并到新表（适合大数据集，避免全量覆盖） #

如果数据量较大，全量覆盖效率低，可以先合并原数据与更新数据，写入新路径后替换原表。

步骤示例：

• 读取原数据和待更新的增量数据
• 通过关联逻辑合并数据（用增量数据覆盖原数据中匹配的行）
• 将合并后的数据写入新存储路径，创建新的外部表
• 删除原无服务器表，将新表重命名为原表名

代码示例：

# 读取原数据
original_df = spark.read.format("parquet").load("abfss://container@yourstorage.dfs.core.windows.net/data/original")
# 读取增量更新数据（比如仅包含需要修改的行）
updates_df = spark.read.format("parquet").load("abfss://container@yourstorage.dfs.core.windows.net/data/updates")

# 合并逻辑：优先保留增量数据中的字段值，原数据中未匹配的行保持不变
merged_df = original_df.join(updates_df, on="id", how="left_outer") \
    .select(
        original_df["id"],
        when(updates_df["status"].isNotNull(), updates_df["status"]).otherwise(original_df["status"]).alias("status"),
        # 其他字段按同样逻辑处理
        original_df["create_time"]
    )

# 写入新路径
merged_df.write.format("parquet").mode("overwrite").save("abfss://container@yourstorage.dfs.core.windows.net/data/merged")

# 在Synapse无服务器SQL池中执行表替换操作
# DROP TABLE IF EXISTS dbo.original_table;
# CREATE EXTERNAL TABLE dbo.original_table (...)
# WITH (LOCATION = 'abfss://container@yourstorage.dfs.core.windows.net/data/merged', ...);

3. 借助专用SQL池中转（需要事务性更新的场景） #

如果必须使用行级UPDATE操作，可以将数据同步到Synapse专用SQL池（支持事务和DML），更新后再导回无服务器SQL池的存储路径：

• 使用Synapse管道或COPY INTO语句将无服务器表的数据导入专用SQL池
• 在专用SQL池中执行UPDATE语句完成行级更新
• 将更新后的数据导出到无服务器表对应的存储路径，或者直接让应用访问专用SQL池的表

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