Spark向SQL Server追加记录失败：主键冲突问题求助

核心问题 #

你的代码存在变量顺序错误，导致删除操作的事务未提交，旧记录仍存在，最终Spark追加时触发主键重复约束。具体来说，你在sql_query变量定义前就执行了cursor.execute(sql_query)，这会触发NameError，代码中断后sql_connection.commit()未执行，删除操作被回滚。

修复步骤 #

1. 修正代码中的变量顺序错误 #

删除多余的提前执行语句，调整后的删除与查询代码如下：

lock = threading.Lock()
lock.acquire()

# 执行删除操作
cursor.execute(f"delete from {tablename} where category_id in {category_id_in_string}")

# 先定义查询语句，再执行查询验证删除结果
sql_query = f"select * from {tablename}"
cursor.execute(sql_query)
records = cursor.fetchall()
for r in records:
    print(r)

# 提交事务，确保删除生效
sql_connection.commit()

lock.release()

2. 验证删除操作的有效性 #

执行上述代码后，检查打印的records，确认目标category_id对应的记录已被完全删除。如果仍存在，需排查：

• category_id_in_string的格式是否正确：比如整数主键需为(6,7,8)，字符串主键需为('6','7')，确保符合SQL语法。
• 表名tablename是否包含正确的 schema（比如retail.categories而非仅categories）。

3. 确保Spark DataFrame无重复主键 #

检查categories_new_df中的category_id列，确认没有重复值，且所有主键都在之前的删除范围内，避免新增不在删除列表中的重复主键。

4. 可选：统一用Spark JDBC执行删除（更可靠） #

如果不想混用pymssql和Spark JDBC连接，可直接用Spark执行删除操作，避免跨连接的事务问题：

# 用Spark执行删除
spark.read.jdbc(url=sql_server_properties['url'], table=f"(DELETE FROM {tablename} WHERE category_id IN {category_id_in_string}) AS tmp", properties=sql_server_properties)

# 或者用Spark SQL（需先注册表）
spark.sql(f"DELETE FROM {tablename} WHERE category_id IN {category_id_in_string}")

# 追加数据
categories_new_df \
    .write \
    .mode("append") \
    .format("jdbc") \
    .options(**sql_server_properties) \
    .save()

关键注意事项 #

• 事务提交必须在删除操作执行且无异常后进行，否则删除会被回滚。
• 多线程环境下的锁需确保覆盖整个删除-提交流程，避免并发操作导致数据不一致。

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