在之前只知道SqlServer支持数据批量插入，殊不知道Oracle、SQLite和MySql也是支持的，不过Oracle需要使用Orace.DataAccess驱动，今天就贴出几种数据库的批量插入解决方法。
首先说一下，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 = );
}
一、SqlServer数据批量插入
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 = )
{
Checker.ArgumentNull(dataTable, "dataTable");
if (dataTable.Rows.Count == )
{
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();
}
}
}
} SqlBulkCopy的ColumnMappings中列的名称受大小写敏感限制，因此在构造DataTable的时候应请注意列名要与表一致。 以上没有使用事务，使用事务在性能上会有一定的影响，如果要使用事务，可以设置SqlBulkCopyOptions.UseInternalTransaction。
二、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 = )
{
Checker.ArgumentNull(dataTable, "dataTable");
if (dataTable.Rows.Count == )
{
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 = ; 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 > )
{
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将后数组作为Parameter的值，也就是说，参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。
三、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 = )
{
Checker.ArgumentNull(dataTable, "dataTable");
if (dataTable.Rows.Count == )
{
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 = ; 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数据批量插入
/// <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 = )
{
Checker.ArgumentNull(dataTable, "dataTable");
if (dataTable.Rows.Count == )
{
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 > )
{
names.Append(",");
}
names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, c.ColumnName));
types.Add(c.DataType.GetDbType());
}); var i = ;
foreach (DataRow row in table.Rows)
{
if (i > )
{
values.Append(",");
}
values.Append("(");
for (var j = ; j < count; j++)
{
if (j > )
{
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);
} /// <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)
{
//如果生成全部的参数，则速度会很慢，因此，只有数据类型为字符串(包含'号)和日期型时才添加参数
if ((isStrType && value.ToString().IndexOf('\'') != -) || 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的批量插入，是将值全部写在语句的values里，例如，insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。
五、测试
接下来写一个测试用例来看一下使用批量插入的效果。
[Test]
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 = ; i < ; 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)); })));
}
以下表中列出了四种数据库生成10万条数据各耗用的时间
|
数据库 |
耗用时间 |
| MsSql | 00:00:02.9376300 |
| Oracle | 00:00:01.5155959 |
| SQLite | 00:00:01.6275634 |
| MySql | 00:00:05.4166891 |
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中
/// <summary>
/// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中
/// </summary>
/// <param name="strTableName">数据库中对应的表名</param>
/// <param name="dtData">数据集</param>
public void SqlBulkCopyInsert(string strTableName, DataTable dtData)
{
string ConStr = connectionString;// 数据库连接字符串 try
{
using (SqlBulkCopy sqlRevdBulkCopy = new SqlBulkCopy(ConStr))//引用SqlBulkCopy
{
sqlRevdBulkCopy.DestinationTableName = strTableName;//数据库中对应的表名 sqlRevdBulkCopy.NotifyAfter = dtData.Rows.Count;//有几行数据 sqlRevdBulkCopy.WriteToServer(dtData);//数据导入数据库 sqlRevdBulkCopy.Close();//关闭连接
}
}
catch (Exception ex)
{
WriteErrorLog(ex.Message + "数据库处理出错654行。SqlBulkCopyInsert");
throw (ex);
}
}
#endregion
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
//BCP copy
SqlConnection conn = new SqlConnection();
conn.ConnectionString = "server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS";
conn.Open(); SqlTransaction sqlbulkTransaction = conn.BeginTransaction(); //请在插入数据的同时检查约束，如果发生错误调用sqlbulkTransaction事务
SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction); copy.DestinationTableName = "T_SMS_SendInfo";
foreach (DataColumn dc in dataTable.Columns)
{
copy.ColumnMappings.Add(dc.ColumnName, dc.ColumnName); }
try
{
copy.WriteToServer(dataTable);
sqlbulkTransaction.Commit();
}
catch (Exception ex)
{
sqlbulkTransaction.Rollback();
Console.WriteLine(ex.ToString());
}
finally
{
copy.Close();
conn.Close();
}
BCP copy
C#:几种数据库的大数据批量插入 - faib的更多相关文章
-
c#几种数据库的大数据批量插入（SqlServer、Oracle、SQLite和MySql）
这篇文章主要介绍了c#几种数据库的大数据批量插入(SqlServer.Oracle.SQLite和MySql),需要的朋友可以了解一下. 在之前只知道SqlServer支持数据批量插入,殊不知道Ora ...
-
C#中几种数据库的大数据批量插入
C#语言中对SqlServer.Oracle.SQLite和MySql中的数据批量插入是支持的,不过Oracle需要使用Orace.DataAccess驱动. IProvider里有一个用于实现批量插 ...
-
C#:几种数据库的大数据批量插入
在之前只知道SqlServer支持数据批量插入,殊不知道Oracle.SQLite和MySql也是支持的,不过Oracle需要使用Orace.DataAccess驱动,今天就贴出几种数据库的批量插入解 ...
-
C#:几种数据库的大数据批量插入(转)
在之前只知道SqlServer支持数据批量插入,殊不知道Oracle.SQLite和MySql也是支持的,不过Oracle需要使用Orace.DataAccess驱动,今天就贴出几种数据库的批量插入解 ...
-
【C#-批量插入数据到数据库】DataTable数据批量插入数据的库三种方法：SqlCommand.EcecuteNonQurery()，SqlDataAdapter.Update(DataTable) ，SqlBulkCopy.WriteToServer(Datatable)
第一种方法:使用SqlCommand.EcecuteNonQurery() 效率最慢 第二种方法:使用SqlDataAdapter.Update(DataTable) 效率次之 第三种方法:使用 ...
-
大数据批量插入数据库使用（SqlBulkCopy ）效率更高
SqlBulkCopy类是System.Data.SqlClient下的类,我们开发中不常用,甚至不知道有这么一个类的存在,但确实比sql插入,事务批量插入,sql批量拼接插入快很多,比调用存储过程插 ...
-
oracle基于3种方法的大数据量插入更新
过程插入更新的3种方法: a.逐条检查插入或更新,同时执行插入或更新 b.逐条merge into(逐条是为了记录过程日志与错误信息) c.基于关联数组的检查插入.更新,通过forall批量sql执行 ...
-
java，大数据批量插入、更新
public void exec(Connection conn){ try { conn.setAutoCommit(false); Long beginTime = System.currentT ...
-
EF大数据批量添加性能问题（续）
昨天在园子里发了一篇如题的文章EF大数据批量添加性能问题,就引来一大堆的吐槽,我认为知识就应该这样分享出来,不然总以为自己很了不起:再说说昨天那篇文章,很多自认为很牛逼的人都评论说把SaveChang ...
随机推荐
-
Torch7在Ubuntu下的安装与配置
Torch7的本系列教程的主要目的是介绍Torch的入门使用.今天首先分享一下Torch7的安装.(在Ubuntu14.04安装torch7) 为什么选择Torch Torch的目标是在建立科学算法的 ...
-
Atitit.js javascript的rpc框架选型
Atitit.js javascript的rpc框架选型 1. Dwr1 2. 使用AJAXRPC1 2.2. 数据类型映射表1 3. json-rpc轻量级远程调用协议介绍及使用2 3.1. 2.3 ...
-
100726A
迭代深搜,从最深的地方搜,然后一个数被搜过了,标记用过,以后不再访问 #include<iostream> #include<cstring> #include<map& ...
-
EF DbModelBuilder
protected override void OnModelCreating(DbModelBuilder modelBuilder) { var model = modelBuilder.Buil ...
-
JMeter工具的使用-ForEach
1,Add Thread group this detail information about this panel as below link http://jmeter.apache.org/u ...
-
Batch file Functions
Quote from: http://ss64.com/nt/syntax-functions.html Batch file Functions Packaging up code into a d ...
-
给Object扩展新方法
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8&quo ...
-
hdu多校第4场 B Harvest of Apples（莫队）
Problem B. Harvest of Apples Time Limit: / MS (Java/Others) Memory Limit: / K (Java/Others) Total Su ...
-
笨办法学Python - 习题11-12: Asking Questions &; Prompting People
目录 1.习题 11: 提问 2.习题 12: 提示别人 3.总结 1.习题 11: 提问 学习目标:了解人机交互场景,熟悉raw_input 的用法. 1.在 Python2.x 中 raw_inp ...
-
Java程序员职业规划
Java 程序员职业规划 无论你是学习了 Java 即将进入企业工作,还是已经踏入了工作岗位的程序员.但是面对层出不穷的新技术,激增的就业压力,不断分化的开发角色,再加上 IT 发展的不明确,做出职业 ...