通过PySpark向外部数据库写入时如何实现更新、删除等自定义操作

Spark原生JDBC写入方法仅提供append、overwrite等基础写入模式，要实现更新、删除等自定义逻辑，可通过以下两种方案实现，两种方案都支持自定义SQL语句控制写入逻辑：

方案1：逐分区执行自定义SQL（灵活度最高） #

该方案基于foreachPartition算子按分区处理数据，每个分区复用一个JDBC连接，既保证了自定义逻辑的灵活度，也避免了单条数据创建连接的性能损耗，可以实现任意自定义DML操作（更新、删除、Upsert等）。

Upsert（存在更新、不存在插入）示例代码 #

import pymysql

# MySQL连接配置
db_host = "<Azure-MySQL域名>"
db_port = 3306
db_user = "<用户名>"
db_passwd = "<密码>"
db_name = "<数据库名>"
target_table = "<目标表名>"

def upsert_partition(partition):
    # 每个分区创建1次数据库连接
    conn = pymysql.connect(
        host=db_host,
        port=db_port,
        user=db_user,
        password=db_passwd,
        database=db_name
    )
    cursor = conn.cursor()
    # 自定义Upsert SQL，示例中主键为id，匹配到主键则更新name、age字段
    upsert_sql = f"""
    INSERT INTO {target_table} (id, name, age)
    VALUES (%s, %s, %s)
    ON DUPLICATE KEY UPDATE name=VALUES(name), age=VALUES(age)
    """
    batch_data = []
    try:
        for row in partition:
            batch_data.append((row.id, row.name, row.age))
            # 每1000条批量提交一次，可根据实际数据量调整阈值
            if len(batch_data) >= 1000:
                cursor.executemany(upsert_sql, batch_data)
                conn.commit()
                batch_data = []
        # 提交剩余的不足1000条的数据
        if batch_data:
            cursor.executemany(upsert_sql, batch_data)
            conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()
        conn.close()

# 执行写入操作
spark_df.foreachPartition(upsert_partition)

删除操作示例代码 #

只需要修改自定义SQL即可实现删除逻辑，示例为匹配DataFrame中的id字段删除目标表对应行：

def delete_partition(partition):
    conn = pymysql.connect(
        host=db_host,
        port=db_port,
        user=db_user,
        password=db_passwd,
        database=db_name
    )
    cursor = conn.cursor()
    delete_sql = f"DELETE FROM {target_table} WHERE id = %s"
    batch_data = []
    try:
        for row in partition:
            batch_data.append((row.id,))
            if len(batch_data) >= 1000:
                cursor.executemany(delete_sql, batch_data)
                conn.commit()
                batch_data = []
        if batch_data:
            cursor.executemany(delete_sql, batch_data)
            conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()
        conn.close()

spark_df.foreachPartition(delete_partition)

方案2：临时表中转合并（适合大数据量场景） #

如果写入数据量较大，可先将DataFrame写入MySQL临时表，再通过原生SQL的合并逻辑将临时表数据同步到目标表，性能比逐行处理更高。

# 1. 将DataFrame写入MySQL临时表
temp_table = "temp_merge_data"
mysql_url = "jdbc:mysql://<Azure-MySQL域名>:3306/<数据库名>?rewriteBatchedStatements=true"
spark_df.write.jdbc(
    url=mysql_url,
    table=temp_table,
    mode="overwrite",
    properties={"user":db_user, "password": db_passwd, "driver": "com.mysql.cj.jdbc.Driver" }
)

# 2. 执行合并SQL将临时表数据同步到目标表
conn = pymysql.connect(host=db_host, port=db_port, user=db_user, password=db_passwd, database=db_name)
cursor = conn.cursor()
merge_sql = f"""
INSERT INTO {target_table} (id, name, age)
SELECT id, name, age FROM {temp_table}
ON DUPLICATE KEY UPDATE name=VALUES(name), age=VALUES(age)
"""
cursor.execute(merge_sql)
# 合并完成后删除临时表
cursor.execute(f"DROP TABLE IF EXISTS {temp_table}")
conn.commit()
cursor.close()
conn.close()

注意事项 #

• 依赖部署：如果使用pymysql操作数据库，需要确保所有Spark Worker节点都已经安装pymysql库，避免运行时报错
• 性能优化：JDBC URL添加rewriteBatchedStatements=true参数可大幅提升批量SQL执行效率
• 索引要求：使用ON DUPLICATE KEY UPDATE前需要确保目标表存在主键或唯一索引，否则只会执行普通插入逻辑
• SSL适配：如果Azure MySQL开启了强制SSL连接，需要在pymysql连接参数中添加SSL证书路径配置

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