如何用Spark向外部数据库执行非读写类SQL语句（如MERGE）？

Spark本身没有原生提供类似SQLAlchemy的任意SQL执行API，但可以通过两种可靠方式实现你的需求：

方法1：直接通过JDBC连接执行语句（推荐） #

借助Java的DriverManager类直接获取RDS的JDBC连接，执行MERGE、GRANT这类无返回操作，这是最直接高效的方式。

操作步骤示例 #

1. 定义RDS PostgreSQL的连接参数：

jdbc_url = "jdbc:postgresql://<你的RDS端点>:5432/<数据库名>"
jdbc_properties = {
    "user": "<数据库用户名>",
    "password": "<数据库密码>",
    "driver": "org.postgresql.Driver"
}

2. 编写PostgreSQL兼容的MERGE语句（注意：PostgreSQL 15及以上支持标准MERGE，低版本需用UPSERT语法替代）：

MERGE INTO test t
USING staging_test st
ON t.id = st.id  -- 替换为你的匹配主键/唯一键
WHEN MATCHED THEN
  UPDATE SET col1 = st.col1, col2 = st.col2  -- 替换为需要更新的字段
WHEN NOT MATCHED THEN
  INSERT (id, col1, col2) VALUES (st.id, st.col1, st.col2);  -- 替换为插入字段

3. 在Databricks中执行语句：

Python版本

from java.sql import DriverManager

# 获取JDBC连接
conn = DriverManager.getConnection(jdbc_url, jdbc_properties["user"], jdbc_properties["password"])
try:
    stmt = conn.createStatement()
    # 执行MERGE语句
    stmt.execute("""
        MERGE INTO test t
        USING staging_test st
        ON t.id = st.id
        WHEN MATCHED THEN
          UPDATE SET col1 = st.col1, col2 = st.col2
        WHEN NOT MATCHED THEN
          INSERT (id, col1, col2) VALUES (st.id, st.col1, st.col2);
    """)
    conn.commit()
finally:
    stmt.close()
    conn.close()

Scala版本

import java.sql.DriverManager

val conn = DriverManager.getConnection(jdbc_url, jdbc_properties("user"), jdbc_properties("password"))
try {
  val stmt = conn.createStatement()
  stmt.execute("""
    MERGE INTO test t
    USING staging_test st
    ON t.id = st.id
    WHEN MATCHED THEN
      UPDATE SET col1 = st.col1, col2 = st.col2
    WHEN NOT MATCHED THEN
      INSERT (id, col1, col2) VALUES (st.id, st.col1, st.col2);
  """)
  conn.commit()
} finally {
  stmt.close()
  conn.close()
}

方法2：通过临时视图映射远程表执行SQL #

如果需要复用JDBC连接，可以将RDS表映射为Databricks临时视图，再通过spark.sql执行MERGE语句：

# 注册临时视图（仅建立连接映射，不会加载全量数据）
spark.read.jdbc(url=jdbc_url, table="test", properties=jdbc_properties).createOrReplaceTempView("test_jdbc")
spark.read.jdbc(url=jdbc_url, table="staging_test", properties=jdbc_properties).createOrReplaceTempView("staging_test_jdbc")

# 执行MERGE语句（SQL会被发送到RDS端执行）
spark.sql("""
    MERGE INTO test_jdbc t
    USING staging_test_jdbc st
    ON t.id = st.id
    WHEN MATCHED THEN
      UPDATE SET col1 = st.col1, col2 = st.col2
    WHEN NOT MATCHED THEN
      INSERT (id, col1, col2) VALUES (st.id, st.col1, st.col2);
""")

关键注意事项 #

• 确保Databricks集群已安装PostgreSQL JDBC驱动，可通过集群库添加org.postgresql:postgresql:<对应版本>。
• PostgreSQL 15以下版本不支持标准MERGE，需用UPSERT语法替代：

INSERT INTO test (id, col1, col2)
SELECT id, col1, col2 FROM staging_test
ON CONFLICT (id) DO UPDATE SET
  col1 = EXCLUDED.col1,
  col2 = EXCLUDED.col2;

• 执行无返回语句时必须用execute()，不能用executeQuery()（后者会尝试获取结果集导致报错）。
• 操作完成后务必关闭连接，避免资源泄漏。

内容的提问来源于stack exchange，提问作者Eugenio.Gastelum96