Azure Databricks Spark是否支持Postgres连接器及更新写入问题咨询

1. Azure Databricks环境下Spark的Postgres连接器支持情况 #

Azure Databricks 原生内置兼容JDBC协议的PostgreSQL连接器，你使用的Spark 3.1.1版本无需额外安装第三方依赖包，直接配置JDBC连接参数即可对接PostgreSQL数据库。

2. Postgres记录upsert/更新实现及overwrite写入问题处理 #

2.1 overwrite模式表被截断但无数据写入的排查方案 #

该问题通常由配置错误、权限不足、数据不匹配导致，按以下顺序排查即可：

• 确认JDBC连接参数配置正确，url格式为jdbc:postgresql://<Postgres主机地址>:<端口>/<数据库名>，user、password、dbtable参数无拼写错误
• 确认Databricks集群的网络安全组、Postgres侧的防火墙、白名单配置允许集群访问数据库端口
• 校验待写入DataFrame的字段名、字段类型、数值精度、字符串长度与Postgres目标表的定义完全匹配
• 确认使用的数据库账号对目标表拥有TRUNCATE、INSERT权限

overwrite写入的标准参考代码如下：

# PostgreSQL连接配置
pg_config = {
    "driver": "org.postgresql.Driver",
    "url": "jdbc:postgresql://192.168.1.100:5432/test_db",
    "dbtable": "public.user_info",
    "user": "pg_user",
    "password": "pg_password"
}

# 执行overwrite写入
df.write.format("jdbc") \
    .options(**pg_config) \
    .mode("overwrite") \
    .save()

2.2 Postgres upsert（更新插入）实现方案 #

Spark JDBC原生没有直接提供upsert接口，适配Spark 3.1.1的常用实现方案有两种：

方案1：foreachPartition批量执行ON CONFLICT语句

适合中小数据量场景，直接利用Postgres原生的ON CONFLICT语法实现upsert，需要提前在目标表的唯一标识字段上创建主键或唯一约束：

def batch_upsert(partition_rows):
    import psycopg2
    # 建立数据库连接
    conn = psycopg2.connect(
        host="192.168.1.100",
        port="5432",
        dbname="test_db",
        user="pg_user",
        password="pg_password"
    )
    cursor = conn.cursor()
    for row in partition_rows:
        # 替换为实际的表名、字段名、唯一约束字段
        upsert_sql = """
        INSERT INTO public.user_info (user_id, user_name, user_age, update_time)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (user_id) DO UPDATE SET
        user_name = EXCLUDED.user_name,
        user_age = EXCLUDED.user_age,
        update_time = EXCLUDED.update_time
        """
        cursor.execute(upsert_sql, (row.user_id, row.user_name, row.user_age, row.update_time))
    conn.commit()
    cursor.close()
    conn.close()

# 对DataFrame每个分区执行upsert
df.rdd.foreachPartition(batch_upsert)

方案2：临时表中转实现批量upsert

适合大数据量场景，减少数据库连接次数，性能更高：

# 第一步：将待写入数据写入Postgres临时表
temp_config = pg_config.copy()
temp_config["dbtable"] = "public.temp_user_upsert"
df.write.format("jdbc") \
    .options(**temp_config) \
    .mode("overwrite") \
    .save()

# 第二步：执行SQL将临时表数据upsert到目标表，完成后删除临时表
spark.read.format("jdbc") \
    .options(**pg_config) \
    .option("query", """
        INSERT INTO public.user_info (user_id, user_name, user_age, update_time)
        SELECT user_id, user_name, user_age, update_time FROM public.temp_user_upsert
        ON CONFLICT (user_id) DO UPDATE SET
        user_name = EXCLUDED.user_name,
        user_age = EXCLUDED.user_age,
        update_time = EXCLUDED.update_time;
        DROP TABLE IF EXISTS public.temp_user_upsert;
    """) \
    .load()

注意事项 #

• 执行upsert前必须确认目标表的唯一标识字段（比如user_id）已经创建主键或唯一约束，否则ON CONFLICT逻辑不会触发
• 数据量较大时可以对DataFrame开启重分区，控制每个分区的行数在1000~10000条之间，避免数据库压力过高
• 正式写入前可以先取少量数据做测试，确认字段匹配、权限正常后再执行全量写入

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