使用MySqlBulkLoader批量插入数据

 
最近在项目中遇到插入数据瓶颈，几万、几十万、几百万的数据保存到MYSQL数据库，使用EF插入数据速度非常慢，数据量非常大时EF插入需要几十分钟，甚至几个小时，这样子的速度肯定不是我们所期望的。
后面经过了解与研究发现MySqlBulkLoader，可以批量将数据插入到数据库并且速度上面远远优于EF。
MySqlBulkLoader主要的实现方式：将需要插入的数据转成DataTable，DataTable转成一个CSV文件，将CSV文件使用批量导入的形式导入到数据库里面去。
 
注意：
1）.数据库连接地址需要添加配置AllowLoadLocalInfile=true,允许本地文件导入；
Data Source = 数据库地址; Port = 端口; Initial Catalog = 数据库名; User Id = 用户名; Password = 密码;AllowLoadLocalInfile=true;
2）.插入的时候会返回插入行数，但是检查所有的数据都正确，也没有报异常，却返回了插入数量为0，可以检查表是否有唯一索引，插入的数据是否违反了唯一索引。
（以下分块展示了代码，如果需要看完整的代码直接看 5.完整的代码） 
 
1.将List转化为DataTable 

1.    /// <summary>
2.         /// 将List转化为DataTable
3.         /// </summary>
4.         /// <returns></returns>
5.         public DataTable ListToDataTable<T>(List<T> data)
6.         {
7.             #region 创建一个DataTable，以实体名称作为DataTable名称
9.             var tableName = typeof(T).Name;
10.             tableName = tableName.ToSnakeCase(); /*实体名称与表名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/
11.             DataTable dt = new DataTable
12.             {
13.                 TableName = tableName
14.             };
16.             #endregion
18.             #region 拿取列名，以实体的属性名作为列名      
20.             var properties = typeof(T).GetProperties();
21.             foreach (var item in properties)
22.             {
23.                 var curFileName = item.Name;
24.                 curFileName = curFileName.ToSnakeCase();/*列名与字段名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/
25.                 dt.Columns.Add(curFileName);
26.             }
28.             #endregion
30.             #region 列赋值
31.             foreach (var item in data)
32.             {
33.                 DataRow dr = dt.NewRow();
34.                 var columns = dt.Columns;
36.                 var curPropertyList = item.GetType().GetProperties();
37.                 foreach (var p in curPropertyList)
38.                 {
39.                     var name = p.Name;
40.                     name = name.ToSnakeCase();/*列名与字段名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/
41.                     var curValue = p.GetValue(item);
43.                     int i = columns.IndexOf(name);
44.                     dr[i] = curValue;
45.                 }
47.                 dt.Rows.Add(dr);
48.             }
50.             #endregion  
52.             return dt;
53.         }

复制代码

  
2.将DataTable转换为标准的CSV文件 

1.   /// <summary>
2.     /// csv扩展
3.     /// </summary>
4.     public static class CSVEx
5.     {
6.         /// <summary>
7.         ///将DataTable转换为标准的CSV文件
8.         /// </summary>
9.         /// <param name="table">数据表</param>
10.         /// <param name="tmpPath">文件地址</param>
11.         /// <returns>返回标准的CSV</returns>
12.         public static void ToCsv(this DataTable table, string tmpPath)
13.         {
14.             //以半角逗号（即,）作分隔符，列为空也要表达其存在。
15.             //列内容如存在半角逗号（即,）则用半角引号（即""）将该字段值包含起来。
16.             //列内容如存在半角引号（即"）则应替换成半角双引号（""）转义，并用半角引号（即""）将该字段值包含起来。
17.             StringBuilder sb = new StringBuilder();
18.             DataColumn colum;
19.             foreach (DataRow row in table.Rows)
20.             {
21.                 for (int i = 0; i < table.Columns.Count; i++)
22.                 {
23.                     Type _datatype = typeof(DateTime);
24.                     colum = table.Columns[i];
25.                     if (i != 0) sb.Append("\t");
26.                     //if (colum.DataType == typeof(string) && row[colum].ToString().Contains(","))
27.                     //{
28.                     //    sb.Append(""" + row[colum].ToString().Replace(""", """") + """);
29.                     //}
30.                     if (colum.DataType == _datatype)
31.                     {
32.                         sb.Append(((DateTime)row[colum]).ToString("yyyy/MM/dd HH:mm:ss"));
33.                     }
34.                     else sb.Append(row[colum].ToString());
35.                 }
36.                 sb.Append("\r\n");
37.             }
38.             StreamWriter sw = new StreamWriter(tmpPath, false, UTF8Encoding.UTF8);
39.             sw.Write(sb.ToString());
40.             sw.Close();
41.         }
43.     }

复制代码

 
3.CSV文件导入数据到数据库

1.     /// <summary>
2.     /// 批量导入mysql帮助类
3.     /// </summary>
4.     public static class MySqlHelper
5.     {
6.         /// <summary>
7.         /// MySqlBulkLoader批量导入
8.         /// </summary>
9.         /// <param name="_mySqlConnection">数据库连接地址</param>
10.         /// <param name="table"></param>
11.         /// <param name="csvName"></param>
12.         /// <returns></returns>
13.         public static int BulkLoad(MySqlConnection _mySqlConnection, DataTable table, string csvName)
14.         {
15.             var columns = table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList();
16.             MySqlBulkLoader bulk = new MySqlBulkLoader(_mySqlConnection)
17.             {
18.                 FieldTerminator = "\t",
19.                 FieldQuotationCharacter = '"',
20.                 EscapeCharacter = '"',
21.                 LineTerminator = "\r\n",
22.                 FileName = csvName,
23.                 NumberOfLinesToSkip = 0,
24.                 TableName = table.TableName,
26.             };
28.             bulk.Columns.AddRange(columns);
29.             return bulk.Load();
30.         }
31.     }

复制代码

  
4.使用MySqlBulkLoader批量插入数据
[code]        ///         /// 使用MySqlBulkLoader批量插入数据        ///         ///         ///         ///         ///         public int BulkLoaderData(List data)        {            if (data.Count