PySpark DataFrame多线程写入Snowflake时出现锁表问题求助

报错根因 #

你遇到的报错是Snowflake表级锁等待队列溢出导致的：默认情况下，Snowflake对表执行DML写入操作时会加表级锁，每个写入事务持有锁期间，后续的写入请求会进入等待队列，队列长度上限为20，超过上限后新的请求会直接被终止。
Python多线程场景下每个写入任务会独立开启事务，锁释放不及时就会快速触发队列上限。

可行解决建议 #

• 优先采用Spark原生批量写入，取消Python多线程逻辑
  Spark本身是分布式计算框架，单批次写入Snowflake时会自动利用集群资源做并行加载，性能远高于Python多线程小批量写入。你可以将多线程中待写入的所有DataFrame先做Union合并，再调用单次写入方法即可，从根源避免锁冲突。

# 合并多份待写入DataFrame后单次写入
all_df = df1.unionByName(df2).unionByName(df3) # 按实际场景合并所有数据
sfOptions = {
  "sfURL" : "XXXXXXXXXXXX",
  "sfAccount" : "XXXXXXXXXXXX",
  "sfUser" : "XXXXXXXXXXXX",
  "sfPassword" : "XXXXXXXXXXXX",
  "sfDatabase" : "XXXXXXXXXXXX",
  "sfSchema" : "XXXXXXXXXXXX",
  "sfWarehouse" : "XXXXXXXXXXXX",
  "sfRole" : "XXXXXXXXXXXX",
  "column_mapping" : "name",
  "column_mismatch_behavior":"ignore",
  "autocommit": "true" # 新增自动提交配置，写入完成立刻释放锁
}
all_df.write.format("snowflake") \
              .options(**sfOptions) \
              .option("dbtable", table).mode("append").save()

• 保留多线程逻辑的优化方案
  如果业务逻辑必须使用多线程写入，可做以下调整：
  1. 在sfOptions中新增"autocommit": "true"配置，让每次写入完成后立刻提交事务释放锁，大幅缩短锁持有时间
  2. 对多线程做限流，将并发写入线程数控制在10个以内，避免请求堆积超过等待队列上限
  3. 调整Snowflake参数，加大锁等待队列上限、延长锁等待超时时间，可在sfOptions中追加如下配置：

  "sfsqlTransactionLockTimeout": "300", # 锁等待超时时间，单位秒
  "sfMaxConcurrentTransactions": "100" # 并发事务上限
  

• 大表场景优化
  如果写入的表数据量较大，可将表设置为分区表，写入时指定分区键，不同分区的写入操作不会互相抢占表锁，可大幅提升并发写入能力。

内容的提问来源于stack exchange，提问作者SAI BHARATH KOTHAKOTA