如何高效写入1.2GB/1400万行Spark DataFrame到MSSQL？当前方案耗时10小时+

现存方案性能瓶颈定位 #

• 原生Spark JDBC写入报错：未开启MSSQL JDBC驱动专属批量写入参数，默认单条提交、单分区传输，1400万行的单事务写入直接打满MSSQL事务日志，触发超时中断
• 逐行写入方案：df.rdd.toLocalIterator() 强制把全量数据拉到驱动节点单线程循环写入，哪怕开启fast_executemany，单连接单线程的写入吞吐量上限仅1000-2000行/秒，1400万行耗时正好在8-10小时区间，完全没有利用Spark的并行能力
• CSV+BULK INSERT方案：默认导出的CSV是碎片化的多小文件，逐文件执行导入时每次都要重新做格式校验、锁申请，没有合并大文件、开启最小日志模式，性能和逐行写入基本持平
• 单文件RDD导出思路：哪怕最终合并成单文件，整个写入链路还是单节点单线程传输，没有把并行能力延伸到数据库写入环节

高收益优化方案（按预期提速幅度排序） #

方案1：修正原生Spark JDBC配置，走MSSQL Bulk Copy协议 #

改动量最小，预期1400万行写入耗时2-5分钟
原JDBC写入代码仅做基础连接配置，缺失MSSQL驱动必需的批量写入参数，修正后的完整实现：

def FUNC_A1_OPTIMIZED(df): 
    start = time.time()
    jdbc_url = f"jdbc:sqlserver://{config.get('mydb','host')}:1434;database={config.get('mydb', 'database')}"
    df.select("F1", "F2", "F3", "F4", "F5", "F6", "F7", "F8", "F9") \
      .write.format("jdbc").mode("overwrite") \
      .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")\
      .option("url", jdbc_url) \
      .option("dbtable", "dbo.tblAFS") \
      .option("user", config.get('mydb', 'username'))\
      .option("password", config.get('mydb', 'password'))\
      # 核心性能优化参数
      .option("batchsize", "10000") \
      .option("isolationLevel", "READ_UNCOMMITTED") \
      .option("rewriteBatchedStatements", "true") \
      .option("truncate", "true") \
      .option("tableLock", "true") \
      .option("bulkCopyBatchSize", "100000") \
      .option("bulkCopyTimeout", "600") \
      .save()
    print(f"写入总耗时：{round(time.time()-start, 2)} 秒")

参数作用：

• bulkCopyBatchSize 控制单批次10万行传输，替代默认100行的单批次提交
• tableLock 写入时对目标表加表锁，消除行锁、页锁的额外开销
• truncate 替代overwrite模式下逐行删除表数据的元数据操作
• 开启批量写入后会自动走MSSQL原生Bulk Copy通道，和命令行bcp工具的传输效率完全一致，彻底解决原FUNC_A1的报错问题

方案2：CSV导出优化+最小日志模式BULK INSERT #

如果JDBC驱动版本过低不支持批量参数，可走该方案，预期写入耗时3-7分钟

1. CSV导出阶段提前合并分区，把碎片化的小文件合并为2个不超过1GB的大文件，避免逐文件导入的额外开销：

   df.coalesce(2).write.mode("overwrite") \
     .options(header=True, delimiter="|", escape='"') \
     .csv(file_path)
   

2. BULK INSERT导入前调整库的恢复模式，开启最小日志写入，替代默认的全量事务日志记录：

   -- 导入前切为大容量日志模式，大幅减少事务日志写入量
   ALTER DATABASE [目标库名] SET RECOVERY BULK_LOGGED;
   GO
   BULK INSERT dbo.tblAFS FROM '/文件存储路径/part-*.csv' 
   WITH (
       FORMAT = 'CSV', 
       FIRSTROW = 2,
       TABLOCK,
       CHECK_CONSTRAINTS = OFF,
       BATCHSIZE = 100000,
       MAXERRORS = 10
   );
   GO
   -- 导入完成后切回完整恢复模式
   ALTER DATABASE [目标库名] SET RECOVERY FULL;
   GO
   

• 注意：不要保留大量带表头的小CSV文件，逐文件跳表头、申请锁的开销会让整体性能下降90%以上。

方案3：修正逐行写入逻辑为Executor端并行写入 #

如果必须使用pyodbc实现写入，该方案比原逐行写入快8倍，预期耗时1小时以内
原方案的核心问题是把数据全部拉到驱动端单线程处理，调整逻辑为把写入操作下发到每个Spark Executor端，利用Spark的分区并行能力多连接同时写入：

def parallel_write_partition(rows):
    import pyodbc
    # 每个分区单独创建数据库连接，避免连接竞争
    conn = pyodbc.connect(shared.get_odbcconn(config))
    cursor = conn.cursor()
    cursor.fast_executemany = True
    insert_sql = """
           INSERT INTO dbo.tblAFS 
           ([F1],[F2],[F3],[F4],[F5],[F6],[F7],[F8],[F9])
           VALUES (?,?,?,?,?,?,?,?,?)
    """
    batch_buffer = []
    for row in rows:
        batch_buffer.append((row["F1"], row["F2"], row["F3"], row["F4"], row["F5"],
                 row["F6"], row["F7"], row["F8"], row["F9"]))
        # 每累计1万行提交一次，减少事务交互次数
        if len(batch_buffer) >= 10000:
            cursor.executemany(insert_sql, batch_buffer)
            conn.commit()
            batch_buffer = []
    # 提交分区内剩余的行
    if batch_buffer:
        cursor.executemany(insert_sql, batch_buffer)
        conn.commit()
    conn.close()
    return []

# 按8个分区并行写入，匹配MSSQL 8vCPU的处理能力
df.repartition(8).rdd.mapPartitions(parallel_write_partition).collect()

避坑提示 #

• 写入前不要提前创建非聚簇索引、开启触发器，否则写入耗时会上涨10倍以上，所有索引、约束等数据全量导入完成后再创建
• 连接字符串不要强制开启未优化的SSL加密，会让传输吞吐量下降70%
• 写入分区数不要超过MSSQL服务端的最大工作线程数，8vCPU的MSSQL服务端最多并行处理16个写入连接，开太多分区反而会造成锁竞争
• 不要使用rdd.collect()把全量数据拉到驱动内存做写入操作，极易触发OOM，也会丢失并行能力

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