几种数据库的大数据批量插入【转】 C#:几种数据库的大数据批量插入
c#\u5b9e\u73b0\u51e0\u79cd\u6570\u636e\u5e93\u7684\u5927\u6570\u636e\u6279\u91cf\u63d2\u5165\u4e00\u3001SqlServer\u6570\u636e\u6279\u91cf\u63d2\u5165
SqlServer\u7684\u6279\u91cf\u63d2\u5165\u5f88\u7b80\u5355\uff0c\u4f7f\u7528SqlBulkCopy\u5c31\u53ef\u4ee5\uff0c\u4ee5\u4e0b\u662f\u8be5\u7c7b\u7684\u5b9e\u73b0\uff1a
\u590d\u5236\u4ee3\u7801
///
/// \u4e3a System.Data.SqlClient \u63d0\u4f9b\u7684\u7528\u4e8e\u6279\u91cf\u64cd\u4f5c\u7684\u65b9\u6cd5\u3002
///
public sealed class MsSqlBatcher : IBatcherProvider
{
///
/// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002
///
public ServiceContext ServiceContext { get; set; }
///
/// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002
///
/// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002
/// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002
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();
//\u7ed9\u8868\u540d\u52a0\u4e0a\u524d\u540e\u5bfc\u7b26
var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService(), dataTable.TableName);
using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)
{
DestinationTableName = tableName,
BatchSize = batchSize
})
{
//\u5faa\u73af\u6240\u6709\u5217\uff0c\u4e3abulk\u6dfb\u52a0\u6620\u5c04
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();
}
}
}
}
SqlBulkCopy\u7684ColumnMappings\u4e2d\u5217\u7684\u540d\u79f0\u53d7\u5927\u5c0f\u5199\u654f\u611f\u9650\u5236\uff0c\u56e0\u6b64\u5728\u6784\u9020DataTable\u7684\u65f6\u5019\u5e94\u8bf7\u6ce8\u610f\u5217\u540d\u8981\u4e0e\u8868\u4e00\u81f4\u3002
\u4ee5\u4e0a\u6ca1\u6709\u4f7f\u7528\u4e8b\u52a1\uff0c\u4f7f\u7528\u4e8b\u52a1\u5728\u6027\u80fd\u4e0a\u4f1a\u6709\u4e00\u5b9a\u7684\u5f71\u54cd\uff0c\u5982\u679c\u8981\u4f7f\u7528\u4e8b\u52a1\uff0c\u53ef\u4ee5\u8bbe\u7f6eSqlBulkCopyOptions.UseInternalTransaction\u3002
\u590d\u5236\u4ee3\u7801
\u4e8c\u3001Oracle\u6570\u636e\u6279\u91cf\u63d2\u5165
System.Data.OracleClient\u4e0d\u652f\u6301\u6279\u91cf\u63d2\u5165\uff0c\u56e0\u6b64\u53ea\u80fd\u4f7f\u7528Oracle.DataAccess\u7ec4\u4ef6\u6765\u4f5c\u4e3a\u63d0\u4f9b\u8005\u3002
View Code
\u4ee5\u4e0a\u6700\u91cd\u8981\u7684\u4e00\u6b65\uff0c\u5c31\u662f\u5c06DataTable\u8f6c\u4e3a\u6570\u7ec4\u7684\u6570\u7ec4\u8868\u793a\uff0c\u5373object[][]\uff0c\u524d\u6570\u7ec4\u7684\u4e0a\u6807\u662f\u5217\u7684\u4e2a\u6570\uff0c\u540e\u6570\u7ec4\u662f\u884c\u7684\u4e2a\u6570\uff0c\u56e0\u6b64\u5faa\u73afColumns\u5c06\u540e\u6570\u7ec4\u4f5c\u4e3aParameter\u7684\u503c\uff0c\u4e5f\u5c31\u662f\u8bf4\uff0c\u53c2\u6570\u7684\u503c\u662f\u4e00\u4e2a\u6570\u7ec4\u3002\u800cinsert\u8bed\u53e5\u4e0e\u4e00\u822c\u7684\u63d2\u5165\u8bed\u53e5\u6ca1\u6709\u4ec0\u4e48\u4e0d\u4e00\u6837\u3002
\u4e09\u3001SQLite\u6570\u636e\u6279\u91cf\u63d2\u5165
SQLite\u7684\u6279\u91cf\u63d2\u5165\u53ea\u9700\u5f00\u542f\u4e8b\u52a1\u5c31\u53ef\u4ee5\u4e86\uff0c\u8fd9\u4e2a\u5177\u4f53\u7684\u539f\u7406\u4e0d\u5f97\u800c\u77e5\u3002
View Code
\u56db\u3001MySql\u6570\u636e\u6279\u91cf\u63d2\u5165
\u590d\u5236\u4ee3\u7801
///
/// \u4e3a MySql.Data \u7ec4\u4ef6\u63d0\u4f9b\u7684\u7528\u4e8e\u6279\u91cf\u64cd\u4f5c\u7684\u65b9\u6cd5\u3002
///
public sealed class MySqlBatcher : IBatcherProvider
{
///
/// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002
///
public ServiceContext ServiceContext { get; set; }
///
/// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002
///
/// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002
/// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002
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();
}
}
}
///
/// \u751f\u6210\u63d2\u5165\u6570\u636e\u7684sql\u8bed\u53e5\u3002
///
///
///
///
///
private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)
{
var names = new StringBuilder();
var values = new StringBuilder();
var types = new List();
var count = table.Columns.Count;
var syntax = database.Provider.GetService();
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("'{0}'", 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);
}
///
/// \u5224\u65ad\u662f\u5426\u4e3a\u5b57\u7b26\u4e32\u7c7b\u522b\u3002
///
///
///
private bool IsStringType(DbType dbType)
{
return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength;
}
///
/// \u521b\u5efa\u53c2\u6570\u3002
///
///
///
///
///
///
///
///
///
private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col)
{
//\u5982\u679c\u751f\u6210\u5168\u90e8\u7684\u53c2\u6570\uff0c\u5219\u901f\u5ea6\u4f1a\u5f88\u6162\uff0c\u56e0\u6b64\uff0c\u53ea\u6709\u6570\u636e\u7c7b\u578b\u4e3a\u5b57\u7b26\u4e32(\u5305\u542b'\u53f7)\u548c\u65e5\u671f\u578b\u65f6\u624d\u6dfb\u52a0\u53c2\u6570
if ((isStrType && value.ToString().IndexOf('\'') != -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;
}
}
\u9996\u5148\u8bf4\u4e00\u4e0b\uff0cIProvider\u91cc\u6709\u4e00\u4e2a\u7528\u4e8e\u5b9e\u73b0\u6279\u91cf\u63d2\u5165\u7684\u63d2\u4ef6\u670d\u52a1\u63a5\u53e3IBatcherProvider\uff0c\u6b64\u63a5\u53e3\u5728\u524d\u4e00\u7bc7\u6587\u7ae0\u4e2d\u5df2\u7ecf\u63d0\u5230\u8fc7\u4e86\u3002 /// /// \u63d0\u4f9b\u6570\u636e\u6279\u91cf\u5904\u7406\u7684\u65b9\u6cd5\u3002 /// public interface IBatcherProvider : IProviderService { /// /// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002 /// /// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002 /// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002 void Insert(DataTable dataTable, int batchSize = 10000); } \u4e00\u3001SqlServer\u6570\u636e\u6279\u91cf\u63d2\u5165 SqlServer\u7684\u6279\u91cf\u63d2\u5165\u5f88\u7b80\u5355\uff0c\u4f7f\u7528SqlBulkCopy\u5c31\u53ef\u4ee5\uff0c\u4ee5\u4e0b\u662f\u8be5\u7c7b\u7684\u5b9e\u73b0\uff1a /// /// \u4e3a System.Data.SqlClient \u63d0\u4f9b\u7684\u7528\u4e8e\u6279\u91cf\u64cd\u4f5c\u7684\u65b9\u6cd5\u3002 /// public sealed class MsSqlBatcher : IBatcherProvider { /// /// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002 /// public ServiceContext ServiceContext { get; set; } /// /// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002 /// /// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002 /// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002 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(); //\u7ed9\u8868\u540d\u52a0\u4e0a\u524d\u540e\u5bfc\u7b26 var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService(), dataTable.TableName); using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null) { DestinationTableName = tableName, BatchSize = batchSize }) { //\u5faa\u73af\u6240\u6709\u5217\uff0c\u4e3abulk\u6dfb\u52a0\u6620\u5c04 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(); } } } } SqlBulkCopy\u7684ColumnMappings\u4e2d\u5217\u7684\u540d\u79f0\u53d7\u5927\u5c0f\u5199\u654f\u611f\u9650\u5236\uff0c\u56e0\u6b64\u5728\u6784\u9020DataTable\u7684\u65f6\u5019\u5e94\u8bf7\u6ce8\u610f\u5217\u540d\u8981\u4e0e\u8868\u4e00\u81f4\u3002 \u4ee5\u4e0a\u6ca1\u6709\u4f7f\u7528\u4e8b\u52a1\uff0c\u4f7f\u7528\u4e8b\u52a1\u5728\u6027\u80fd\u4e0a\u4f1a\u6709\u4e00\u5b9a\u7684\u5f71\u54cd\uff0c\u5982\u679c\u8981\u4f7f\u7528\u4e8b\u52a1\uff0c\u53ef\u4ee5\u8bbe\u7f6eSqlBulkCopyOptions.UseInternalTransaction\u3002 \u4e8c\u3001Oracle\u6570\u636e\u6279\u91cf\u63d2\u5165 System.Data.OracleClient\u4e0d\u652f\u6301\u6279\u91cf\u63d2\u5165\uff0c\u56e0\u6b64\u53ea\u80fd\u4f7f\u7528Oracle.DataAccess\u7ec4\u4ef6\u6765\u4f5c\u4e3a\u63d0\u4f9b\u8005\u3002 /// /// Oracle.Data.Access \u7ec4\u4ef6\u63d0\u4f9b\u7684\u7528\u4e8e\u6279\u91cf\u64cd\u4f5c\u7684\u65b9\u6cd5\u3002 /// public sealed class OracleAccessBatcher : IBatcherProvider { /// /// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002 /// public ServiceContext ServiceContext { get; set; } /// /// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002 /// /// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002 /// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002 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(); } } } /// /// \u751f\u6210\u63d2\u5165\u6570\u636e\u7684sql\u8bed\u53e5\u3002 /// /// /// /// /// private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); //\u5c06\u4e00\u4e2aDataTable\u7684\u6570\u636e\u8f6c\u6362\u4e3a\u6570\u7ec4\u7684\u6570\u7ec4 var data = table.ToArray(); //\u8bbe\u7f6eArrayBindCount\u5c5e\u6027 command.GetType().GetProperty("ArrayBindCount").SetValue(command, table.Rows.Count, null); var syntax = database.Provider.GetService(); for (var i = 0; i 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); } } \u4ee5\u4e0a\u6700\u91cd\u8981\u7684\u4e00\u6b65\uff0c\u5c31\u662f\u5c06DataTable\u8f6c\u4e3a\u6570\u7ec4\u7684\u6570\u7ec4\u8868\u793a\uff0c\u5373object[][]\uff0c\u524d\u6570\u7ec4\u7684\u4e0a\u6807\u662f\u5217\u7684\u4e2a\u6570\uff0c\u540e\u6570\u7ec4\u662f\u884c\u7684\u4e2a\u6570\uff0c\u56e0\u6b64\u5faa\u73afColumns\u5c06\u540e\u6570\u7ec4\u4f5c\u4e3aParameter\u7684\u503c\uff0c\u4e5f\u5c31\u662f\u8bf4\uff0c\u53c2\u6570\u7684\u503c\u662f\u4e00\u4e2a\u6570\u7ec4\u3002\u800cinsert\u8bed\u53e5\u4e0e\u4e00\u822c\u7684\u63d2\u5165\u8bed\u53e5\u6ca1\u6709\u4ec0\u4e48\u4e0d\u4e00\u6837\u3002 \u4e09\u3001SQLite\u6570\u636e\u6279\u91cf\u63d2\u5165 SQLite\u7684\u6279\u91cf\u63d2\u5165\u53ea\u9700\u5f00\u542f\u4e8b\u52a1\u5c31\u53ef\u4ee5\u4e86\uff0c\u8fd9\u4e2a\u5177\u4f53\u7684\u539f\u7406\u4e0d\u5f97\u800c\u77e5\u3002 public sealed class SQLiteBatcher : IBatcherProvider { /// /// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002 /// public ServiceContext ServiceContext { get; set; } /// /// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002 /// /// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002 /// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002 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 { 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); } } \u56db\u3001MySql\u6570\u636e\u6279\u91cf\u63d2\u5165 /// /// \u4e3a MySql.Data \u7ec4\u4ef6\u63d0\u4f9b\u7684\u7528\u4e8e\u6279\u91cf\u64cd\u4f5c\u7684\u65b9\u6cd5\u3002 /// public sealed class MySqlBatcher : IBatcherProvider { /// /// \u83b7\u53d6\u6216\u8bbe\u7f6e\u63d0\u4f9b\u8005\u670d\u52a1\u7684\u4e0a\u4e0b\u6587\u3002 /// public ServiceContext ServiceContext { get; set; } /// /// \u5c06 \u7684\u6570\u636e\u6279\u91cf\u63d2\u5165\u5230\u6570\u636e\u5e93\u4e2d\u3002 /// /// \u8981\u6279\u91cf\u63d2\u5165\u7684 \u3002 /// \u6bcf\u6279\u6b21\u5199\u5165\u7684\u6570\u636e\u91cf\u3002 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(); } } } /// /// \u751f\u6210\u63d2\u5165\u6570\u636e\u7684sql\u8bed\u53e5\u3002 /// /// /// /// /// private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); var types = new List(); var count = table.Columns.Count; var syntax = database.Provider.GetService(); 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 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("'{0}'", 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); } /// /// \u5224\u65ad\u662f\u5426\u4e3a\u5b57\u7b26\u4e32\u7c7b\u522b\u3002 /// /// /// private bool IsStringType(DbType dbType) { return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength; } /// /// \u521b\u5efa\u53c2\u6570\u3002 /// /// /// /// /// /// /// /// /// private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col) { //\u5982\u679c\u751f\u6210\u5168\u90e8\u7684\u53c2\u6570\uff0c\u5219\u901f\u5ea6\u4f1a\u5f88\u6162\uff0c\u56e0\u6b64\uff0c\u53ea\u6709\u6570\u636e\u7c7b\u578b\u4e3a\u5b57\u7b26\u4e32(\u5305\u542b'\u53f7)\u548c\u65e5\u671f\u578b\u65f6\u624d\u6dfb\u52a0\u53c2\u6570 if ((isStrType && value.ToString().IndexOf('\'') != -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\u7684\u6279\u91cf\u63d2\u5165\uff0c\u662f\u5c06\u503c\u5168\u90e8\u5199\u5728\u8bed\u53e5\u7684values\u91cc\uff0c\u4f8b\u5982\uff0cinsert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')\u3002 \u4e94\u3001\u6d4b\u8bd5 \u63a5\u4e0b\u6765\u5199\u4e00\u4e2a\u6d4b\u8bd5\u7528\u4f8b\u6765\u770b\u4e00\u4e0b\u4f7f\u7528\u6279\u91cf\u63d2\u5165\u7684\u6548\u679c\u3002
首先说一下，IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider，此接口在前一篇文章中已经提到过了。///<summary>/// 提供数据批量处理的方法。 ///</summary>publicinterface 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); }一、SqlServer数据批量插入SqlServer的批量插入很简单，使用SqlBulkCopy就可以，以下是该类的实现：///<summary>/// 为System.Data.SqlClient 提供的用于批量操作的方法。 ///</summary>publicsealedclass 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>publicvoid 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) { thrownew BatcherException(exp); } finally { connection.TryClose(); } } } }以上没有使用事务，使用事务在性能上会有一定的影响，如果要使用事务，可以设置SqlBulkCopyOptions.UseInternalTransaction。二、Oracle数据批量插入System.Data.OracleClient不支持批量插入，因此只能使用Oracle.DataAccess组件来作为提供者。///<summary>/// Oracle.Data.Access 组件提供的用于批量操作的方法。 ///</summary>publicsealedclass 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>publicvoid 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) { thrownew BatcherException(new ArgumentException("command")); } command.Connection = connection; command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable); command.ExecuteNonQuery(); } } catch (Exception exp) { thrownew BatcherException(exp); } finally { connection.TryClose(); } } } ///<summary>/// 生成插入数据的sql语句。 ///</summary>///<param name="database"></param>///<param name="command"></param>///<param name="table"></param>///<returns></returns>privatestring 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); } returnstring.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }以上最重要的一步，就是将DataTable转为数组的数组表示，即object[][]，前数组的上标是列的个数，后数组是行的个数，因此循环Columns将后数组作为Parameter的值，也就是说，参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。三、SQLite数据批量插入SQLite的批量插入只需开启事务就可以了，这个具体的原理不得而知。publicsealedclass 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>publicvoid 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) { thrownew 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(); } thrownew BatcherException(exp); } finally { connection.TryClose(); } } } privatevoid 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>privatestring 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); }); returnstring.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } } 四、MySql数据批量插入///<summary>/// 为MySql.Data 组件提供的用于批量操作的方法。 ///</summary>publicsealedclass 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>publicvoid 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) { thrownew 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) { thrownew BatcherException(exp); } finally { connection.TryClose(); } } } ///<summary>/// 生成插入数据的sql语句。 ///</summary>///<param name="database"></param>///<param name="command"></param>///<param name="table"></param>///<returns></returns>privatestring 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); } elseif (isStrType) { values.AppendFormat("'{0}'", row[j]); } else { values.Append(row[j]); } } values.Append(")"); i++; } returnstring.Format("INSERT INTO {0}({1}) VALUES {2}", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } ///<summary>/// 判断是否为字符串类别。 ///</summary>///<param name="dbType"></param>///<returns></returns>privatebool 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) { //如果生成全部的参数，则速度会很慢，因此，只有数据类型为字符串(包含'号)和日期型时才添加参数if ((isStrType && value.ToString().IndexOf('\'') != -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; } returnnull; } }MySql的批量插入，是将值全部写在语句的values里，例如，insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。五、测试接下来写一个测试用例来看一下使用批量插入的效果。绛旓細鍏崇郴鏁版嵁搴銆侀潪鍏崇郴鍨嬫暟鎹簱銆1銆佸叧绯绘暟鎹簱 鐗圭偣锛氭暟鎹泦涓帶鍒讹紱鍑忓皯鏁版嵁鍐椾綑绛夈傞傜敤鑼冨洿锛氬浜庣粨鏋勫寲鏁版嵁鐨勫鐞嗘洿鍚堥傦紝濡傚鐢熸垚缁┿佸湴鍧绛夛紝杩欐牱鐨勬暟鎹竴鑸儏鍐典笅闇瑕佷娇鐢ㄧ粨鏋勫寲鐨勬煡璇2銆侀潪鍏崇郴鏁版嵁搴 鐗圭偣锛氭槗鎵╁睍锛澶ф暟鎹閲忥紝楂樻ц兘锛涚伒娲荤殑鏁版嵁妯″瀷绛夈備娇鐢ㄨ寖鍥达細鎹ā鍨嬫瘮杈冪畝鍗曪紱闇瑕佺伒娲绘ф洿寮...
绛旓細鏁版嵁搴鏈変袱绉嶇被鍨嬶紝鍒嗗埆鏄叧绯诲瀷鏁版嵁搴撲笌闈炲叧绯诲瀷鏁版嵁搴撱1銆佸叧绯绘暟鎹簱 鍖呮嫭锛歁ySQL銆丮ariaDB锛圡ySQL鐨勪唬鏇垮搧锛岃嫳鏂囩淮鍩虹櫨绉戜粠MySQL杞悜MariaDB锛夈丳ercona Server锛圡ySQL鐨勪唬鏇垮搧锛夈丳ostgreSQL銆丮icrosoft Access銆丮icrosoft SQL Server銆丟oogle Fusion Tables銆侳ileMaker銆丱racle鏁版嵁搴撱丼ybase銆乨BASE銆丆lipper...
绛旓細JobRepository鏉ュ瓨鍌↗ob鎵ц鏈熺殑鍏冩暟鎹(杩欓噷鐨勫厓鏁版嵁鏄寚JobInstance銆丣obExecution銆丣obParameters銆丼tepExecution銆丒xecutionContext绛夋暟鎹),骞舵彁渚涗袱绉嶉粯璁ゅ疄鐜般備竴绉嶆槸瀛樻斁鍦ㄥ唴瀛樹腑;鍙︿竴绉嶅皢鍏冩暟鎹瓨鏀惧湪鏁版嵁搴涓傞氳繃灏嗗厓鏁版嵁瀛樻斁鍦ㄦ暟鎹簱涓,鍙互闅忔椂鐩戞帶鎵瑰鐞咼ob鐨勬墽琛岀姸鎬併侸ob鎵ц缁撴灉鏄垚鍔熻繕鏄け璐,骞朵笖浣垮緱鍦↗ob...
绛旓細ORACLE銆丏B2銆丼QL SERVER閮藉彲浠ワ紝鍏抽敭涓嶆槸閫変粈涔鏁版嵁搴锛岃屾槸鏁版嵁搴撳浣曚紭鍖栵紒 闇瑕佺湅浣犳棩甯稿浣曟搷浣滐紝浠ユ煡璇负涓绘垨鏄互瀛樺偍涓轰富鎴2鑰咃紝杩樿鐪嬩綘鐨勬暟鎹粨鏋勶紝閮借鍥犲湴鍒跺疁鐨勫幓浼樺寲锛佹墍浠ヤ笉鏄竴鍙ヨ瘽璇寸殑娓呯殑锛
绛旓細-access Access鏄竴绉嶆闈鏁版嵁搴锛屽彧閫傚悎鏁版嵁閲忓皯鐨勫簲鐢紝鍦ㄥ鐞嗗皯閲忔暟鎹拰鍗曟満璁块棶鐨勬暟鎹簱鏃舵槸寰堝ソ鐨勶紝鏁堢巼涔熷緢楂樸 浣嗘槸瀹冪殑鍚屾椂璁块棶瀹㈡埛绔笉鑳藉浜4涓俛ccess鏁版嵁搴撴湁涓瀹氱殑鏋侀檺锛屽鏋滄暟鎹揪鍒100M宸﹀彸锛屽緢瀹规槗閫犳垚鏈嶅姟鍣╥is鍋囨锛屾垨鑰呮秷鑰楁帀鏈嶅姟鍣ㄧ殑鍐呭瓨瀵艰嚧鏈嶅姟鍣ㄥ穿婧冦 - ...
绛旓細澶ф暟鎹甯哥敤鐨勬暟鎹鐞嗘柟寮忎富瑕佹湁浠ヤ笅鍑犵锛1. 鎵归噺澶勭悊锛圔ulk Processing锛: 鎵归噺澶勭悊鏄竴绉嶅湪澶ч噺鏁版嵁涓婃墽琛屾煇椤圭壒瀹氫换鍔＄殑鏂规硶銆傝繖绉嶆柟娉曢氬父鐢ㄤ簬鍒嗘瀽宸茬粡瀛樺偍鍦鏁版嵁搴涓殑鍘嗗彶鏁版嵁銆傛壒閲忓鐞嗙殑涓昏浼樼偣鏄晥鐜囬珮锛屽彲浠ュ湪澶ч噺鏁版嵁涓婁竴娆℃ф墽琛屼换鍔★紝浠庤岃妭鐪佹椂闂村拰璁＄畻璧勬簮銆2. 娴佸鐞嗭紙Streaming Processing锛...
绛旓細鍚屾椂锛屽畠鏈夊姪浜庨檷浣庡瓨鍌ㄦ垚鏈紝閫氳繃杩佺Щ鑷充綆鎴愭湰璁惧鎴栦簯绔紝鐏垫椿璋冩暣璧勬簮銆傛洿閲嶈鐨勬槸锛屾暟鎹縼绉诲己鍖栦簡瀹夊叏鎬э紝閫氳繃鍔犲瘑鍜屽畨鍏ㄥ瓨鍌ㄧ幆澧冿紝淇濇姢浼佷笟鏁忔劅淇℃伅鍏嶅彈濞佽儊銆傛澶栵紝瀹冭繕鑳芥敮鎸佹暟鎹垎鏋愶紝閫氳繃鏁版嵁娓呮礂鍜屾牸寮忚浆鎹紝涓烘繁鍏ユ寲鎺樻彁渚涘彲鑳姐傚父瑙佺殑鏁版嵁杩佺Щ鏂规硶 浼佷笟甯哥敤鐨勫洓绉澶ф暟鎹杩佺Щ绛栫暐鍖呮嫭锛鏁版嵁搴杩佺Щ锛...
绛旓細闈炲叧绯诲瀷鏁版嵁搴 闈炲叧绯诲瀷鏁版嵁搴撴槸鐩墠姣旇緝鏂扮殑涓绉嶆暟鎹簱,鐗圭偣灏辨槸鏁版嵁鍏ㄩ儴鐢遍敭鍊煎(key/value)缁勬垚.鑾峰彇鏁版嵁涓鑸彧閫氳繃閿(key)鏉ヨ幏鍙.渚嬪:ID Value 1 aaa.avi 2 bbb.MP4 杩欑鏁版嵁搴撲紭鐐规槸,閫熷害蹇,闇瑕佹槑纭殑鐩爣key鏉ュ揩閫熸寚瀹氬拰鑾峰彇鐩爣.涓鑸洰鍓嶅湪澶ф暟鎹瀛樺偍涓婁綋鐜扮潃浼樺娍.渚嬪澶у瀷瑙嗛...
绛旓細閽堝澶ц妯℃暟鎹鐨勬壒閲澶勭悊閲囩敤Sqoop娴佽绠澶ф暟鎹璁＄畻妯″紡銆係qoop锛氭槸涓娆惧紑婧愮殑宸ュ叿锛屼富瑕佺敤浜庡湪Hadoop锛圚ive锛変笌浼犵粺鐨鏁版嵁搴锛圡ySQL銆乸ost-gresql绛夛級闂磋繘琛屾暟鎹殑浼犻掞紝鍙互灏嗕竴涓叧绯诲瀷鏁版嵁搴撲腑鐨勬暟鎹鍏adoop鐨凥DFS涓紝涔熷彲浠ュ皢HDFS鐨勬暟鎹鍏ュ叧绯诲瀷鏁版嵁搴撲腑銆
绛旓細鍘熷洜閫氬父鏄細1銆佹暟鎹噺姣旇緝澶ф椂鍒嗘壒鎻掑叆锛屾湁鏃剁敋鑷宠繕浼氬垎鎴愬嚑涓〃鏉鎻掑叆鏁版嵁銆2銆澶ф暟鎹鍒嗘壒鎻掑叆鐨勫師鍥犳槸锛屼负浜嗚鍙栭熷害鐨勯棶棰樹篃浼氳繖鏍风殑锛岃繖鏍峰氨鍙互鎻愰珮杞欢鐨勬ц兘銆3銆佸垎鎵规彃鍏ヨ繕鏈夋椂涓轰簡鍑忓皯琛ㄧ殑鍐椾綑搴︺
