0

出于某种疯狂的原因，我在将数据以合理的块发送到 SQL 时遇到了 OutOfMemoryException，并且几乎没有使用任何内存：

System.OutOfMemoryException: Exception of type 'System.OutOfMemoryException' was thrown.
   at System.Data.DataTable.NewRowArray(Int32 size)
   at System.Data.RecordManager.GrowRecordCapacity()
   at System.Data.RecordManager.NewRecordBase()
   at System.Data.DataTable.NewRecord(Int32 sourceRecord)
   at Company.PA.Data.PADbContext.d__22`1.MoveNext() in D:\Agent_A\_work\7\s\Company.PA.DataLayer\Company.PA.Data\BulkInsert\StreamedSqlBulkCopy.cs:line 46

在下面的 while 循环中调用时发生错误dataTable.NewRow()，一旦我超过第 30 百万行：

/// <summary>Helper to stream a large number of records into SQL without
/// ever having to materialize the entire enumerable into memory at once.</summary>
/// <param name="destinationTableName">The name of the table in the database to copy data to.</param>
/// <param name="dataTable">A new instance of the DataTable class that matches the schema of the table to insert to.
/// This should match exactly (same column names) what is in SQL, for automatic column mapping to work.</param>
/// <param name="sourceData">The enumerable of data that will be used to generate DataRows</param>
/// <param name="populateRow">A delegate function that populates and returns a new data row for a given record.</param>
/// <param name="memoryBatchSize">The number of DataRows to generate in memory before passing them to SqlBulkCopy</param>
/// <param name="insertBatchSize">The batch size of inserts performed by SqlBulkCopy utility.</param>
public async Task StreamedSqlBulkCopy<T>(
    string destinationTableName, DataTable dataTable,
    IEnumerable<T> sourceData, Func<T, DataRow, DataRow> populateRow,
    int memoryBatchSize = 1000000, int insertBatchSize = 5000)
{
    using (SqlConnection connection = new SqlConnection(Database.Connection.ConnectionString))
    {
        connection.Open();
        using (SqlBulkCopy bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.TableLock, null))
        using (IEnumerator<T> enumerator = sourceData.GetEnumerator())
        {
            // Configure the single SqlBulkCopy instance that will be used to copy all "batches"
            bulkCopy.DestinationTableName = destinationTableName;
            bulkCopy.BatchSize = insertBatchSize;
            bulkCopy.BulkCopyTimeout = _bulkInsertTimeOut;
            foreach (DataColumn column in dataTable.Columns)
                bulkCopy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
            // Begin enumerating over all records, preparing batches no larger than "memoryBatchSize"
            bool hasNext = true;
            while (hasNext)
            {
                DataRow[] batch = new DataRow[memoryBatchSize];
                int filled = 0;
                while ((hasNext = enumerator.MoveNext()) && filled < memoryBatchSize)
                    batch[filled++] = populateRow(enumerator.Current, dataTable.NewRow());
                // When we reach the end of the enumerable, we need to shrink the final buffer array
                if (filled < memoryBatchSize)
                    Array.Resize(ref batch, filled);
                await bulkCopy.WriteToServerAsync(batch);
            }
        }
    }
}

希望很清楚，上述帮助程序的目的是IEnumerable<T>使用读取器和委托将（非常大的）数据流式传输到 SQL 表SqlBulkCopy，该委托将为给定元素填充一行。

示例用法是：

public async Task SaveExchangeRates(List<FxRate> fxRates)
{
    var createDate = DateTimeOffset.UtcNow;
    await StreamedSqlBulkCopy("RefData.ExchangeRate",
        GetExchangeRateDataTable(), fxRates, (fx, newRow) =>
        {
            newRow["BaseCurrency"] = "USD";
            newRow["TargetCurrency"] = fx.CurrencyCode;
            newRow["ExchangeDate"] = fx.ExchangeRateDate;
            newRow["DollarValue"] = fx.ValueInUsd;
            return newRow;
        });
}

private DataTable GetExchangeRateDataTable()
{
    var dataTable = new DataTable();
    dataTable.Columns.Add("ExchangeDate", typeof(DateTime));
    dataTable.Columns.Add("BaseCurrency", typeof(string));
    dataTable.Columns.Add("TargetCurrency", typeof(string));
    dataTable.Columns.Add("DollarValue", typeof(double));
    return dataTable;
}
4

2 回答 2

0

事实证明，即使您只是DataTable出于架构目的将实例用作空结构，并且即使您从未调用dataTable.Rows.Add()将其实际添加到表中，但每次调用时它都会在其中增加一个计数器，NewRow并且显然甚至会增加一个占位符数组，它希望您最终插入所有这些行？

无论如何，解决方法是通过使用自身的克隆覆盖它来定期“重置”您的模板：

dataTable = dataTable.Clone();

可以肯定的是不优雅，但比尝试实现自己的更容易，IDataReader这是利用 SQLBulkCopy 的唯一其他方法。（也就是说 - 对于任何其他试图流式传输到 SQL 批量复制的人，他们不受限制避免像我一样避免使用 3rd 方库，请查看FastMember包中的 Marc Gravel 的 ObjectReader 和这个答案：https ://stackoverflow.com/a /47208127/529618 )

于 2021-03-11T00:47:09.477 回答
0

另一种简化的方法（但以额外开销为代价）是接受我们的命运并使用DataTable类而不是数组DataRow- 但Clone()定期创建原始表的副本以避免明显的硬最大限制 16,777,216 行。

我不喜欢DataTable为您使用它创建的所有行维护一个数组，即使它们最终没有被添加 - 所以我们不妨利用而不是分配我们自己的。

使用 a 的一些开销DataTable可以通过设置其初始容量以确保它不会增长（内存分配）并禁用尽可能多的事件来抵消：

相关变更如下：

bool hasNext = true;
while (hasNext)
{
    using (DataTable tableChunk = dataTable.Clone())
    {
        tableChunk.MinimumCapacity = memoryBatchSize + 1; // Avoid triggering resizing
        tableChunk.BeginLoadData(); // Speeds up inserting a large volume of rows a little
        int filled = 0;
        while ((hasNext = enumerator.MoveNext()) && filled++ < memoryBatchSize)
            tableChunk.Rows.Add(populateRow(enumerator.Current, tableChunk.NewRow()));
        await bulkCopy.WriteToServerAsync(tableChunk);
    }
}
于 2021-03-11T01:24:18.940 回答