PySpark写入PostgreSQL遇重复数据问题，求应用层解决方案

问题根源 #

使用mode("overwrite")配合truncate=True写入PostgreSQL时，Spark的执行流程存在容错漏洞：

1. Driver先执行TRUNCATE TABLE清空目标表；
2. 多个Executor并行写入各自分区的数据，每个分区的写入是独立事务。

当Executor中途宕机，Spark会重试对应分区的写入任务。如果该分区的第一次写入已成功提交部分（或全部）数据，重试时会再次写入整个分区的内容，最终导致目标表出现重复数据。由于添加主键会大幅降低写入性能，以下提供无需主键的应用层解决方案。

方案1：原子替换临时表（推荐） #

核心思路是先将完整数据写入临时表，确认写入成功后，通过原子DDL操作替换目标表。这种方式确保目标表要么保留旧数据，要么完全替换为新数据，不会出现部分写入或重复的中间状态。

实现代码 #

import uuid
from pyspark.sql import SparkSession

spark = SparkSession.getActiveSession()

# 生成唯一临时表名，避免多任务并发冲突
temp_table = f"tmp_{MY_DB_TABLE}_{uuid.uuid4().hex[:8]}"

# 第一步：将数据写入临时表
df.write \
  .mode("overwrite") \
  .format("jdbc") \
  .option("url", MY_DB_URL) \
  .option("dbtable", temp_table) \
  .option("user", MY_USERNAME) \
  .option("password", MY_PASS) \
  .option("driver", "org.postgresql.Driver") \
  .option("stringtype", "unspecified") \
  .option("numPartitions", 90) \
  .option("batchsize", 100000) \
  .save()

# 第二步：原子替换目标表（事务包裹确保原子性）
replace_query = f"""
BEGIN;
DROP TABLE IF EXISTS {MY_DB_TABLE};
ALTER TABLE {temp_table} RENAME TO {MY_DB_TABLE};
COMMIT;
"""

# 通过JDBC直接执行DDL
conn = spark.sparkContext._jvm.java.sql.DriverManager.getConnection(
    MY_DB_URL, MY_USERNAME, MY_PASS
)
try:
    stmt = conn.createStatement()
    stmt.execute(replace_query)
finally:
    conn.close()

关键优势 #

• 原子性保障：PostgreSQL的ALTER TABLE RENAME是原子操作，替换过程要么完全成功，要么原表保持不变。
• 无性能损耗：无需添加主键，写入临时表的性能与原方案一致，替换操作是轻量DDL，几乎不占用额外资源。
• 容错性强：如果写入临时表时Executor宕机，临时表会被后续任务覆盖或自动清理，原表不受任何影响。

注意事项 #

• 确保数据库有足够存储空间容纳临时表（数据量与目标表一致）。
• 若目标表存在关联视图或外键，需提前处理依赖关系，避免替换后出现报错。
• 临时表名采用UUID后缀，避免多任务并发执行时的命名冲突。

方案2：PostgreSQL COPY命令（高性能备选） #

针对超大数据集，可先将Spark数据导出为文本文件，再通过PostgreSQL的COPY命令批量导入临时表，最后执行原子替换。COPY比JDBC批量写入性能更高，适合TB级数据：

import uuid
import psycopg2
from pyspark.sql import SparkSession

spark = SparkSession.getActiveSession()
temp_table = f"tmp_{MY_DB_TABLE}_{uuid.uuid4().hex[:8]}"

# 第一步：将DataFrame导出为TSV文件（存储到PostgreSQL可访问的路径）
df.write.mode("overwrite").csv("/tmp/spark_export", header=False, sep="\t")

# 第二步：通过psycopg2执行COPY导入与表替换
conn = psycopg2.connect(
    host=DB_HOST,
    database=DB_NAME,
    user=MY_USERNAME,
    password=MY_PASS
)
cur = conn.cursor()

try:
    # 创建与目标表结构一致的临时表
    cur.execute(f"CREATE TABLE {temp_table} AS SELECT * FROM {MY_DB_TABLE} LIMIT 0")
    # 批量导入数据
    with open("/tmp/spark_export/part-00000", "r") as f:
        cur.copy_from(f, temp_table, sep="\t")
    conn.commit()
    
    # 原子替换目标表
    cur.execute(f"DROP TABLE IF EXISTS {MY_DB_TABLE}; ALTER TABLE {temp_table} RENAME TO {MY_DB_TABLE}")
    conn.commit()
finally:
    cur.close()
    conn.close()

注意：若使用分布式存储（如HDFS），需确保PostgreSQL服务器能访问到导出文件（可通过NFS挂载或hdfs dfs -get同步到本地）。

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