如何通过PySpark使用JDBC连接器删除Azure SQL表中的数据？

你之前使用spark.read接口执行DELETE语句的方式不成立，因为该接口仅用于执行返回结果集的查询类语句，不支持增删改这类DML操作。以下是两种可直接在Python中实现的可行方案：

方案1：单节点执行（适合删除逻辑简单、数据量不大的场景） #

直接通过Py4J网关调用Java JDBC接口执行删除语句，无需额外安装Python依赖，仅要求Spark集群已加载SQL Server JDBC驱动。

# 配置Azure SQL连接参数
jdbc_url = "jdbc:sqlserver://<你的服务器名>.database.windows.net:1433;databaseName=<你的库名>;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.database.windows.net;loginTimeout=30;"
db_user = "<用户名>"
db_pwd = "<密码>"
delete_sql = "DELETE FROM <你的表名> WHERE <删除条件>"

# 获取Java侧的JDBC DriverManager
driver_manager = spark._sc._jvm.java.sql.DriverManager
connection = driver_manager.getConnection(jdbc_url, db_user, db_pwd)

try:
    statement = connection.createStatement()
    affected_rows = statement.executeUpdate(delete_sql)
    print(f"删除完成，受影响行数：{affected_rows}")
finally:
    # 手动关闭连接避免资源泄漏
    if connection:
        connection.close()

方案2：分布式执行（适合需根据Spark DataFrame数据批量删除的场景） #

如果删除条件对应的数据存储在Spark DataFrame中（比如要删除的主键ID列表），可以用foreachPartition分布式执行删除操作，提升大批量删除的效率。

# 示例：delete_id_df 为存储了待删除记录主键ID的Spark DataFrame，仅包含id列
delete_sql_template = "DELETE FROM <你的表名> WHERE id = ?"
jdbc_url = "jdbc:sqlserver://<你的服务器名>.database.windows.net:1433;databaseName=<你的库名>;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.database.windows.net;loginTimeout=30;"
db_user = "<用户名>"
db_pwd = "<密码>"

def delete_partition(rows):
    # 分区内复用JDBC连接，减少连接开销
    driver_manager = spark._sc._jvm.java.sql.DriverManager
    conn = driver_manager.getConnection(jdbc_url, db_user, db_pwd)
    try:
        statement = conn.prepareStatement(delete_sql_template)
        for row in rows:
            statement.setInt(1, row.id)
            statement.addBatch()
        # 批量执行当前分区的所有删除操作
        statement.executeBatch()
    finally:
        if conn:
            conn.close()

# 触发分布式执行
delete_id_df.foreachPartition(delete_partition)

注意事项 #

• 执行删除操作前务必先在测试环境验证删除条件，避免误删数据，可先将DELETE语句替换为SELECT COUNT(*)验证匹配的行数是否符合预期。
• 提交Spark作业时可通过--packages com.microsoft.sqlserver:mssql-jdbc:12.4.2.jre11参数自动加载对应版本的SQL Server JDBC驱动，无需手动上传jar包。
• 大批量删除数据时建议分批执行，避免长时间锁表影响业务正常访问。

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