C#高效读取含560万行的CSV文件方案咨询

嘿，针对你处理超大CSV并同步到SQL Server 2016的需求，我整理了几个高效的实现方案，兼顾性能和易用性，你可以根据自己的技术栈和环境来选：

1. C#流式读取 + SqlBulkCopy + MERGE（推荐，代码可控性强） #

因为CSV文件高达6.5GB，绝对不能一次性加载到内存，必须用流式处理，只提取需要的36列，再结合批量写入和数据库端的合并操作完成插入/更新。

步骤分解： #

• 流式读取CSV并筛选列：用轻量高效的CsvHelper库（或者自己写StreamReader逐行解析），只映射你需要的36列，避免加载冗余数据。
  示例代码片段：

  using (var stream = new FileStream("your_large_data.csv", FileMode.Open, FileAccess.Read, FileShare.Read))
  using (var reader = new StreamReader(stream))
  using (var csv = new CsvReader(reader, CultureInfo.InvariantCulture))
  {
      csv.Configuration.HasHeaderRecord = true;
      // 注册映射，只保留需要的36列
      csv.Configuration.RegisterClassMap<NPIDataMap>();
      
      var batch = new List<NPIDataEntity>();
      const int batchSize = 10000; // 可根据内存调整，1-5万都合适
  
      foreach (var record in csv.GetRecords<NPIDataEntity>())
      {
          batch.Add(record);
          if (batch.Count >= batchSize)
          {
              await BulkInsertToTempTable(batch);
              batch.Clear();
          }
      }
      // 处理剩余的零散记录
      if (batch.Count > 0) await BulkInsertToTempTable(batch);
  }
  

  其中NPIDataEntity只定义你需要的36个字段，NPIDataMap负责把CSV列名和实体属性一一对应。

• 用SqlBulkCopy写入临时表：创建一个和目标表结构匹配的临时表（或者只包含需要的字段），开启流式写入减少内存占用，批量提交数据。

  private async Task BulkInsertToTempTable(List<NPIDataEntity> records)
  {
      using (var conn = new SqlConnection("your_sql_connection_string"))
      {
          await conn.OpenAsync();
          using (var bulkCopy = new SqlBulkCopy(conn))
          {
              bulkCopy.DestinationTableName = "#TempNPIData";
              bulkCopy.BatchSize = 10000;
              bulkCopy.EnableStreaming = true; // 流式传输，降低内存消耗
              
              // 手动映射列（避免自动映射出错）
              bulkCopy.ColumnMappings.Add("NPI", "NPI");
              bulkCopy.ColumnMappings.Add("ColumnName1", "DBColumnName1");
              // 剩下34列的映射...
              
              await bulkCopy.WriteToServerAsync(records.AsDataReader());
          }
      }
  }
  

• MERGE语句完成插入/更新：临时表数据导入完成后，执行T-SQL的MERGE语句，根据NPI字段判断是插入新记录还是更新已有记录，这比逐行执行INSERT/UPDATE效率高几个量级。

  MERGE INTO dbo.TargetNPITable AS Target
  USING #TempNPIData AS Source
  ON Target.NPI = Source.NPI
  WHEN MATCHED THEN
      UPDATE SET 
          Target.Column1 = Source.Column1,
          Target.Column2 = Source.Column2,
          -- 其他需要更新的列...
  WHEN NOT MATCHED THEN
      INSERT (NPI, Column1, Column2, ...)
      VALUES (Source.NPI, Source.Column1, Source.Column2, ...);
  

  注意：确保目标表的NPI字段有主键或唯一索引，这会大幅提升MERGE的匹配效率。

2. 使用SSIS（SQL Server Integration Services） #

如果你的环境允许使用SSIS，这是ETL场景的原生解决方案，几乎不需要写代码，可视化配置即可完成：

• 操作步骤：
  • 新建SSIS包，添加「平面文件源」组件，配置时只勾选需要的36列。
  • 添加「OLE DB目标」组件，选择你的SQL Server 2016数据库，启用「快速加载」模式（底层就是SqlBulkCopy）。
  • 配置「查找」组件判断NPI是否存在，实现插入/更新逻辑；或者先把数据导入临时表，再执行一个SQL任务跑MERGE语句。
• 优势：SSIS针对大数据量做了优化，支持断点续传、错误捕获和日志记录，适合非代码优先的场景。

3. BCP工具 + T-SQL（命令行运维方案） #

如果不想写代码，也可以用SQL Server自带的BCP命令行工具，性能拉满：

• 操作步骤：
  • 先预处理CSV：用PowerShell或Python脚本提取需要的36列，生成一个精简版CSV（虽然还是大，但比原文件小很多）。
  • 用BCP把精简CSV导入临时表：

    bcp #TempNPIData in "trimmed_npi_data.csv" -S your_sql_server -d your_database -U username -P password -c -t, -r\n -b 10000
    

  • 最后执行和方案1一样的MERGE语句完成同步。
• 优势：BCP是原生工具，性能极高，适合运维人员或者熟悉命令行的场景。

关键性能优化点 #

• 绝对避免全量加载到内存：不管用哪种方案，都必须用流式/逐行处理，否则直接内存溢出。
• 合理设置批量大小：BatchSize建议在10000-50000之间，太小会增加数据库连接开销，太大会占用过多内存。
• 临时表优化：创建临时表时可以先禁用索引和约束，导入数据后再启用；如果需要跨会话访问，用全局临时表（##TempNPIData）。
• 数据库端优化：执行MERGE前可以禁用目标表的非必要索引，执行完再重建；如果数据库是简单恢复模式，批量插入的日志开销会大幅降低。

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