ホームページ >バックエンド開発 >C#.Net チュートリアル >C# での複数のデータベース (SqlServer、Oracle、SQLite、MySql) のビッグ データ バッチ挿入

C# での複数のデータベース (SqlServer、Oracle、SQLite、MySql) のビッグ データ バッチ挿入

高洛峰
高洛峰オリジナル
2017-01-13 15:40:192679ブラウズ

以前は、SqlServer がデータのバッチ挿入をサポートしていることだけを知っていましたが、Oracle、SQLite、MySql もサポートしていることはほとんど知りませんでした。しかし、今日はいくつかのバッチ挿入ソリューションを投稿します。データベース用。

まず、IProvider にはバッチ挿入用のプラグイン サービス インターフェイス IBatcherProvider があります。このインターフェイスは前の記事で説明しました。

/// <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. SqlServer データのバッチ挿入

SqlBulkCopy を使用するだけで、このクラスの実装は非常に簡単です。

/// <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();
        }
      }
    }
  }

上記はトランザクションを使用しません。トランザクションを使用する場合は、SqlBulkCopyOptions.UseInternalTransaction を設定できます。

2. Oracle データのバッチ挿入

System.Data.OracleClient はバッチ挿入をサポートしていないため、プロバイダとして Oracle.DataAccess コンポーネントのみを使用できます。

/// <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);
    }
  }

上記の最も重要な手順は、DataTable を配列の配列表現、つまり object[][] に変換することです。前者の配列の上付き文字は列の数であり、後者の配列は列の数です。したがって、ループ Columns はパラメータの値として配列されます。つまり、パラメータの値は配列です。 insert ステートメントは、通常の insert ステートメントと何ら変わりません。

3. SQLite データのバッチ挿入

SQLite のバッチ挿入では、トランザクションを開くだけで済みます。その具体的な原理は不明です。

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);
    }
  }

4. MySqlデータの一括挿入

/// <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;
    }
  }

MySqlの一括挿入とは、ステートメントの値にすべての値を書き込むことです。たとえば、insert Batcher(id, name)の値です。 (1, '1', 2, '2' , 3, '3', ......... 10, '10')。

5. テスト

次に、バッチ挿入を使用した効果を確認するテスト ケースを作成します。

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));
 
         })));
   }

次の表は、4 つのデータベースのそれぞれが 100,000 個のデータを生成するのにかかる時間を示しています

C# での複数のデータベース (SqlServer、Oracle、SQLite、MySql) のビッグ データ バッチ挿入

複数のデータベース (SqlServer、Oracle、SQLite、MySql) の C# ビッグ データ バッチ挿入の詳細については、関連記事をご覧ください。 PHP中国語ウェブサイトをフォローしてください!

声明:
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。