如何提升SQL代码中12万条数据Upsert操作的性能？

针对12万条数据的夜间同步场景，逐行循环执行Upsert的方式会因频繁网络往返、单条事务开销导致严重性能瓶颈，以下是几个更高效的实现方案：

1. 临时表+MERGE语句批量处理 #

这是大数据量同步最常用的优化方式，核心是先将所有数据批量导入临时表，再通过一条MERGE语句完成批量更新/插入，大幅减少数据库交互次数。

实现步骤： #

• 创建与目标表字段匹配的临时表
• 用SqlBulkCopy将12万条数据批量写入临时表（吞吐量远高于逐行插入）
• 执行MERGE语句，以Name为匹配条件完成批量更新和插入
• 清理临时表

示例代码： #

using (var connection = new SqlConnection(connectionString))
{
    connection.Open();

    // 1. 创建临时表
    var createTempTableSql = @"
        CREATE TABLE #TempExpandedGroup (
            ObjectGuid UNIQUEIDENTIFIER,
            Description NVARCHAR(MAX),
            IndexInFind INT,
            Name NVARCHAR(255),
            SchoolCode NVARCHAR(50),
            SchoolName NVARCHAR(255),
            WhenChanged DATETIME,
            WhenCreated DATETIME,
            CanExpanded BIT
        )";
    connection.Execute(createTempTableSql);

    // 2. 批量导入数据到临时表
    var dataTable = ConvertGroupsToDataTable(groups);
    using (var bulkCopy = new SqlBulkCopy(connection))
    {
        bulkCopy.DestinationTableName = "#TempExpandedGroup";
        // 字段映射（DataTable与临时表字段一致可省略）
        bulkCopy.ColumnMappings.Add("ObjectGuid", "ObjectGuid");
        bulkCopy.ColumnMappings.Add("Description", "Description");
        bulkCopy.ColumnMappings.Add("IndexInFind", "IndexInFind");
        bulkCopy.ColumnMappings.Add("Name", "Name");
        bulkCopy.ColumnMappings.Add("SchoolCode", "SchoolCode");
        bulkCopy.ColumnMappings.Add("SchoolName", "SchoolName");
        bulkCopy.ColumnMappings.Add("WhenChanged", "WhenChanged");
        bulkCopy.ColumnMappings.Add("WhenCreated", "WhenCreated");
        bulkCopy.ColumnMappings.Add("CanExpanded", "CanExpanded");
        
        bulkCopy.WriteToServer(dataTable);
    }

    // 3. 执行MERGE完成批量Upsert
    var mergeSql = @"
        MERGE INTO ExpandedGroupInformation AS Target
        USING #TempExpandedGroup AS Source
        ON Target.Name = Source.Name
        WHEN MATCHED THEN
            UPDATE SET 
                Description = Source.Description,
                IndexInFind = Source.IndexInFind,
                SchoolCode = Source.SchoolCode,
                SchoolName = Source.SchoolName,
                WhenChanged = Source.WhenChanged,
                WhenCreated = Source.WhenCreated,
                Deleted = 0,
                CanExpanded = Source.CanExpanded,
                StudentCount = 0,
                TeacherCount = 0,
                ParentCount = 0
        WHEN NOT MATCHED THEN
            INSERT (ObjectGuid, Description, IndexInFind, Name, SchoolCode, SchoolName, WhenChanged, WhenCreated, Deleted, CanExpanded, StudentCount, TeacherCount, ParentCount, Type)
            VALUES (Source.ObjectGuid, Source.Description, Source.IndexInFind, Source.Name, Source.SchoolCode, Source.SchoolName, Source.WhenChanged, Source.WhenCreated, 0, Source.CanExpanded, 0, 0, 0, 0);
        ";
    connection.Execute(mergeSql);

    // 4. 删除临时表
    connection.Execute("DROP TABLE #TempExpandedGroup");
}

// 辅助方法：将Groups集合转换为DataTable
private DataTable ConvertGroupsToDataTable(IEnumerable<Group> groups)
{
    var dt = new DataTable();
    dt.Columns.Add("ObjectGuid", typeof(Guid));
    dt.Columns.Add("Description", typeof(string));
    dt.Columns.Add("IndexInFind", typeof(int));
    dt.Columns.Add("Name", typeof(string));
    dt.Columns.Add("SchoolCode", typeof(string));
    dt.Columns.Add("SchoolName", typeof(string));
    dt.Columns.Add("WhenChanged", typeof(DateTime));
    dt.Columns.Add("WhenCreated", typeof(DateTime));
    dt.Columns.Add("CanExpanded", typeof(bool));

    foreach (var group in groups)
    {
        var row = dt.NewRow();
        row["ObjectGuid"] = group.ObjectGuid;
        row["Description"] = group.Description;
        row["IndexInFind"] = group.IndexInFind;
        row["Name"] = group.Name;
        row["SchoolCode"] = group.SchoolCode;
        row["SchoolName"] = group.SchoolName;
        row["WhenChanged"] = group.WhenChanged;
        row["WhenCreated"] = group.WhenCreated;
        row["CanExpanded"] = group.CanExpanded;
        dt.Rows.Add(row);
    }
    return dt;
}

2. 表值参数（Table-Valued Parameters）批量执行 #

如果不想依赖临时表，可以定义SQL Server表值类型，直接将数据集合作为参数传入执行批量Upsert。

实现步骤： #

• 在SQL Server中创建表值类型：

CREATE TYPE ExpandedGroupType AS TABLE (
    ObjectGuid UNIQUEIDENTIFIER,
    Description NVARCHAR(MAX),
    IndexInFind INT,
    Name NVARCHAR(255),
    SchoolCode NVARCHAR(50),
    SchoolName NVARCHAR(255),
    WhenChanged DATETIME,
    WhenCreated DATETIME,
    CanExpanded BIT
)

• 在代码中使用表值参数执行MERGE：

using (var connection = new SqlConnection(connectionString))
{
    connection.Open();
    var dataTable = ConvertGroupsToDataTable(groups); // 复用上面的转换方法
    
    var mergeSql = @"
        MERGE INTO ExpandedGroupInformation AS Target
        USING @GroupData AS Source
        ON Target.Name = Source.Name
        WHEN MATCHED THEN
            UPDATE SET 
                Description = Source.Description,
                IndexInFind = Source.IndexInFind,
                SchoolCode = Source.SchoolCode,
                SchoolName = Source.SchoolName,
                WhenChanged = Source.WhenChanged,
                WhenCreated = Source.WhenCreated,
                Deleted = 0,
                CanExpanded = Source.CanExpanded,
                StudentCount = 0,
                TeacherCount = 0,
                ParentCount = 0
        WHEN NOT MATCHED THEN
            INSERT (ObjectGuid, Description, IndexInFind, Name, SchoolCode, SchoolName, WhenChanged, WhenCreated, Deleted, CanExpanded, StudentCount, TeacherCount, ParentCount, Type)
            VALUES (Source.ObjectGuid, Source.Description, Source.IndexInFind, Source.Name, Source.SchoolCode, Source.SchoolName, Source.WhenChanged, Source.WhenCreated, 0, Source.CanExpanded, 0, 0, 0, 0);
        ";
    connection.Execute(mergeSql, new { GroupData = dataTable.AsTableValuedParameter("ExpandedGroupType") });
}

3. 辅助优化点 #

• 索引优化：确保目标表的Name字段有唯一索引或普通索引，MERGE时依赖该索引快速定位数据，避免全表扫描
• 批量事务管控：如果拆分批次处理（如每1万条一批），可为每个批次添加事务，既减少单条事务开销，又降低大事务的回滚风险
• 关闭自动提交：批量操作前关闭数据库自动提交模式，减少事务频繁创建和提交的额外开销

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