MSSQL 2017：高效插入CSV大规模数据集至多表的优化方法

你的逐行调用存储过程+事务的方式慢是必然的——每一行都要走一次网络往返，还要处理事务的开启/提交开销，1万行1分钟的话，百万行确实要花太久。下面给你几个高效的解决方案，兼顾速度和主键重复的处理：

方案1：SqlBulkCopy导入临时表 + 数据库端批量分表（推荐，速度最快） #

这个方案把数据批量导入数据库临时表，再在数据库内部完成拆分和插入，几乎消除了网络往返的开销，是处理大规模数据的首选。

具体步骤： #

1. 创建临时/暂存表
   在SQL Server里创建一个和CSV结构完全匹配的暂存表（比如Staging_CSV_Data），用来临时存放所有CSV数据：

   CREATE TABLE Staging_CSV_Data (
       -- 字段和CSV一一对应，示例如下
       CSV_ID INT,
       Table1_Key INT,
       Table1_Col VARCHAR(50),
       Table2_Key INT,
       Table2_Col DATETIME,
       -- 其他对应四个表的字段
   )
   

2. 用SqlBulkCopy批量导入CSV
   在C#里用SqlBulkCopy直接把CSV数据（可以先读成DataTable或者用流式读取）导入暂存表，这一步速度极快，因为是批量传输：

   using (var bulkCopy = new SqlBulkCopy(connectionString))
   {
       bulkCopy.DestinationTableName = "Staging_CSV_Data";
       bulkCopy.BatchSize = 10000; // 可根据内存情况调整，1万-10万都可以试试
       bulkCopy.EnableStreaming = true; // 启用流式传输，减少内存占用
       // 映射CSV字段到暂存表字段（列名一致时可省略）
       bulkCopy.ColumnMappings.Add("CSV_ID", "CSV_ID");
       bulkCopy.ColumnMappings.Add("Table1_Key", "Table1_Key");
       // ... 完成其他字段映射
   
       // 假设你已经把CSV读成了DataTable csvDataTable
       bulkCopy.WriteToServer(csvDataTable);
   }
   

3. 数据库端批量拆分插入+处理主键重复
   写一个带事务的存储过程，把暂存表的数据拆分到四个目标表，用MERGE语句处理主键重复（比如直接忽略重复行）：

   CREATE PROCEDURE dbo.BulkInsertFromStaging
   AS
   BEGIN
       SET NOCOUNT ON;
       BEGIN TRANSACTION;
       BEGIN TRY
           -- 插入表1，忽略主键重复的行
           MERGE INTO Table1 t1
           USING (SELECT Table1_Key, Table1_Col FROM Staging_CSV_Data) s
           ON t1.Key = s.Table1_Key
           WHEN NOT MATCHED THEN
               INSERT (Key, Col) VALUES (s.Table1_Key, s.Table1_Col);
   
           -- 插入表2，同理处理主键重复
           MERGE INTO Table2 t2
           USING (SELECT Table2_Key, Table2_Col FROM Staging_CSV_Data) s
           ON t2.Key = s.Table2_Key
           WHEN NOT MATCHED THEN
               INSERT (Key, Col) VALUES (s.Table2_Key, s.Table2_Col);
   
           -- 重复上述逻辑处理表3和表4
   
           -- 清空暂存表，方便下次使用
           TRUNCATE TABLE Staging_CSV_Data;
   
           COMMIT TRANSACTION;
       END TRY
       BEGIN CATCH
           ROLLBACK TRANSACTION;
           THROW; -- 抛出异常让C#端处理
       END CATCH
   END
   

   调用这个存储过程就能一次性完成所有数据的拆分和插入，效率比逐行操作高几个数量级。

方案2：Table-Valued Parameters（TVPs）批量提交 #

如果必须在C#端完成数据拆分，可以用TVPs把批量数据传给存储过程，减少网络往返次数。

具体步骤： #

1. 创建用户定义表类型（UDTT）
   为每个目标表创建对应的UDTT：

   CREATE TYPE Table1_Type AS TABLE (
       Key INT,
       Col VARCHAR(50)
   );
   -- 同理创建Table2_Type、Table3_Type、Table4_Type
   

2. C#端批量拆分并传递TVPs
   在C#里把CSV行拆分后，按表分组，每批次比如1000行填充DataTable，然后作为参数传给存储过程：

   // 假设已经把CSV拆分到四个DataTable：table1Data, table2Data, table3Data, table4Data
   using (var conn = new SqlConnection(connectionString))
   {
       conn.Open();
       using (var cmd = new SqlCommand("dbo.BulkInsertWithTVPs", conn))
       {
           cmd.CommandType = CommandType.StoredProcedure;
           // 添加TVP参数
           cmd.Parameters.Add("@Table1Data", SqlDbType.Structured).Value = table1Data;
           cmd.Parameters.Add("@Table2Data", SqlDbType.Structured).Value = table2Data;
           cmd.Parameters.Add("@Table3Data", SqlDbType.Structured).Value = table3Data;
           cmd.Parameters.Add("@Table4Data", SqlDbType.Structured).Value = table4Data;
   
           cmd.ExecuteNonQuery();
       }
   }
   

3. 存储过程中批量插入+处理主键重复
   存储过程里用TVP作为数据源，同样用MERGE或者INSERT ... WHERE NOT EXISTS处理重复：

   CREATE PROCEDURE dbo.BulkInsertWithTVPs
       @Table1Data Table1_Type READONLY,
       @Table2Data Table2_Type READONLY,
       @Table3Data Table3_Type READONLY,
       @Table4Data Table4_Type READONLY
   AS
   BEGIN
       SET NOCOUNT ON;
       BEGIN TRANSACTION;
       BEGIN TRY
           MERGE INTO Table1 t1
           USING @Table1Data s ON t1.Key = s.Key
           WHEN NOT MATCHED THEN INSERT (Key, Col) VALUES (s.Key, s.Col);
   
           -- 处理其他三个表...
   
           COMMIT TRANSACTION;
       END TRY
       BEGIN CATCH
           ROLLBACK TRANSACTION;
           THROW;
       END CATCH
   END
   

额外优化建议 #

• 索引优化：插入前可以禁用目标表的非聚集索引，插入完成后再重建，减少插入时的索引维护开销（主键索引不能禁用，但非聚集索引可以）。
• 事务批次：如果数据量超大（比如千万级），可以把暂存表的数据分成多个批次处理，避免单次事务日志过大。
• 恢复模式：如果是测试环境或者允许数据丢失风险，把数据库恢复模式改成简单模式，可以大幅减少事务日志的生成，提升插入速度。
• 流式读取CSV：读取CSV时不要一次性加载到内存，用流式读取（比如CsvHelper的GetRecords<T>配合IDataReader），避免内存溢出。

内容的提问来源于stack exchange，提问作者Matěj Řehák