如何在PySpark中高效更新S3上的Delta表：先删重日期再追加

1. 替换循环删除，提升批量删除性能 #

你之前的循环删除之所以慢，是因为每次循环都会单独发起一次Delta事务，频繁的元数据同步和IO操作拖慢了速度。直接用批量匹配的方式只执行一次删除操作就能解决：

方法一：PySpark API批量删除 #

用isin函数一次性匹配所有要删除的日期，只触发一次事务：

from pyspark.sql.functions import col

# 批量匹配所有待删除日期，一次完成删除
bronze_df.delete(col("day").isin(incremental_day_list))

方法二：Delta SQL批量删除 #

如果想用SQL实现，先把Delta表注册成临时视图，再用IN子句批量过滤日期：

# 将Delta表注册为临时视图
bronze_df.createOrReplaceTempView("bronze_delta_table")

# 构造日期列表的SQL格式字符串，比如('2024-05-01','2024-05-02')
date_list_str = "','".join(incremental_day_list)
delete_sql = f"""
DELETE FROM bronze_delta_table
WHERE day IN ('{date_list_str}')
"""

# 执行SQL删除
spark.sql(delete_sql)

2. 追加CSV数据到Delta表 #

删除完成后，直接用Delta格式写入即可，确保数据以Delta格式保存在S3：

# 假设csv_df是你读取CSV后得到的DataFrame
csv_df.write.format("delta").mode("append").save("s3://your-delta-table-path/")

重要说明 #

• 所有对Delta表的操作（删除、追加等）都会自动维护S3上的Delta元数据和数据文件，结果始终是Delta格式存储，不需要额外转换。
• 批量操作相比循环单条操作，能大幅减少事务次数和IO开销，性能提升非常明显。

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