Big-Data-Batch-Einfügung mehrerer Datenbanken in c# (SqlServer, Oracle, SQLite und MySql)

In der Vergangenheit wusste ich nur, dass SqlServer das Batch-Einfügen von Daten unterstützt. Allerdings muss Oracle den Orace.DataAccess-Treiber verwenden Einfügungslösungen für Datenbanken.

Zuallererst gibt es in IProvider eine Plug-in-Dienstschnittstelle IBatcherProvider für die Stapeleinfügung. Diese Schnittstelle wurde im vorherigen Artikel erwähnt.

/// <summary>
  /// 提供数据批量处理的方法。
  /// </summary>
  public interface IBatcherProvider : IProviderService
  {
    /// <summary>
    /// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
    /// </summary>
    /// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
    /// <param name="batchSize">每批次写入的数据量。</param>
    void Insert(DataTable dataTable, int batchSize = 10000);
  }

1. Batch-Einfügung von SqlServer-Daten ist sehr einfach, verwenden Sie einfach SqlBulkCopy. Das Folgende ist die Implementierung dieser Klasse:

Oben Wenn keine Transaktionen verwendet werden, hat die Verwendung von Transaktionen einen gewissen Einfluss auf die Leistung. Wenn Sie Transaktionen verwenden möchten, können Sie SqlBulkCopyOptions.UseInternalTransaction festlegen.

/// <summary>
  /// 为 System.Data.SqlClient 提供的用于批量操作的方法。
  /// </summary>
  public sealed class MsSqlBatcher : IBatcherProvider
  {
    /// <summary>
    /// 获取或设置提供者服务的上下文。
    /// </summary>
    public ServiceContext ServiceContext { get; set; }
 
    /// <summary>
    /// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
    /// </summary>
    /// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
    /// <param name="batchSize">每批次写入的数据量。</param>
    public void Insert(DataTable dataTable, int batchSize = 10000)
    {
      Checker.ArgumentNull(dataTable, "dataTable");
      if (dataTable.Rows.Count == 0)
      {
        return;
      }
      using (var connection = (SqlConnection)ServiceContext.Database.CreateConnection())
      {
        try
        {
          connection.TryOpen();
          //给表名加上前后导符
          var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider>(), dataTable.TableName);
          using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)
            {
              DestinationTableName = tableName, 
              BatchSize = batchSize
            })
          {
            //循环所有列，为bulk添加映射
            dataTable.EachColumn(c => bulk.ColumnMappings.Add(c.ColumnName, c.ColumnName), c => !c.AutoIncrement);
            bulk.WriteToServer(dataTable);
            bulk.Close();
          }
        }
        catch (Exception exp)
        {
          throw new BatcherException(exp);
        }
        finally
        {
          connection.TryClose();
        }
      }
    }
  }

2. Oracle-Daten-Batch-Einfügung

System.Data.OracleClient unterstützt keine Batch-Einfügung, daher können Sie nur die Oracle.DataAccess-Komponente als Anbieter verwenden.

Der wichtigste Schritt oben besteht darin, die DataTable in eine Array-Darstellung eines Arrays umzuwandeln, d. h. object[][]. Letzteres Array ist die Anzahl der Zeilen, daher verwendet die Schleife Columns das letzte Array als Parameterwert, d. h. der Wert des Parameters ist ein Array. Die INSERT-Anweisung unterscheidet sich nicht von einer normalen INSERT-Anweisung.

/// <summary>
  /// Oracle.Data.Access 组件提供的用于批量操作的方法。
  /// </summary>
  public sealed class OracleAccessBatcher : IBatcherProvider
  {
    /// <summary>
    /// 获取或设置提供者服务的上下文。
    /// </summary>
    public ServiceContext ServiceContext { get; set; }
 
    /// <summary>
    /// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
    /// </summary>
    /// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
    /// <param name="batchSize">每批次写入的数据量。</param>
    public void Insert(DataTable dataTable, int batchSize = 10000)
    {
      Checker.ArgumentNull(dataTable, "dataTable");
      if (dataTable.Rows.Count == 0)
      {
        return;
      }
      using (var connection = ServiceContext.Database.CreateConnection())
      {
        try
        {
          connection.TryOpen();
          using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())
          {
            if (command == null)
            {
              throw new BatcherException(new ArgumentException("command"));
            }
            command.Connection = connection;
            command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);
            command.ExecuteNonQuery();
          }
        }
        catch (Exception exp)
        {
          throw new BatcherException(exp);
        }
        finally
        {
          connection.TryClose();
        }
      }
    }
 
    /// <summary>
    /// 生成插入数据的sql语句。
    /// </summary>
    /// <param name="database"></param>
    /// <param name="command"></param>
    /// <param name="table"></param>
    /// <returns></returns>
    private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)
    {
      var names = new StringBuilder();
      var values = new StringBuilder();
      //将一个DataTable的数据转换为数组的数组
      var data = table.ToArray();
 
      //设置ArrayBindCount属性
      command.GetType().GetProperty("ArrayBindCount").SetValue(command, table.Rows.Count, null);
 
      var syntax = database.Provider.GetService<ISyntaxProvider>();
      for (var i = 0; i < table.Columns.Count; i++)
      {
        var column = table.Columns[i];
 
        var parameter = database.Provider.DbProviderFactory.CreateParameter();
        if (parameter == null)
        {
          continue;
        }
        parameter.ParameterName = column.ColumnName;
        parameter.Direction = ParameterDirection.Input;
        parameter.DbType = column.DataType.GetDbType();
        parameter.Value = data[i];
 
        if (names.Length > 0)
        {
          names.Append(",");
          values.Append(",");
        }
        names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, column.ColumnName));
        values.AppendFormat("{0}{1}", syntax.ParameterPrefix, column.ColumnName);
 
        command.Parameters.Add(parameter);
      }
      return string.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values);
    }
  }

3. Stapeleinfügung von SQLite-Daten

Für die Stapeleinfügung von SQLite ist lediglich das Öffnen einer Transaktion erforderlich.

4. MySql-Daten-Batch-Einfügung

public sealed class SQLiteBatcher : IBatcherProvider
  {
    /// <summary>
    /// 获取或设置提供者服务的上下文。
    /// </summary>
    public ServiceContext ServiceContext { get; set; }
 
    /// <summary>
    /// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
    /// </summary>
    /// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
    /// <param name="batchSize">每批次写入的数据量。</param>
    public void Insert(DataTable dataTable, int batchSize = 10000)
    {
      Checker.ArgumentNull(dataTable, "dataTable");
      if (dataTable.Rows.Count == 0)
      {
        return;
      }
      using (var connection = ServiceContext.Database.CreateConnection())
      {
        DbTransaction transcation = null;
        try
        {
          connection.TryOpen();
          transcation = connection.BeginTransaction();
          using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())
          {
            if (command == null)
            {
              throw new BatcherException(new ArgumentException("command"));
            }
            command.Connection = connection;
 
            command.CommandText = GenerateInserSql(ServiceContext.Database, dataTable);
            if (command.CommandText == string.Empty)
            {
              return;
            }
 
            var flag = new AssertFlag();
            dataTable.EachRow(row =>
              {
                var first = flag.AssertTrue();
                ProcessCommandParameters(dataTable, command, row, first);
                command.ExecuteNonQuery();
              });
          }
          transcation.Commit();
        }
        catch (Exception exp)
        {
          if (transcation != null)
          {
            transcation.Rollback();
          }
          throw new BatcherException(exp);
        }
        finally
        {
          connection.TryClose();
        }
      }
    }
 
    private void ProcessCommandParameters(DataTable dataTable, DbCommand command, DataRow row, bool first)
    {
      for (var c = 0; c < dataTable.Columns.Count; c++)
      {
        DbParameter parameter;
        //首次创建参数，是为了使用缓存
        if (first)
        {
          parameter = ServiceContext.Database.Provider.DbProviderFactory.CreateParameter();
          parameter.ParameterName = dataTable.Columns[c].ColumnName;
          command.Parameters.Add(parameter);
        }
        else
        {
          parameter = command.Parameters[c];
        }
        parameter.Value = row[c];
      }
    }
 
    /// <summary>
    /// 生成插入数据的sql语句。
    /// </summary>
    /// <param name="database"></param>
    /// <param name="table"></param>
    /// <returns></returns>
    private string GenerateInserSql(IDatabase database, DataTable table)
    {
      var syntax = database.Provider.GetService<ISyntaxProvider>();
      var names = new StringBuilder();
      var values = new StringBuilder();
      var flag = new AssertFlag();
      table.EachColumn(column =>
        {
          if (!flag.AssertTrue())
          {
            names.Append(",");
            values.Append(",");
          }
          names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName));
          values.AppendFormat("{0}{1}", syntax.ParameterPrefix, column.ColumnName);
        });
      return string.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values);
    }
  }

MySql-Batch-Einfügung besteht darin, alle Werte in die Werte der Anweisung zu schreiben, zum Beispiel insertbatcher( id, name) Werte (1, '1', 2, '2', 3, '3', ......... 10, '10').

/// <summary>
  /// 为 MySql.Data 组件提供的用于批量操作的方法。
  /// </summary>
  public sealed class MySqlBatcher : IBatcherProvider
  {
    /// <summary>
    /// 获取或设置提供者服务的上下文。
    /// </summary>
    public ServiceContext ServiceContext { get; set; }
 
    /// <summary>
    /// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
    /// </summary>
    /// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
    /// <param name="batchSize">每批次写入的数据量。</param>
    public void Insert(DataTable dataTable, int batchSize = 10000)
    {
      Checker.ArgumentNull(dataTable, "dataTable");
      if (dataTable.Rows.Count == 0)
      {
        return;
      }
      using (var connection = ServiceContext.Database.CreateConnection())
      {
        try
        {
          connection.TryOpen();
          using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())
          {
            if (command == null)
            {
              throw new BatcherException(new ArgumentException("command"));
            }
            command.Connection = connection;
 
            command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);
            if (command.CommandText == string.Empty)
            {
              return;
            }
            command.ExecuteNonQuery();
          }
        }
        catch (Exception exp)
        {
          throw new BatcherException(exp);
        }
        finally
        {
          connection.TryClose();
        }
      }
    }
 
    /// <summary>
    /// 生成插入数据的sql语句。
    /// </summary>
    /// <param name="database"></param>
    /// <param name="command"></param>
    /// <param name="table"></param>
    /// <returns></returns>
    private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)
    {
      var names = new StringBuilder();
      var values = new StringBuilder();
      var types = new List<DbType>();
      var count = table.Columns.Count;
      var syntax = database.Provider.GetService<ISyntaxProvider>();
      table.EachColumn(c =>
        {
          if (names.Length > 0)
          {
            names.Append(",");
          }
          names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, c.ColumnName));
          types.Add(c.DataType.GetDbType());
        });
 
      var i = 0;
      foreach (DataRow row in table.Rows)
      {
        if (i > 0)
        {
          values.Append(",");
        }
        values.Append("(");
        for (var j = 0; j < count; j++)
        {
          if (j > 0)
          {
            values.Append(", ");
          }
          var isStrType = IsStringType(types[j]);
          var parameter = CreateParameter(database.Provider, isStrType, types[j], row[j], syntax.ParameterPrefix, i, j);
          if (parameter != null)
          {
            values.Append(parameter.ParameterName);
            command.Parameters.Add(parameter);
          }
          else if (isStrType)
          {
            values.AppendFormat("&#39;{0}&#39;", row[j]);
          }
          else
          {
            values.Append(row[j]);
          }
        }
        values.Append(")");
        i++;
      }
      return string.Format("INSERT INTO {0}({1}) VALUES {2}", DbUtility.FormatByQuote(syntax, table.TableName), names, values);
    }
 
    /// <summary>
    /// 判断是否为字符串类别。
    /// </summary>
    /// <param name="dbType"></param>
    /// <returns></returns>
    private bool IsStringType(DbType dbType)
    {
      return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength;
    }
 
    /// <summary>
    /// 创建参数。
    /// </summary>
    /// <param name="provider"></param>
    /// <param name="isStrType"></param>
    /// <param name="dbType"></param>
    /// <param name="value"></param>
    /// <param name="parPrefix"></param>
    /// <param name="row"></param>
    /// <param name="col"></param>
    /// <returns></returns>
    private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col)
    {
      //如果生成全部的参数，则速度会很慢，因此，只有数据类型为字符串(包含&#39;号)和日期型时才添加参数
      if ((isStrType && value.ToString().IndexOf(&#39;\&#39;&#39;) != -1) || dbType == DbType.DateTime)
      {
        var name = string.Format("{0}p_{1}_{2}", parPrefix, row, col);
        var parameter = provider.DbProviderFactory.CreateParameter();
        parameter.ParameterName = name;
        parameter.Direction = ParameterDirection.Input;
        parameter.DbType = dbType;
        parameter.Value = value;
        return parameter;
      }
      return null;
    }
  }

5. Test

Als nächstes schreiben Sie einen Testfall, um die Auswirkung der Stapeleinfügung zu sehen.

Die folgende Tabelle listet die Zeit auf, die jede der vier Datenbanken benötigt, um 100.000 Daten zu generieren.

public void TestBatchInsert()
   {
     Console.WriteLine(TimeWatcher.Watch(() =>
       InvokeTest(database =>
         {
           var table = new DataTable("Batcher");
           table.Columns.Add("Id", typeof(int));
           table.Columns.Add("Name1", typeof(string));
           table.Columns.Add("Name2", typeof(string));
           table.Columns.Add("Name3", typeof(string));
           table.Columns.Add("Name4", typeof(string));
 
           //构造100000条数据
           for (var i = 0; i < 100000; i++)
           {
             table.Rows.Add(i, i.ToString(), i.ToString(), i.ToString(), i.ToString());
           }
 
           //获取 IBatcherProvider
           var batcher = database.Provider.GetService<IBatcherProvider>();
           if (batcher == null)
           {
             Console.WriteLine("不支持批量插入。");
           }
           else
           {
             batcher.Insert(table);
           }
 
           //输出batcher表的数据量
           var sql = new SqlCommand("SELECT COUNT(1) FROM Batcher");
           Console.WriteLine("当前共有 {0} 条数据", database.ExecuteScalar(sql));
 
         })));
   }

Weitere C#-Datenbanken verschiedener Typen Für Artikel zum Thema Big-Data-Batch-Einfügung (SQLServer, Oracle, SQLite und MySql) beachten Sie bitte die chinesische PHP-Website!