performance - 使用 Dapper 的批量插入花费的时间比预期的要长

Question

读完这篇文章后，我决定仔细看看我使用 Dapper 的方式。

我在一个空数据库上运行了这段代码

var members = new List<Member>();
for (int i = 0; i < 50000; i++)
{
    members.Add(new Member()
    {
        Username = i.toString(),
        IsActive = true
    });
}

using (var scope = new TransactionScope())
{
    connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members);

    scope.Complete();
}

大约花了20秒。那是 2500 次插入/秒。不错，但考虑到博客达到 45k 插入/秒，也不是很好。在 Dapper 中有没有更有效的方法来做到这一点？

另外，作为旁注，通过 Visual Studio 调试器运行此代码需要3 多分钟！我认为调试器会减慢它的速度，但看到这么多我真的很惊讶。

更新

所以这

using (var scope = new TransactionScope())
{
    connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members);

    scope.Complete();
}

还有这个

    connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members);

两者都花了 20 秒。

但这花了4秒！

SqlTransaction trans = connection.BeginTransaction();

connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members, transaction: trans);

trans.Commit();

score 84 · Accepted Answer

我能达到的最好成绩是使用这种方法在 4 秒内记录 50k 条记录

SqlTransaction trans = connection.BeginTransaction();

connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members, transaction: trans);

trans.Commit();

score 14 · Accepted Answer

我最近偶然发现了这一点，并注意到 TransactionScope 是在打开连接后创建的（我假设这是因为 Dappers Execute 不会打开连接，这与 Query 不同）。根据此处的答案 Q4：https ://stackoverflow.com/a/2886326/455904这不会导致 TransactionScope 处理连接。我的同事做了一些快速测试，在 TransactionScope 之外打开连接大大降低了性能。

因此更改为以下内容应该有效：

// Assuming the connection isn't already open
using (var scope = new TransactionScope())
{
    connection.Open();
    connection.Execute(@"
insert Member(Username, IsActive)
values(@Username, @IsActive)", members);

    scope.Complete();
}

score 4 · Accepted Answer

我创建了一个扩展方法，可以让您非常快速地进行批量插入。

public static class DapperExtensions
{
    public static async Task BulkInsert<T>(
        this IDbConnection connection,
        string tableName,
        IReadOnlyCollection<T> items,
        Dictionary<string, Func<T, object>> dataFunc)
    {
        const int MaxBatchSize = 1000;
        const int MaxParameterSize = 2000;

        var batchSize = Math.Min((int)Math.Ceiling((double)MaxParameterSize / dataFunc.Keys.Count), MaxBatchSize);
        var numberOfBatches = (int)Math.Ceiling((double)items.Count / batchSize);
        var columnNames = dataFunc.Keys;
        var insertSql = $"INSERT INTO {tableName} ({string.Join(", ", columnNames.Select(e => $"[{e}]"))}) VALUES ";
        var sqlToExecute = new List<Tuple<string, DynamicParameters>>();

        for (var i = 0; i < numberOfBatches; i++)
        {
            var dataToInsert = items.Skip(i * batchSize)
                .Take(batchSize);
            var valueSql = GetQueries(dataToInsert, dataFunc);

            sqlToExecute.Add(Tuple.Create($"{insertSql}{string.Join(", ", valueSql.Item1)}", valueSql.Item2));
        }

        foreach (var sql in sqlToExecute)
        {
            await connection.ExecuteAsync(sql.Item1, sql.Item2, commandTimeout: int.MaxValue);
        }
    }

    private static Tuple<IEnumerable<string>, DynamicParameters> GetQueries<T>(
        IEnumerable<T> dataToInsert,
        Dictionary<string, Func<T, object>> dataFunc)
    {
        var parameters = new DynamicParameters();

        return Tuple.Create(
            dataToInsert.Select(e => $"({string.Join(", ", GenerateQueryAndParameters(e, parameters, dataFunc))})"),
            parameters);
    }

    private static IEnumerable<string> GenerateQueryAndParameters<T>(
        T entity,
        DynamicParameters parameters,
        Dictionary<string, Func<T, object>> dataFunc)
    {
        var paramTemplateFunc = new Func<Guid, string>(guid => $"@p{guid.ToString().Replace("-", "")}");
        var paramList = new List<string>();

        foreach (var key in dataFunc)
        {
            var paramName = paramTemplateFunc(Guid.NewGuid());
            parameters.Add(paramName, key.Value(entity));
            paramList.Add(paramName);
        }

        return paramList;
    }
}

然后要使用此扩展方法，您将编写如下代码：

await dbConnection.BulkInsert(
    "MySchemaName.MyTableName",
    myCollectionOfItems,
    new Dictionary<string, Func<MyObjectToInsert, object>>
        {
            { "ColumnOne", u => u.ColumnOne },
            { "ColumnTwo", u => u.ColumnTwo },
            ...
        });

这是相当原始的，还有进一步改进的空间，例如传入事务或 commandTimeout 值，但它对我有用。

score 0 · Accepted Answer

我发现所有这些例子都不完整。

这是一些使用后正确关闭连接的代码，并且也正确使用事务范围来增强执行性能，基于此线程中更新和更好的答案。

using (var scope = new TransactionScope()) 
{
    Connection.Open();
    Connection.Execute(sqlQuery, parameters);

    scope.Complete();
}

score 0 · Accepted Answer

仅使用Execute一个插入语句的方法将永远不会进行批量插入或高效。即使是接受的答案 aTransaction也不做 a Bulk Insert。

如果要执行Bulk Insert，请使用SqlBulkCopy https://msdn.microsoft.com/en-us/library/system.data.sqlclient.sqlbulkcopy

你不会找到比这更快的东西。

小巧玲珑

免责声明：我是Dapper Plus项目的所有者

该项目不是免费的，但提供所有批量操作：

批量插入
批量更新
批量删除
批量合并

（在引擎盖下使用SqlBulkCopy）

还有更多选项，例如输出标识值：

// CONFIGURE & MAP entity
DapperPlusManager.Entity<Order>()
                 .Table("Orders")
                 .Identity(x => x.ID);

// CHAIN & SAVE entity
connection.BulkInsert(orders)
          .AlsoInsert(order => order.Items);
          .Include(x => x.ThenMerge(order => order.Invoice)
                         .AlsoMerge(invoice => invoice.Items))
          .AlsoMerge(x => x.ShippingAddress);

我们的库支持多个提供者：

SQL 服务器
SQL 紧凑型
甲骨文
mysql
PostgreSQL
SQLite
火鸟

score -4 · Accepted Answer

对我来说最快的变种：

            var dynamicParameters = new DynamicParameters();
            var selects = new List<string>();
            for (var i = 0; i < members.Length; i++)
            {
                var member = members[i];
                var pUsername = $"u{i}";
                var pIsActive = $"a{i}";
                dynamicParameters.Add(pUsername, member.Username);
                dynamicParameters.Add(pIsActive, member.IsActive);
                selects.Add("select @{pUsername},@{pIsActive}");
            }
            con.Execute($"insert into Member(Username, IsActive){string.Join(" union all ", selects)}", dynamicParameters);

生成如下的sql：

INSERT TABLENAME (Column1,Column2,...)
 SELECT @u0,@a0...
 UNION ALL
 SELECT @u1,@a1...
 UNION ALL
 SELECT @u2,@a2...

此查询工作得更快，因为 sql 添加一组行而不是一次添加 1 行。瓶颈不是写入数据，而是在日志中写入您正在执行的操作。

此外，查看最少记录事务的规则。

performance - 使用 Dapper 的批量插入花费的时间比预期的要长

6 回答 6

小巧玲珑

Related

Reference