16万条数据批量插入性能优化：IEnumerable转DataTable过慢求助

核心问题 #

需要处理16万条记录，从临时表取数后做字段缺失、重复校验，EF批量插入太慢改用SqlBulkCopy，但实体转DataTable耗时超20分钟（手动添加行也无改善），求优化方法或替代方案。

当前转换代码：

private static DataTable CreateRegisterDatatable(List<ApiRegisterData> apiRegisterData, List<Election> elections, IEnumerable<int> existingElectorIds, IEnumerable<int> missingElectionIds)
{
   var registerTable = new DataTable();

   foreach (var prop in typeof(Register).GetProperties())
   {
       DataColumn column = new DataColumn(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
       registerTable.Columns.Add(column);
   }

   foreach (var apiRegister in apiRegisterData)
   {
       if (!missingElectionIds.Contains(apiRegister.ElectionId) && !existingElectorIds.Contains(apiRegister.ElectorID))
       {
           DataRow row = registerTable.NewRow();
           row["ElectionID"] = elections[0].ElectionID;
           row["ElectorSuffix"] = apiRegister.ElectorSuffix.ToString();
           row["ElectorNumber"] = apiRegister.ElectorNumber;
           row["Title"] = string.IsNullOrEmpty(apiRegister.Title) ? DBNull.Value : apiRegister.Title;
           row["FirstName"] = string.IsNullOrEmpty(apiRegister.First_Name) ? DBNull.Value : apiRegister.First_Name;
           row["Initials"] = string.IsNullOrEmpty(apiRegister.Initials) ? DBNull.Value : apiRegister.Initials;
           row["Surname"] = string.IsNullOrEmpty(apiRegister.Surname) ? DBNull.Value : apiRegister.Surname;
           row["Suffix"] = string.IsNullOrEmpty(apiRegister.Suffix) ? DBNull.Value : apiRegister.Suffix;
           row["IneligibleReason"] = string.IsNullOrEmpty(apiRegister.IneligibleReason) ? DBNull.Value : apiRegister.IneligibleReason;
           row["Markers"] = string.IsNullOrEmpty(apiRegister.Markers) ? DBNull.Value : apiRegister.Markers;
           row["Nationality"] = string.IsNullOrEmpty(apiRegister.Nationality) ? DBNull.Value : apiRegister.Nationality;
           row["DOB"] = apiRegister.DOB is null ? DBNull.Value : apiRegister.DOB;
           row["DateOfAttainment"] = apiRegister.Date_Of_Attainment is null ? DBNull.Value : apiRegister.Date_Of_Attainment;
           row["OtherElector"] = Convert.ToBoolean(apiRegister.OtherElector);
           row["StreetName"] = string.IsNullOrEmpty(apiRegister.StreetName) ? DBNull.Value : apiRegister.StreetName;
           row["LocalityAddress1"] = string.IsNullOrEmpty(apiRegister.LocalityAddress1) ? DBNull.Value : apiRegister.LocalityAddress1;
           row["LocalityAddress2"] = string.IsNullOrEmpty(apiRegister.LocalityAddress2) ? DBNull.Value : apiRegister.LocalityAddress2;
           row["LocalityAddress3"] = string.IsNullOrEmpty(apiRegister.LocalityAddress3) ? DBNull.Value : apiRegister.LocalityAddress3;
           row["PostCode"] = string.IsNullOrEmpty(apiRegister.Postcode) ? DBNull.Value : apiRegister.Postcode;
           row["AbsentVoteAddress1"] = string.IsNullOrEmpty(apiRegister.ava1) ? DBNull.Value : apiRegister.ava1;
           row["AbsentVoteAddress2"] = string.IsNullOrEmpty(apiRegister.ava2) ? DBNull.Value : apiRegister.ava2;
           row["AbsentVoteAddress3"] = string.IsNullOrEmpty(apiRegister.ava3) ? DBNull.Value : apiRegister.ava3;
           row["AbsentVoteAddress4"] = string.IsNullOrEmpty(apiRegister.ava4) ? DBNull.Value : apiRegister.ava4;
           row["AbsentVoteAddress5"] = string.IsNullOrEmpty(apiRegister.ava5) ? DBNull.Value : apiRegister.ava5;
           row["AbsentVoteAddress6"] = string.IsNullOrEmpty(apiRegister.ava6) ? DBNull.Value : apiRegister.ava6;
           row["AbsentVotePostcode"] = string.IsNullOrEmpty(apiRegister.avap) ? DBNull.Value : apiRegister.avap;
           row["ProxyName"] = string.IsNullOrEmpty(apiRegister.ProxyName) ? DBNull.Value : apiRegister.ProxyName;
           row["CreatedDateTime"] = DateTime.UtcNow;
           row["UpdatedDateTime"] = DateTime.UtcNow;

           registerTable.Rows.Add(row);
        }
    }

    return registerTable;
}

优化方案 #

1. 修复Contains查询的性能瓶颈 #

当前代码里的missingElectionIds.Contains和existingElectorIds.Contains是最大性能杀手：IEnumerable<T>.Contains是O(n)复杂度，16万条循环里每次都要遍历整个集合，总复杂度是O(16万 * n)，完全不可接受。

改成HashSet<int>，查询复杂度降到O(1)：

// 提前把IEnumerable转成HashSet
var existingElectorIdSet = new HashSet<int>(existingElectorIds);
var missingElectionIdSet = new HashSet<int>(missingElectionIds);

// 循环里改成：
if (!missingElectionIdSet.Contains(apiRegister.ElectionId) && !existingElectorIdSet.Contains(apiRegister.ElectorID))

2. 预设置DataTable的容量 #

DataTable默认初始容量很小，添加大量行时会频繁扩容，浪费内存和时间。提前设置预估容量：

var registerTable = new DataTable();
// 预设置最小容量，避免多次扩容
registerTable.MinimumCapacity = apiRegisterData.Count;

3. 用列索引替代列名访问 #

row["ColumnName"]是字符串查找，每次都要遍历列集合找对应列。提前缓存列的索引，用索引直接访问：

// 初始化列后，缓存索引
var colIndices = new Dictionary<string, int>();
foreach (DataColumn col in registerTable.Columns)
{
    colIndices[col.ColumnName] = col.Ordinal;
}

// 循环里赋值时：
row[colIndices["ElectionID"]] = elections[0].ElectionID;
row[colIndices["ElectorSuffix"]] = apiRegister.ElectorSuffix.ToString();
// ... 其他字段同理

4. 用自定义IDataReader替代DataTable #

SqlBulkCopy支持直接读取IDataReader，不用先转DataTable，能省掉大量内存开销和转换时间。写一个简单的自定义Reader：

public class RegisterDataReader : DbDataReader
{
    private readonly IEnumerator<ApiRegisterData> _enumerator;
    private readonly HashSet<int> _missingElectionIds;
    private readonly HashSet<int> _existingElectorIds;
    private readonly int _electionId;
    private readonly DateTime _currentTime;

    public RegisterDataReader(List<ApiRegisterData> data, HashSet<int> missingElectionIds, HashSet<int> existingElectorIds, int electionId)
    {
        _enumerator = data.GetEnumerator();
        _missingElectionIds = missingElectionIds;
        _existingElectorIds = existingElectorIds;
        _electionId = electionId;
        _currentTime = DateTime.UtcNow;
    }

    // 实现必要的方法，核心是Read和GetValue
    public override bool Read()
    {
        // 跳过不符合条件的记录
        while (_enumerator.MoveNext())
        {
            var item = _enumerator.Current;
            if (!_missingElectionIds.Contains(item.ElectionId) && !_existingElectorIds.Contains(item.ElectorID))
            {
                return true;
            }
        }
        return false;
    }

    public override object GetValue(int ordinal)
    {
        var item = _enumerator.Current;
        return ordinal switch
        {
            0 => _electionId,
            1 => item.ElectorSuffix.ToString(),
            2 => item.ElectorNumber,
            3 => string.IsNullOrEmpty(item.Title) ? DBNull.Value : (object)item.Title,
            4 => string.IsNullOrEmpty(item.First_Name) ? DBNull.Value : (object)item.First_Name,
            // ... 对应其他列的逻辑，按顺序返回值
            28 => _currentTime,
            29 => _currentTime,
            _ => DBNull.Value
        };
    }

    // 必须实现的其他抽象方法（可以返回默认值）
    public override int FieldCount => 30; // 对应Register实体的属性数量
    public override string GetName(int ordinal) => 
        ordinal switch { 0 => "ElectionID", 1 => "ElectorSuffix", /* ... 其他列名 */ };
    public override Type GetFieldType(int ordinal) => 
        ordinal switch { 0 => typeof(int), 1 => typeof(string), /* ... 其他列类型 */ };
    public override bool IsDBNull(int ordinal) => GetValue(ordinal) == DBNull.Value;
    public override void Close() => _enumerator.Dispose();
    // 省略其他未实现的抽象方法（可抛出NotImplementedException或返回默认值）
}

使用时直接传给SqlBulkCopy：

var reader = new RegisterDataReader(apiRegisterData, missingElectionIdSet, existingElectorIdSet, elections[0].ElectionID);
using (var bulkCopy = new SqlBulkCopy(connectionString))
{
    bulkCopy.DestinationTableName = "Register";
    // 映射列（列名一致可省略，否则手动映射）
    bulkCopy.ColumnMappings.Add("ElectionID", "ElectionID");
    // ... 其他列映射
    bulkCopy.WriteToServer(reader);
}

5. 分批处理数据 #

一次性处理16万条会占用大量内存，导致GC频繁。分成每批1-5万条，处理一批插入一批：

int batchSize = 10000;
for (int i = 0; i < apiRegisterData.Count; i += batchSize)
{
    var batch = apiRegisterData.Skip(i).Take(batchSize).ToList();
    var dataTable = CreateRegisterDatatable(batch, elections, existingElectorIdSet, missingElectionIdSet);
    // 插入这批数据
    using (var bulkCopy = new SqlBulkCopy(connectionString))
    {
        bulkCopy.DestinationTableName = "Register";
        bulkCopy.WriteToServer(dataTable);
    }
}

6. 数据库层面做校验和插入 #

如果可以把校验逻辑移到数据库，效率会更高：

1. 把所有原始数据批量导入到一个临时表（用SqlBulkCopy直接导，不用客户端校验）
2. 用SQL语句做校验和插入，比如：

INSERT INTO Register (ElectionID, ElectorSuffix, ..., CreatedDateTime, UpdatedDateTime)
SELECT 
    @ElectionID, 
    ElectorSuffix, 
    -- 其他字段映射
    GETUTCDATE(),
    GETUTCDATE()
FROM #TempRegister
WHERE 
    ElectionID NOT IN (SELECT ElectionID FROM MissingElections)
    AND ElectorID NOT IN (SELECT ElectorID FROM ExistingElectors)
    -- 其他字段缺失校验，比如检查非空字段
    AND ElectorNumber IS NOT NULL
    AND Surname IS NOT NULL

这样客户端只需要负责把数据导入临时表，剩下的逻辑交给数据库，性能会提升很多。

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