批量插入在 Azure SQL Server 中无法正常工作答案

【问题标题】：Bulk insert is not working properly in Azure SQL Server批量插入在 Azure SQL Server 中无法正常工作
【发布时间】：2017-01-17 12:28:32
【问题描述】：

我无法使用 C# webapi 将大量数据插入 Azure SQL 服务器数据库

考虑

我想在 SQL 中插入 60K> 数据。在我的本地 sql server 中没有问题，但在 Azure SQL 中连接超时

我的方法：（所有都在本地sql server中工作，但不在Azure sql server中）

1) 尝试使用 EF 将其插入记录一一插入（对于 10000 大约 10 分钟，主要是超时）

2) 尝试使用 Bulk insert Extension 和 EF 3) 在 SqlBulkCopy 中尝试过

4) 尝试增加连接字符串中的连接超时时间

5) 尝试增加 Dbcontext 中的命令超时时间。

异常堆栈跟踪

Execution Timeout Expired.  The timeout period elapsed prior to completion of the operation or the server is not responding.
System.Data.SqlClient.SqlException (0x80131904): Execution Timeout Expired.  The timeout period elapsed prior to completion of the operation or the server is not responding. ---> System.ComponentModel.Win32Exception (0x80004005): The wait operation timed out
   at System.Data.SqlClient.SqlInternalConnection.OnError(SqlException exception, Boolean breakConnection, Action`1 wrapCloseInAction)
   at System.Data.SqlClient.TdsParser.ThrowExceptionAndWarning(TdsParserStateObject stateObj, Boolean callerHasConnectionLock, Boolean asyncClose)
   at System.Data.SqlClient.TdsParserStateObject.ReadSniError(TdsParserStateObject stateObj, UInt32 error)
   at System.Data.SqlClient.TdsParserStateObject.ReadSniSyncOverAsync()
   at System.Data.SqlClient.TdsParserStateObject.TryReadNetworkPacket()
   at System.Data.SqlClient.TdsParserStateObject.TryPrepareBuffer()
   at System.Data.SqlClient.TdsParserStateObject.TryReadByte(Byte& value)
   at System.Data.SqlClient.TdsParser.TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataReader dataStream, BulkCopySimpleResultSet bulkCopyHandler, TdsParserStateObject stateObj, Boolean& dataReady)
   at System.Data.SqlClient.TdsParser.Run(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataReader dataStream, BulkCopySimpleResultSet bulkCopyHandler, TdsParserStateObject stateObj)
   at System.Data.SqlClient.SqlBulkCopy.RunParser(BulkCopySimpleResultSet bulkCopyHandler)
   at System.Data.SqlClient.SqlBulkCopy.CopyBatchesAsyncContinuedOnSuccess(BulkCopySimpleResultSet internalResults, String updateBulkCommandText, CancellationToken cts, TaskCompletionSource`1 source)
   at System.Data.SqlClient.SqlBulkCopy.CopyBatchesAsyncContinued(BulkCopySimpleResultSet internalResults, String updateBulkCommandText, CancellationToken cts, TaskCompletionSource`1 source)
   at System.Data.SqlClient.SqlBulkCopy.CopyBatchesAsync(BulkCopySimpleResultSet internalResults, String updateBulkCommandText, CancellationToken cts, TaskCompletionSource`1 source)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServerInternalRestContinuedAsync(BulkCopySimpleResultSet internalResults, CancellationToken cts, TaskCompletionSource`1 source)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServerInternalRestAsync(CancellationToken cts, TaskCompletionSource`1 source)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServerInternalAsync(CancellationToken ctoken)
   at System.Data.SqlClient.SqlBulkCopy.WriteRowSourceToServerAsync(Int32 columnCount, CancellationToken ctoken)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServer(DataTable table, DataRowState rowState)

是否有任何解决方案或在 Azure 中更改任何配置？

更新

用于批量插入的代码

  using (var dbConnection = new DBModel().Database.Connection as SqlConnection)
                {
                    dbConnection?.Open();
                    using (var sqlBulkCopy = new SqlBulkCopy(dbConnection))
                    {
                        try
                        {
                            /* ColumnMapping
                             * Column is mapped to DB Column to DataTable Column
                             *
                             */
                            sqlBulkCopy.EnableStreaming = true;
                            sqlBulkCopy.BulkCopyTimeout = 500;
                            sqlBulkCopy.DestinationTableName = "LogTable";
                            //dt is object of the Datatable
                            sqlBulkCopy.WriteToServer(dt);
                        }
                        catch (Exception ex)
                        {

                        }
                    }


                }

【问题讨论】：

你的数据库层是什么？
您可以发布您在尝试使用 SqlBulkCopy 时使用的代码吗？您必须至少修改 BulkCopyTimeout 和 BatchSize 属性的默认值。
@TheGameiswar 它的标准
@JonathanMagnan 我也试过了，它不起作用
@Sreemat，您忘记添加批量大小。您当前正尝试一次插入 60,000k 行，这对于您当前的数据库层来说可能太多了。首先尝试设置一个非常低的 Batch Size（比如 100），看看这是否有效。

标签： c# azure azure-sql-database sqlbulkcopy ef-bulkinsert

【解决方案1】：

我建议您将sqlBulkCopy.BatchSize 设置为合理的数量，而不是一次性插入所有内容。根据您插入的数据，尝试从 10.000 开始，然后向上或向下工作，直到您对性能感到满意为止。

编辑一些额外的说明：当您考虑批量大小时，您需要考虑到 SqlBulkCopy 不仅需要插入数据，还需要读取和发送数据 - 最后一部分可能是它在本地 SQL 服务器上工作的原因，但不是Azure - 这也意味着，如果您使用的是大型数据集，您将需要使用较小的批量大小或相当高的 BulkCopyTimeout 设置，以允许每个批次有机会在达到超时限制之前完成。

您可以在这篇文章中了解有关批量大小的更多信息。 What is the recommended batch size for SqlBulkCopy?

其他选项：
我正在阅读此内容，这可能只是因为您的插入达到了关键的 DTU（数据库事务单元，基本上是服务器组合资源的度量）使用点。

性能级别经过校准和管理，可提供运行数据库工作负载所需的资源，达到所选服务层级/性能级别所允许的最大限制。如果您的工作负载达到 CPU/数据 IO/日志 IO 限制之一的限制，您将继续以最大允许级别接收资源，但您可能会看到查询的延迟增加 .这些限制不会导致任何错误，只会导致您的工作负载变慢，除非速度变慢以至于查询开始超时。

取自此链接：https://azure.microsoft.com/da-dk/blog/azure-sql-database-introduces-new-near-real-time-performance-metrics/
尝试在监控 DTU 使用情况的同时再次启动副本，并查看它是否在很长一段时间内保持 100%。如果是这种情况，您可能希望提高数据库的定价层级。

【讨论】：

你可以试试这个。 galleryserverpro.com/…
是的 @lmms90 我也有这个链接
@Sreemat - 我用第二个选项扩展了我的答案，也许会感兴趣
我的观察结果是 SqlBulkCopy 与最大化 DTU 将使 sql server 放弃超时错误。您可以将超时设置为明年的某个时间，这没关系；您仍然会收到超时错误，通常是在非常奇怪的时间（例如 28 秒）...

【解决方案2】：

我在过去两天一直在处理批量插入，这里有一个 通用批量插入类，它允许您排除一些列，如果发现 DbGeography 属性，它会将其转换为 SqlGeography：

 public class BulkInsert<T> where T : class
{
    #region Fields

    private readonly LoggingService _logger = new LoggingService(typeof(BulkInsert<T>));
    private string _connectionString;
    private string _tableName;
    private IEnumerable<string> _excludedPropertyNames;
    private int _batchSize;
    private IEnumerable<T> _data;
    private DataTable _dataTable;

    #endregion

    #region Constructor

    public BulkInsert(
        string connectionString,
        string tableName,
        IEnumerable<T> data,
        IEnumerable<string> excludedPropertyNames,
        int batchSize = 1000)
    {
        if (string.IsNullOrEmpty(connectionString)) throw new ArgumentNullException(nameof(connectionString));
        if (string.IsNullOrEmpty(tableName)) throw new ArgumentNullException(nameof(tableName));
        if (data == null) throw new ArgumentNullException(nameof(data));
        if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize));

        _connectionString = connectionString;
        _tableName = tableName;
        _batchSize = batchSize;
        _data = data;
        _excludedPropertyNames = excludedPropertyNames == null ? new List<string>() : excludedPropertyNames;
        _dataTable = CreateCustomDataTable();
    }

    #endregion

    #region Public Methods

    public void Insert()
    {
        using (var connection = new SqlConnection(_connectionString))
        {
            connection.Open();
            SqlTransaction transaction = connection.BeginTransaction();

            using (var bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.Default | SqlBulkCopyOptions.KeepIdentity, transaction))
            {
                bulkCopy.BatchSize = _batchSize;
                bulkCopy.DestinationTableName = _tableName;

                // Let's fix tons of mapping issues by
                // Setting the column mapping in SqlBulkCopy instance:
                foreach (DataColumn dataColumn in _dataTable.Columns)
                {
                    bulkCopy.ColumnMappings.Add(dataColumn.ColumnName, dataColumn.ColumnName);
                }

                try
                {
                    bulkCopy.WriteToServer(_dataTable);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex.Message);
                    transaction.Rollback();
                    connection.Close();
                }
            }

            transaction.Commit();
        }
    }

    #endregion

    #region Private Helper Methods

    private DataTable CreateCustomDataTable()
    {
        PropertyDescriptorCollection properties = TypeDescriptor.GetProperties(typeof(T));
        var table = new DataTable();
        foreach (PropertyDescriptor prop in properties)
        {
            // Just include the not excluded columns
            if (_excludedPropertyNames.All(epn => epn != prop.Name))
            {                  
                if (prop.PropertyType.Name == "DbGeography")
                {
                    var type = typeof(SqlGeography);
                    table.Columns.Add(prop.Name, type);
                }
                else
                {
                    table.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
                }
            }
        }
        foreach (T item in _data)
        {
            DataRow row = table.NewRow();
            foreach (PropertyDescriptor prop in properties)
            {
                // Just include the values in not excluded properties 
                if (_excludedPropertyNames.All(epn => epn != prop.Name))
                {
                    if (prop.PropertyType.Name == "DbGeography")
                    {                           
                        row[prop.Name] = SqlGeography.Parse(((DbGeography)prop.GetValue(item)).AsText()).MakeValid();
                    }
                    else
                    {
                        row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
                    }
                }
            }
            table.Rows.Add(row);
        }
        return table;
    }

    #endregion

}

好吧，它必须像这样使用：

var myEntityBulk = new BulkInsert<MyEntity>(
    _mYConnectionString,
     "MyEntities", 
      myEntities, 
      new[] { "ObjectState","NavigationPropertyOne", "NavigationPropertyTwo" }
);
myEntityBulk.Insert();

我希望它会有所帮助，我很确定它会...昨天我在几分钟内大量处理了它和 M 实体。

【讨论】：

【解决方案3】：

首先，如果只是向数据库中插入一条记录，请确认是否可以连接到 Azure SQL 数据库并进行操作而不会出现超时错误。

其次，请检查您是否使用ObjectContext上的CommandTimeout属性覆盖了默认超时值，您可以尝试将BulkCopyTimeout属性设置为0（表示没有限制）。

另外，请确保在您进行批量插入时它是否达到您的 Azure SQL 数据库的限制，您可以尝试change the service tier and performance level of your database。

【讨论】：

首先 - 是的，它连接到 Azure 我能够获取数据并插入少于 50K 的行，但如果超过 50K 就会超时。其次 - 厌倦了这个选项也不起作用
SqlBulkCopy 与 ObjectContext 和 Entity Framework 无关。 OP 应该首先设置一个小批量大小，例如 100 或 500。增加超时是另一个选项，将其设置为 0 虽然不是一个非常可扩展的选项 - 它只会冻结应用程序。可以增加批量大小以提高性能而不会导致超时

【解决方案4】：

使用 SqlBulkCopy 时，取决于您的 Web 服务器和 Azure SQL 服务器之间的 Route-Trip-Time（距离），您需要注意两个参数：

BatchSize（对我来说，我将批量大小设置为 500）

BulkCopyTimeOut：默认为 30 秒（我设置为 60 秒）

using (SqlBulkCopy bulkCopy = new SqlBulkCopy(connection)) {
        bulkCopy.BatchSize = array.Length;
        bulkCopy.DestinationTableName = tempTableName;
        bulkCopy.BulkCopyTimeout = 60;...}

使用此配置，我将 20K+ 事件从 EventStore 获取到我的 Projection MicroService 的整个过程，然后微服务将这 20K 事件更新/写入数据库大约需要 15 分钟（取决于事件存储之间的 RTT-路由行程时间和微服务，以及微服务和数据库之间）

【讨论】：