PySpark将3.1GB Delta表写入SQL Server耗时超9小时，求优化建议

一、集群资源与并行度优化 #

• 调整Spark并行度参数：当前集群总核数为8（2个Worker×4核），设置*spark.sql.shuffle.partitions和spark.default.parallelism*为16-24（核数的2-3倍），确保任务能充分利用集群并行能力，避免单分区任务阻塞。

  spark.conf.set("spark.sql.shuffle.partitions", "16")
  spark.conf.set("spark.default.parallelism", "16")
  

• 临时扩容Worker或升级实例规格：将Worker数量临时增加至3-4个，或更换为Standard_DS4_v2（8核28G）实例，提升整体计算资源，加快数据分片处理速度。
• 优化Executor配置：每个Worker分配1个Executor，设置*spark.executor.cores=3、spark.executor.memory=10g*（预留1核4G给系统进程），避免资源浪费。

二、JDBC写入参数优化 #

• 增大*bulkCopyBatchSize*：从10000调整至50000-100000，减少批次提交次数，降低网络交互开销（需确保SQL Server支持该批次大小，避免内存溢出）。
• 启用*truncate=true*：配合overwrite模式，使用截断表而非删除重建表的方式，保留表结构与索引框架，大幅减少前置操作耗时。
• 关闭*bulkCopyTableLock*：设置为false，避免独占表锁导致的写入阻塞，支持多分区并行写入。
• 添加*bulkCopyUseInternalTransaction=true*：让每个批次在内部事务中提交，减少事务管理的额外开销。

三、数据处理前置优化 #

• 调整DataFrame分区数：读取Delta表后，用repartition(16)将数据拆分为16个分区（每个分区约200MB），确保每个Executor能并行处理多个分片：

  df = spark.sql("select * from gold.Business_Txn").repartition(16)
  

• 仅读取必要字段：避免select *，只读取SQL Server表需要的字段，减少数据传输量与转换开销。
• 校验字段类型兼容性：确保Delta表字段类型与SQL Server目标表完全匹配（如Delta的string对应SQL Server的varchar），避免自动类型转换耗时。

四、SQL Server端优化 #

• 写入前禁用非聚集索引与约束：批量写入时维护索引会大幅拖慢速度，写入前执行以下SQL，完成后重建索引：

  -- 禁用索引
  ALTER INDEX ALL ON dbo.Business_Txn DISABLE;
  -- 写入完成后重建索引
  ALTER INDEX ALL ON dbo.Business_Txn REBUILD;
  

• 调整数据库恢复模式为简单模式：减少批量写入产生的日志量，降低IO开销，写入完成后可恢复为原模式。
• 检查SQL Server资源瓶颈：确认目标数据库所在服务器的CPU、内存、磁盘IO是否饱和，必要时升级存储（如Azure Premium Storage）或调整数据库配置。

优化后代码示例 #

# 配置Spark并行度
spark.conf.set("spark.sql.shuffle.partitions", "16")
spark.conf.set("spark.default.parallelism", "16")

# 读取Delta表并调整分区（仅保留需要的字段）
df = spark.sql("select col1, col2, ... from gold.Business_Txn").repartition(16)

try:
    # 配置JDBC基础参数
    jdbc_base_options = {
        "url": azure_sql_url,
        "user": username,
        "password": password
    }

    # 禁用SQL Server表索引
    disable_index_options = jdbc_base_options.copy()
    disable_index_options["dbtable"] = "(ALTER INDEX ALL ON dbo.Business_Txn DISABLE)"
    spark.read.format("jdbc").options(**disable_index_options).load()

    # 写入SQL Server
    df.write.format("com.microsoft.sqlserver.jdbc.spark")\
        .mode("overwrite")\
        .option("truncate", "true")\
        .options(**jdbc_base_options)\
        .option("dbtable", "dbo.Business_Txn")\
        .option("bulkCopyBatchSize", "50000")\
        .option("bulkCopyTableLock", "false")\
        .option("bulkCopyUseInternalTransaction", "true")\
        .option("bulkCopyTimeout", "6000000")\
        .save()

    # 重建索引
    rebuild_index_options = jdbc_base_options.copy()
    rebuild_index_options["dbtable"] = "(ALTER INDEX ALL ON dbo.Business_Txn REBUILD)"
    spark.read.format("jdbc").options(**rebuild_index_options).load()

    print("Table loaded Successfully ")
except Exception as e:
    print(f"Failed to load: {str(e)}")

内容的提问来源于stack exchange，提问作者Ramaraju.d