C# 几种数据库的大数据批量插入

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: 转载:http://www.cnblogs.com/luluping/archive/2012/08/09/2629515.html 在之前只知道SqlServer支持数据批量插入,殊不知道Oracle、SQLite和MySql也是支持的,不过Oracle需要使用Orace.DataAccess驱动,今天就贴出几种数据库的批量插入解决方法。

转载:http://www.cnblogs.com/luluping/archive/2012/08/09/2629515.html

在之前只知道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 = 10000); }
复制代码
复制代码

一、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 = 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。

二、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将后数组作为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 = 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数据批量插入

复制代码
复制代码
 /// <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("'{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('\'') != -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的批量插入,是将值全部写在语句的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 = 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)); }))); }
复制代码
复制代码

以下表中列出了四种数据库生成10万条数据各耗用的时间

数据库

耗用时间

MsSql 00:00:02.9376300
Oracle 00:00:01.5155959
SQLite 00:00:01.6275634
MySql 00:00:05.4166891
相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
13天前
|
数据库连接 数据库 C#
Windows下C# 通过ADO.NET方式连接南大通用GBase 8s数据库(下)
本文接续前文,深入讲解了在Windows环境下使用C#和ADO.NET操作南大通用GBase 8s数据库的方法。通过Visual Studio 2022创建项目,添加GBase 8s的DLL引用,并提供了详细的C#代码示例,涵盖数据库连接、表的创建与修改、数据的增删查改等操作,旨在帮助开发者提高数据库管理效率。
|
2月前
|
算法 大数据 数据库
云计算与大数据平台的数据库迁移与同步
本文详细介绍了云计算与大数据平台的数据库迁移与同步的核心概念、算法原理、具体操作步骤、数学模型公式、代码实例及未来发展趋势与挑战。涵盖全量与增量迁移、一致性与异步复制等内容,旨在帮助读者全面了解并应对相关技术挑战。
44 3
|
2月前
|
SQL 缓存 大数据
C#高效处理大数据的批次处理,以及最好的数据库设计
C#高效处理大数据的批次处理,以及最好的数据库设计
70 0
|
2月前
|
大数据 关系型数据库 数据库
python 批量处理大数据写入数据库
python 批量处理大数据写入数据库
126 0
|
3月前
|
SQL 存储 关系型数据库
C#一分钟浅谈:使用 ADO.NET 进行数据库访问
【9月更文挑战第3天】在.NET开发中,与数据库交互至关重要。ADO.NET是Microsoft提供的用于访问关系型数据库的类库,包含连接数据库、执行SQL命令等功能。本文从基础入手,介绍如何使用ADO.NET进行数据库访问,并提供示例代码,同时讨论常见问题及其解决方案,如连接字符串错误、SQL注入风险和资源泄露等,帮助开发者更好地利用ADO.NET提升应用的安全性和稳定性。
321 6
|
4月前
|
关系型数据库 MySQL 大数据
教你使用Python玩转MySQL数据库,大数据导入不再是难题!
教你使用Python玩转MySQL数据库,大数据导入不再是难题!
|
27天前
|
SQL 关系型数据库 MySQL
12 PHP配置数据库MySQL
路老师分享了PHP操作MySQL数据库的方法,包括安装并连接MySQL服务器、选择数据库、执行SQL语句(如插入、更新、删除和查询),以及将结果集返回到数组。通过具体示例代码,详细介绍了每一步的操作流程,帮助读者快速入门PHP与MySQL的交互。
34 1
|
29天前
|
SQL 关系型数据库 MySQL
go语言数据库中mysql驱动安装
【11月更文挑战第2天】
39 4
|
1月前
|
监控 关系型数据库 MySQL
数据库优化:MySQL索引策略与查询性能调优实战
【10月更文挑战第27天】本文深入探讨了MySQL的索引策略和查询性能调优技巧。通过介绍B-Tree索引、哈希索引和全文索引等不同类型,以及如何创建和维护索引,结合实战案例分析查询执行计划,帮助读者掌握提升查询性能的方法。定期优化索引和调整查询语句是提高数据库性能的关键。
194 1