SSIS脚本任务中从List<T>更新数据表的最优方案探讨

我在多个SSIS项目里处理过类似的场景——从自定义类库拿到List数据后做Upsert到数据库。逐行遍历执行MERGE语句的方式虽然简单，但数据量稍大就会暴露出性能问题（大量数据库往返、小事务堆积）。下面给你推荐两种工业级的最优方案：

方案一：SqlBulkCopy + 临时表 + MERGE（大数据量首选） #

这种方式的核心是先把List批量导入临时表，再用一条MERGE语句完成Upsert，只需要2次数据库往返（批量插入+MERGE），性能比逐行操作提升几个数量级。

步骤&代码示例 #

1. 把List转换成DataTable（SqlBulkCopy需要DataTable/IDataReader作为数据源）：

private DataTable ConvertListToDataTable<T>(List<T> list)
{
    DataTable dt = new DataTable();
    var props = typeof(T).GetProperties(BindingFlags.Public | BindingFlags.Instance);
    
    // 构建DataTable列
    foreach (var prop in props)
    {
        dt.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
    }
    
    // 填充数据行
    foreach (var item in list)
    {
        DataRow row = dt.NewRow();
        foreach (var prop in props)
        {
            row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
        }
        dt.Rows.Add(row);
    }
    return dt;
}

2. Script Task主逻辑：

public void Main()
{
    // 从共享类库获取数据List
    var processDataList = YourSharedLibrary.GetProcessRelatedData(); // 替换为你的实际方法
    
    if (processDataList == null || processDataList.Count == 0)
    {
        Dts.TaskResult = (int)ScriptResults.Success;
        return;
    }

    // 转换为DataTable
    DataTable dataTable = ConvertListToDataTable(processDataList);
    
    // 获取SSIS连接管理器的连接字符串
    string connString = Dts.Connections["YourDBConnection"].ConnectionString; // 替换为你的连接名

    using (SqlConnection conn = new SqlConnection(connString))
    {
        conn.Open();
        // 开启事务确保原子性（可选，根据业务需求）
        using (SqlTransaction tran = conn.BeginTransaction())
        {
            try
            {
                // 1. 创建临时表（结构和目标表完全匹配）
                string createTempSql = @"
                    CREATE TABLE #TempProcessData (
                        Id INT PRIMARY KEY,
                        ProcessName VARCHAR(100),
                        ExecuteTime DATETIME,
                        Status VARCHAR(20),
                        -- 其他字段和目标表一致
                    )";
                using (SqlCommand cmd = new SqlCommand(createTempSql, conn, tran))
                {
                    cmd.ExecuteNonQuery();
                }

                // 2. 批量插入到临时表
                using (SqlBulkCopy bulkCopy = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, tran))
                {
                    bulkCopy.DestinationTableName = "#TempProcessData";
                    bulkCopy.BatchSize = 2000; // 根据数据量调整，一般1000-5000都可以
                    bulkCopy.WriteToServer(dataTable);
                }

                // 3. 执行MERGE做Upsert
                string mergeSql = @"
                    MERGE INTO dbo.ProcessData AS Target
                    USING #TempProcessData AS Source
                    ON Target.Id = Source.Id
                    WHEN MATCHED THEN
                        UPDATE SET 
                            Target.ProcessName = Source.ProcessName,
                            Target.ExecuteTime = Source.ExecuteTime,
                            Target.Status = Source.Status
                    WHEN NOT MATCHED THEN
                        INSERT (Id, ProcessName, ExecuteTime, Status)
                        VALUES (Source.Id, Source.ProcessName, Source.ExecuteTime, Source.Status);
                ";
                using (SqlCommand cmd = new SqlCommand(mergeSql, conn, tran))
                {
                    cmd.ExecuteNonQuery();
                }

                tran.Commit();
            }
            catch (Exception ex)
            {
                tran.Rollback();
                Dts.Events.FireError(0, "UpsertDataError", ex.Message, string.Empty, 0);
                Dts.TaskResult = (int)ScriptResults.Failure;
                return;
            }
        }
    }

    Dts.TaskResult = (int)ScriptResults.Success;
}

方案二：表值参数(TVP) + 存储过程（中等数据量+易维护首选） #

如果数据量不是特别大（几千到几万条），用表值参数把List传给存储过程，在存储过程里做MERGE，代码更整洁，逻辑封装在数据库层，后续维护更方便。

步骤&代码示例 #

1. 在数据库中创建自定义表类型：

CREATE TYPE dbo.ProcessDataTableType AS TABLE (
    Id INT PRIMARY KEY,
    ProcessName VARCHAR(100),
    ExecuteTime DATETIME,
    Status VARCHAR(20)
)

2. 创建Upsert存储过程：

CREATE PROCEDURE dbo.UpsertProcessData
    @ProcessData dbo.ProcessDataTableType READONLY
AS
BEGIN
    SET NOCOUNT ON;
    
    MERGE INTO dbo.ProcessData AS Target
    USING @ProcessData AS Source
    ON Target.Id = Source.Id
    WHEN MATCHED THEN
        UPDATE SET 
            Target.ProcessName = Source.ProcessName,
            Target.ExecuteTime = Source.ExecuteTime,
            Target.Status = Source.Status
    WHEN NOT MATCHED THEN
        INSERT (Id, ProcessName, ExecuteTime, Status)
        VALUES (Source.Id, Source.ProcessName, Source.ExecuteTime, Source.Status);
END

3. Script Task主逻辑：

public void Main()
{
    var processDataList = YourSharedLibrary.GetProcessRelatedData();
    
    if (processDataList == null || processDataList.Count == 0)
    {
        Dts.TaskResult = (int)ScriptResults.Success;
        return;
    }

    DataTable dataTable = ConvertListToDataTable(processDataList);
    string connString = Dts.Connections["YourDBConnection"].ConnectionString;

    using (SqlConnection conn = new SqlConnection(connString))
    {
        conn.Open();
        using (SqlCommand cmd = new SqlCommand("dbo.UpsertProcessData", conn))
        {
            cmd.CommandType = CommandType.StoredProcedure;
            
            // 添加表值参数
            SqlParameter tvpParam = cmd.Parameters.AddWithValue("@ProcessData", dataTable);
            tvpParam.SqlDbType = SqlDbType.Structured;
            tvpParam.TypeName = "dbo.ProcessDataTableType"; // 必须和数据库表类型名称一致

            try
            {
                cmd.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                Dts.Events.FireError(0, "UpsertDataError", ex.Message, string.Empty, 0);
                Dts.TaskResult = (int)ScriptResults.Failure;
                return;
            }
        }
    }

    Dts.TaskResult = (int)ScriptResults.Success;
}

方案对比&注意事项 #

• 逐行遍历的弊端：每次循环都要发送SQL请求，数据库要处理大量小事务，当List数据量超过500条时，延迟会急剧增加，不推荐使用。
• SqlBulkCopy+临时表：适合上万条以上的大数据量，批量插入速度极快，MERGE一次性完成操作，性能最优。
• TVP+存储过程：代码更简洁，逻辑集中在存储过程，便于后续修改业务规则，中等数据量下效率也很高。

额外注意：

• 确保SSIS连接管理器的权限足够创建临时表/执行存储过程。
• List转DataTable时，要保证实体类的属性类型和数据库表字段类型完全匹配，避免转换错误。
• 事务控制：如果需要保证数据一致性，一定要把批量插入和MERGE放在同一个事务里，避免部分成功部分失败的情况。

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