C#读取实时追加CSV并高效写入SQL Server的技术咨询

核心问题分析 #

你原来用的BULK INSERT有几个硬伤：一次性读取整个文件（文件越来越大时耗时爆炸）、没法指定导入列、同步执行阻塞UI导致无响应，而且没处理重复数据。推荐用.NET的SqlBulkCopy配合增量读取逻辑来解决，完全满足你的需求。

实现方案步骤 #

1. 跟踪CSV读取位置，只读新增内容

每次读取后记录文件的字节偏移量，下次从该位置开始读，避免重复处理已导入的行。可以把偏移量存在本地配置文件或数据库里。

2. 异步读取CSV，不阻塞UI

用StreamReader异步读取新增行，配合async/await让UI保持响应。

3. 指定列映射，只导入需要的列

通过SqlBulkCopy.ColumnMappings配置CSV列到SQL表列的对应关系，跳过不需要的列。

4. 批量插入+去重

先把新增数据批量插入临时表，再用MERGE语句将临时表中不存在的行插入正式表；或者给正式表加唯一约束，插入时忽略重复行。

完整示例代码 #

using System;
using System.IO;
using System.Data.SqlClient;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Windows.Forms;

public partial class YourForm : Form
{
    // 记录上次读取到的文件字节位置，可存在配置文件/数据库
    private long _lastReadPosition = 0;
    private readonly string _connectionString = @"Data Source=VKB_LAP_KIRAN\SQLEXPRESS; Initial Catalog=ExpoProCli; User ID=sa;Password=Kiran@123";
    private string _csvFilePath = string.Empty;

    private async void btnPlxl_Click(object sender, EventArgs e)
    {
        using (var openFileDialog = new OpenFileDialog())
        {
            if (openFileDialog.ShowDialog() != DialogResult.OK)
                return;

            _csvFilePath = openFileDialog.FileName;
            MessageBox.Show($"Selected file: {_csvFilePath}");
        }

        // 启动定时任务，每隔20秒读取一次新增内容（可根据你的追加频率调整）
        var timer = new System.Timers.Timer(20000);
        timer.Elapsed += async (s, args) => await ProcessNewCsvRowsAsync();
        timer.Start();

        // 首次执行一次
        await ProcessNewCsvRowsAsync();
    }

    private async Task ProcessNewCsvRowsAsync()
    {
        if (string.IsNullOrEmpty(_csvFilePath))
            return;

        var newRows = new List<string[]>();

        // 异步读取CSV新增行
        using (var stream = new FileStream(_csvFilePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
        using (var reader = new StreamReader(stream))
        {
            // 跳到上次读取的位置
            stream.Position = _lastReadPosition;

            string line;
            // 跳过表头（如果第一次读取已经处理过表头，这里要判断）
            if (_lastReadPosition == 0)
                await reader.ReadLineAsync();

            while ((line = await reader.ReadLineAsync()) != null)
            {
                // 分割CSV行（注意处理带引号的字段），这里可以用CsvHelper库更稳妥
                var columns = line.Split(',');
                newRows.Add(columns);
            }

            // 更新读取位置
            _lastReadPosition = stream.Position;
        }

        if (newRows.Count == 0)
            return;

        // 批量插入到SQL Server
        await BulkInsertWithDeduplicationAsync(newRows);
    }

    private async Task BulkInsertWithDeduplicationAsync(List<string[]> rows)
    {
        using (var conn = new SqlConnection(_connectionString))
        {
            await conn.OpenAsync();

            // 1. 创建临时表（结构和正式表一致，或者只包含需要的列）
            var createTempTableSql = @"
                IF OBJECT_ID('tempdb..#TempBOFile') IS NOT NULL
                    DROP TABLE #TempBOFile;
                CREATE TABLE #TempBOFile (
                    Column1 VARCHAR(100),
                    Column2 INT,
                    Column3 DATETIME
                    -- 这里只定义你需要导入的列，和CSV对应
                )";
            using (var cmd = new SqlCommand(createTempTableSql, conn))
            {
                await cmd.ExecuteNonQueryAsync();
            }

            // 2. 用SqlBulkCopy批量插入临时表，指定列映射
            using (var bulkCopy = new SqlBulkCopy(conn))
            {
                bulkCopy.DestinationTableName = "#TempBOFile";

                // 配置列映射：CSV索引 -> 数据库列名
                bulkCopy.ColumnMappings.Add(0, "Column1"); // CSV第1列对应数据库Column1
                bulkCopy.ColumnMappings.Add(2, "Column3"); // CSV第3列对应数据库Column3（跳过第2列）

                // 批量大小，根据内存调整，建议1000-5000
                bulkCopy.BatchSize = 2000;

                // 将List转换为DataTable或者用IDataReader，这里简单转DataTable
                var dataTable = new System.Data.DataTable();
                dataTable.Columns.Add("Column1");
                dataTable.Columns.Add("Column2");
                dataTable.Columns.Add("Column3");

                foreach (var row in rows)
                {
                    dataTable.Rows.Add(row[0], row[1], row[2]);
                }

                await bulkCopy.WriteToServerAsync(dataTable);
            }

            // 3. 用MERGE语句去重插入正式表
            var mergeSql = @"
                MERGE INTO tbl_BOFile AS Target
                USING #TempBOFile AS Source
                ON Target.UniqueKeyColumn = Source.UniqueKeyColumn -- 这里替换成你的唯一键列，用于判断重复
                WHEN NOT MATCHED THEN
                    INSERT (Column1, Column3) -- 只插入需要的列
                    VALUES (Source.Column1, Source.Column3);";

            using (var cmd = new SqlCommand(mergeSql, conn))
            {
                await cmd.ExecuteNonQueryAsync();
            }
        }

        MessageBox.Show($"成功导入 {rows.Count} 条新数据");
    }
}

关键优化点 #

• 增量读取：通过stream.Position跟踪已读位置，每次只处理新增内容，避免重复读取大文件
• 异步操作：所有IO和数据库操作都用async/await，不会阻塞UI线程
• 指定列导入：通过ColumnMappings精确控制哪些CSV列导入到数据库哪些列
• 去重逻辑：用临时表+MERGE语句，高效判断重复并只插入新数据
• 批量控制：设置BatchSize避免一次性加载过多数据到内存，平衡性能和内存占用

额外建议 #

• 用CsvHelper库替代手动分割CSV，处理带引号、换行的复杂CSV格式更可靠
• 给正式表的唯一键列加唯一索引，既能保证数据唯一，又能加快MERGE的匹配速度
• 关闭数据库的自动提交，在批量插入时用事务包裹，提升性能

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