Databricks使用PySpark连接PostgreSQL删除记录报语法错误咨询

spark.read.format('jdbc').load() 是Spark封装的只读数据读取接口，你传入的query参数会被Spark自动包装成子查询，最终发给PostgreSQL的执行语句长这样：

SELECT * FROM (DELETE FROM meta.test4 WHERE Emp_Id = 1) AS spark_gen_subquery

PostgreSQL不允许在子查询里直接执行DELETE这类无返回结果的DML语句，解析时自然会报FROM附近的语法错误，这个接口从设计上就不支持删、改这类写操作。

方案1：直接通过JDBC连接执行删除（推荐，无额外依赖） #

Databricks运行环境自带JDBC驱动，可以直接通过Py4j调用Java层的JDBC接口执行删除语句，不需要额外安装第三方库，适合固定条件的删除场景：

# 导入JDBC依赖
from py4j.java_gateway import java_import
java_import(spark._jvm, 'java.sql.DriverManager')

# 建立数据库连接
conn = spark._jvm.DriverManager.getConnection(jdbcUrl, user, password)
state = conn.createStatement()

# 执行删除操作
delete_sql = "DELETE FROM meta.test4 WHERE Emp_Id = 1"
# executeUpdate会返回受影响的行数
del_count = state.executeUpdate(delete_sql)
print(f"删除完成，共影响{del_count}条记录")

# 必须关闭连接释放资源
state.close()
conn.close()

方案2：批量删除（适合待删除数据量较大的场景） #

如果你的待删除条件是存放在Spark DataFrame里的批量ID，可以用foreachPartition按分区建立数据库连接批量删除，避免频繁建连的性能损耗：

# 假设待删除的ID存在DataFrame del_df中，字段为Emp_Id
def batch_del(partition_rows):
    # 用集群自带的Java JDBC实现，不需要额外装Python依赖
    from py4j.java_gateway import java_import
    gw = spark.sparkContext._gateway
    java_import(gw.jvm, 'java.sql.DriverManager')
    conn = gw.jvm.DriverManager.getConnection(jdbcUrl, user, password)
    stmt = conn.createStatement()
    id_buffer = []
    for row in partition_rows:
        id_buffer.append(str(row.Emp_Id))
        # 每攒1000条执行一次批量删除，避免单条SQL过长
        if len(id_buffer) >= 1000:
            del_sql = f"DELETE FROM meta.test4 WHERE Emp_Id IN ({','.join(id_buffer)})"
            stmt.executeUpdate(del_sql)
            id_buffer.clear()
    # 处理剩余不足1000条的记录
    if id_buffer:
        del_sql = f"DELETE FROM meta.test4 WHERE Emp_Id IN ({','.join(id_buffer)})"
        stmt.executeUpdate(del_sql)
    conn.commit()
    stmt.close()
    conn.close()

del_df.foreachPartition(batch_del)

• 不要尝试用spark.write.jdbc的overwrite模式做条件删除，该模式会直接清空整表后写入新数据，很容易造成全表数据误删。
• 执行删除前建议先把DELETE语句换成SELECT COUNT(*)执行下，确认待删除的数据量符合预期，避免误删。

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