PySpark通过JDBC向PostgreSQL同表Upsert时数据被清空的问题

嘿，我来帮你拆解这个问题——你遇到的情况其实是Spark JDBC写入参数组合的冲突导致的，咱们一步步理清楚：

先还原你的操作场景 #

你做了这几步操作：

• 从PostgreSQL的test_spark表读取数据生成DataFrame
• 用withColumn更新了id=10的记录为id=20
• 尝试把更新后的DataFrame Upsert回原表时，表被清空了，但写入新表test_spark1却完全正常

问题到底出在哪？ #

核心冲突点在**mode("overwrite")、truncate=True和upsert=True这三个参数的组合逻辑**：

1. 当你指定mode("overwrite")+truncate=True时，Spark会优先执行TRUNCATE TABLE清空目标表——这一步是在任何Upsert逻辑之前跑的，等于先把原表数据全删了，再插入你更新后的DataFrame。如果你的更新后的DataFrame只包含部分原表数据（比如只修改了一行），那看起来就像是“表被清空了”。
2. 而写入新表时，要么表压根不存在（Spark会自动创建），要么是空表，truncate=True根本起不到作用，Upsert逻辑自然能正常把数据插进去。
3. 另外要提一句：Spark标准JDBC其实并不原生支持upsert和condition_columns这两个参数，这大概率是你用了第三方扩展（比如Databricks的插件或者自定义驱动）。但不管怎样，overwrite+truncate的优先级远高于Upsert逻辑，这才是问题根源。

正确实现Upsert的两种方案 #

要实现PostgreSQL的Upsert（也就是INSERT ... ON CONFLICT DO UPDATE的逻辑），你得换掉overwrite和truncate，试试这两种方法：

方案1：用foreachBatch结合原生PostgreSQL语句（最可靠）

这种方式完全由你控制Upsert逻辑，兼容性最好：

from pyspark.sql import functions as F

def upsert_batch_to_postgres(batch_df, batch_id):
    # 把当前批次的数据写入临时视图
    batch_df.createOrReplaceTempView("temp_update_data")
    
    # 执行原生的PostgreSQL Upsert语句，替换成你实际的列名
    batch_df.sparkSession.sql("""
        INSERT INTO test_spark (id, col1, col2)
        SELECT id, col1, col2 FROM temp_update_data
        ON CONFLICT (id) DO UPDATE SET
            col1 = EXCLUDED.col1,
            col2 = EXCLUDED.col2
    """)

# 批处理场景直接用这个
newDf.write.foreachBatch(upsert_batch_to_postgres).save()

方案2：调整JDBC参数（依赖第三方支持）

如果你的JDBC驱动确实支持upsert参数，那可以改成append模式，去掉overwrite和truncate：

newDf.write.mode("append")
       .option("upsert", True)
       .option("condition_columns", "id")
       .format("jdbc")
       .option("url", url)
       .option("dbtable", "test_spark")
       .save()

注意：这个方案只适用于支持upsert参数的扩展驱动，标准Spark JDBC用不了，所以更推荐方案1。

为啥新表就没问题？ #

再补一句你可能好奇的点：写入test_spark1时，要么表不存在（Spark自动建表后插入数据），要么表是空表——truncate清空空表等于没操作，之后Upsert逻辑直接插入数据，自然不会有问题。而原表有数据，truncate先把数据清了，再插入更新后的DataFrame，就出现了你看到的“数据被清空”的现象。

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