如何使用Python将Databricks中的DataFrame写入PostgreSQL？

嘿，我来帮你搞定这个把Databricks的DataFrame写入PostgreSQL的需求～首先先补全用psycopg2逐行写入的完整代码，同时得提醒你：这种方式只适合小数据量场景，数据量大时效率极低，后面我会给你更高效的方案。

用psycopg2逐行写入的完整实现 #

首先确保你已经在Databricks集群里安装了psycopg2-binary库（可以通过集群库管理添加），然后就可以用下面的代码：

import psycopg2

# 假设你已经有了customerDetails这个Spark DataFrame
# 先把数据拉到Driver节点（注意：数据量大时collect()会导致内存溢出！）
customer_records = customerDetails.collect()

# 建立PostgreSQL连接，替换成你的数据库参数
db_conn = psycopg2.connect(
    dbname="your_database_name",
    user="your_db_user",
    password="your_db_password",
    host="your_db_host",
    port="your_db_port"
)
cursor = db_conn.cursor()

# 定义插入SQL，假设你的PostgreSQL表名为customer_details，字段对应
insert_sql = """
INSERT INTO customer_details (customer_name, customer_id)
VALUES (%s, %s)
"""

try:
    # 循环每一行插入
    for row in customer_records:
        # 注意Spark Row对象的字段访问，strip()用来清理数据里的空格
        cursor.execute(insert_sql, (row.customerName.strip(), row.customerId.strip()))
    # 提交事务
    db_conn.commit()
    print("所有数据成功写入PostgreSQL！")
except Exception as err:
    # 出错就回滚事务
    db_conn.rollback()
    print(f"写入失败：{str(err)}")
finally:
    # 一定要关闭游标和连接
    cursor.close()
    db_conn.close()

划重点：逐行写入的坑 #

• 内存风险：collect()会把整个DataFrame的数据加载到Driver节点内存里，如果数据量超过Driver内存，直接就OOM崩溃了。
• 性能极差：每一行都要单独发起一次数据库请求，数据量稍微大一点（比如几万行），写入速度会慢到让人崩溃。

更推荐的高效方案：用Spark JDBC直接写入 #

这才是Databricks写入关系型数据库的标准姿势，分布式并行写入，完全不用把数据拉到Driver，代码还更简洁：

# 配置JDBC连接信息
jdbc_url = "jdbc:postgresql://your_db_host:your_db_port/your_database_name"
jdbc_properties = {
    "user": "your_db_user",
    "password": "your_db_password",
    "driver": "org.postgresql.Driver"
}

# 写入PostgreSQL，mode参数可选：append（追加）、overwrite（覆盖）、ignore（忽略已存在）、errorifexists（已存在则报错）
customerDetails.write.jdbc(
    url=jdbc_url,
    table="customer_details",
    mode="append",
    properties=jdbc_properties
)

这种方式的优势不用多说：分布式处理，速度快，内存安全，代码还少，几乎是Databricks写入PostgreSQL的首选方法。

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