基于Databricks的PySpark写入Azure SQL数据库性能调优咨询

问题背景 #

我们的数据存储在Azure Data Lake Storage Gen2中，每日需处理约2500万条（30列）数据至Azure SQL Server数据库，使用Databricks及PySpark从Delta Table提取数据并加载至SQL表。目前以500万条数据样本测试ETL流程，平均耗时25分钟，寻求进一步优化方案以满足生产环境需求。

当前技术配置 #

Databricks集群 #

• 4个executor，每个4核、8GB内存
• Driver：8GB内存、4核
• Runtime版本：12.2 LTS（含Apache Spark 3.3.2）

SQL数据库 #

• 初始规格：标准S2（50 DTU），写入时DTU使用率达100%
• 升级至S3（100 DTU）后，500万条数据耗时降至21分钟，但DTU仍满负载

现有代码示例 #

dataframe = spark.table("hive_metastore.catalog.table") #5M records
dataframe_repartioned = dataframe.repartition(64)  

sql_host_name = f'sql_server_hostname'
sql_db_name = f'sql_server_database'
sql_user_id = "admin"
sql_server_pwd = "***"

jdbc_url = f'jdbc:sqlserver://{sql_host_name}:1433;databaseName={sql_db_name};user={sql_user_id};password={sql_server_pwd};'

dataframe_repartioned.write \
 .format("com.microsoft.sqlserver.jdbc.spark") \
 .option("truncate", "true") \
 .option("url", jdbc_url) \
 .option("dbtable", "schema.TABLE") \
 .option("tableLock", "true") \
 .option("batchsize", "10000") \
 .option("numPartitions", 64) \
 .mode("overwrite") \
 .save()

已尝试的优化及效果 #

• 调整SQL表nvarchar字段长度：将所有nvarchar(max)改为适配数据的长度（多为nvarchar(10)），处理时间从35分钟降至25分钟
• 重分区DataFrame：将单分区Delta表数据重分区为64个分区，仅减少45秒耗时
• 更换JDBC连接器：使用com.microsoft.sqlserver.jdbc.spark连接器，无明显效果
• 启用表锁：设置.option("tableLock", "true")，耗时减少1分钟
• 调整批处理大小：设置.option("batchsize", "10000")，无明显影响

除提升DTU外的优化方案 #

1. 优化Spark写入的分区策略 #

当前设置numPartitions=64但效果有限，需结合集群资源调整：

• 集群共4个executor，每个4核，理论并行度为4*4=16，64分区会导致任务过多、调度开销增大。建议将分区数调整为16-32，与集群并行度匹配，减少分区间的调度成本
• 避免使用repartition（全量 shuffle），改用coalesce减少分区（若数据无需重新分布），或基于Delta表的分区键进行repartitionByRange，减少数据 shuffle 开销

2. 优化JDBC写入参数 #

• 调整batchsize：尝试增大批处理大小至50000-100000，减少JDBC连接的交互次数。注意需结合SQL Server的max_allowed_packet配置（默认4MB），确保单批数据大小不超过该值
• 禁用自动提交：添加.option("isolationLevel", "READ_COMMITTED")和.option("autoCommit", "false")，减少事务提交的开销
• 使用bulkCopy模式：针对com.microsoft.sqlserver.jdbc.spark连接器，启用批量复制模式，直接调用SQL Server的BULK INSERT，效率远高于普通JDBC批量插入：

  dataframe_repartioned.write \
   .format("com.microsoft.sqlserver.jdbc.spark") \
   .option("url", jdbc_url) \
   .option("dbtable", "schema.TABLE") \
   .option("bulkCopyBatchSize", "50000") \
   .option("bulkCopyTableLock", "true") \
   .option("bulkCopyTimeout", "600") \
   .mode("overwrite") \
   .save()
  

3. 优化SQL Server端配置 #

• 临时禁用索引和约束：写入前禁用目标表的非聚集索引、外键约束，写入完成后重建。索引维护是大量数据写入时的主要开销：

  -- 禁用非聚集索引
  ALTER INDEX ALL ON schema.TABLE DISABLE;
  -- 禁用外键约束
  ALTER TABLE schema.TABLE NOCHECK CONSTRAINT ALL;
  
  -- 写入完成后重建索引
  ALTER INDEX ALL ON schema.TABLE REBUILD;
  -- 启用约束
  ALTER TABLE schema.TABLE CHECK CONSTRAINT ALL;
  

• 调整恢复模式：将数据库恢复模式临时改为BULK_LOGGED（仅在批量写入期间），减少日志生成量，写入完成后改回FULL模式保证可恢复性
• 使用弹性DTU：在ETL期间临时提升DTU峰值，完成后回落，平衡成本与性能

4. 数据预处理优化 #

• 过滤不必要字段：仅保留目标表需要的字段，减少数据传输量
• 对齐数据类型：确保Spark DataFrame的数据类型与SQL Server表严格匹配，避免自动类型转换开销（例如SparkStringType对应SQL Servernvarchar而非varchar）

5. 改用批量加载工具 #

• PolyBase：通过Databricks将数据写入ADLS Gen2的Parquet文件，再使用SQL Server的PolyBase直接批量导入，绕过JDBC性能瓶颈
• Azure Data Factory：使用ADF的Copy Activity直接从Delta Lake复制到SQL Server，ADF针对Azure服务做了深度优化，批量加载效率更高

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