利用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名称 
8.  
9.             var tableName = typeof(T).Name; 
10.             tableName = tableName.ToSnakeCase(); /*实体名称与表名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/ 
11.             DataTable dt = new DataTable 
12.             { 
13.                 TableName = tableName 
14.             }; 
15.  
16.             #endregion 
17.  
18.             #region 拿取列名，以实体的属性名作为列名        
19.  
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.             } 
27.  
28.             #endregion 
29.  
30.             #region 列赋值 
31.             foreach (var item in data) 
32.             { 
33.                 DataRow dr = dt.NewRow(); 
34.                 var columns = dt.Columns; 
35.  
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); 
42.  
43.                     int i = columns.IndexOf(name); 
44.                     dr[i] = curValue; 
45.                 } 
46.  
47.                 dt.Rows.Add(dr); 
48.             } 
49.  
50.             #endregion   
51.  
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. } 
42.  
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, 
25.  
26. }; 
27.  
28. bulk.Columns.AddRange(columns); 
29. return bulk.Load(); 
30. } 
31. } 

4.使用MySqlBulkLoader批量插入数据

1. /// <summary> 
2. /// 使用MySqlBulkLoader批量插入数据 
3. /// </summary> 
4. /// <typeparam name="T"></typeparam> 
5. /// <param name="data"></param> 
6. /// <returns></returns> 
7. /// <exception cref="Exception"></exception> 
8. public int BulkLoaderData<T>(List<T> data) 
9. { 
10. if (data.Count <= 0) return 0; 
11.  
12. var connectString = "数据库连接地址"; 
13. using (MySqlConnection connection = new MySqlConnection(connectString)) 
14. { 
15. MySqlTransaction sqlTransaction = null; 
16. try 
17. { 
18. if (connection.State == ConnectionState.Closed) 
19. { 
20. connection.Open(); 
21. } 
22. sqlTransaction = connection.BeginTransaction(); 
23.  
24.  
25. var dt = ListToDataTable<T>(data); //将List转成dataTable 
26. string tmpPath = Path.GetTempFileName(); 
27. dt.ToCsv(tmpPath); //将DataTable转成CSV文件 
28. var insertCount = MySqlHelper.BulkLoad(connection, dt, tmpPath); //使用MySqlBulkLoader插入数据 
29. sqlTransaction.Commit(); 
30.  
31. try 
32. { 
33. if (File.Exists(tmpPath)) File.Delete(tmpPath); 
34. } 
35. catch (Exception) 
36. { 
37. //删除文件失败 
38.  
39. } 
40. return insertCount; //返回执行成功的条数 
41. } 
42. catch (Exception e) 
43. { 
44. if (sqlTransaction != null) 
45. { 
46. sqlTransaction.Rollback(); 
47. } 
48. //执行异常  
49. throw e; 
50. } 
51. } 
52.  
53. } 

5.完整的代码

1. namespace WebApplication1.BrantchInsert 
2. { 
3.  
4.     /// <summary> 
5.     /// 批量插入 
6.     /// </summary> 
7.     public class BulkLoader 
8.     { 
9.  
10.  
11.         /// <summary> 
12.         /// 测试批量插入入口 
13.         /// </summary> 
14.         /// <returns></returns> 
15.         public int BrantchDataTest() 
16.         { 
17.  
18.             #region 模拟数据 
19.             var data = new List<CrmCouponTestDto>() { 
20.                  new CrmCouponTestDto { 
21.                      Id=1, 
22.                      CouponCode="test001", 
23.                      CouponId = 1, 
24.                      MemberId=100, 
25.                      IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), 
26.                      UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), 
27.                      UsageShopId=0, 
28.                      UsageBillNo="", 
29.                      EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), 
30.                      EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), 
31.                      Status=0 
32.                  }, 
33.                  new CrmCouponTestDto { 
34.                      Id=2, 
35.                      CouponCode="test002", 
36.                      CouponId = 1, 
37.                        MemberId=101, 
38.                      IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), 
39.                      UsageTime=Convert.ToDateTime("2022-06-27 14:30:00"), 
40.                      UsageShopId=2, 
41.                      UsageBillNo="CS202206271430001", 
42.                      EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), 
43.                      EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), 
44.                      Status=1 
45.                  }, 
46.                   new CrmCouponTestDto { 
47.                      Id=3, 
48.                      CouponCode="test003", 
49.                      CouponId = 1, 
50.                      MemberId=102, 
51.                      IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), 
52.                      UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), 
53.                      UsageShopId=0, 
54.                      UsageBillNo="", 
55.                      EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), 
56.                      EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), 
57.                      Status=0 
58.                  }, 
59.                     new CrmCouponTestDto { 
60.                      Id=4, 
61.                      CouponCode="test004", 
62.                      CouponId = 1, 
63.                      MemberId=103, 
64.                      IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), 
65.                      UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), 
66.                      UsageShopId=0, 
67.                      UsageBillNo="", 
68.                      EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), 
69.                      EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), 
70.                      Status=0 
71.                  } 
72.              }; 
73.             #endregion 
74.             var result = BulkLoaderData<CrmCouponTestDto>(data); 
75.             return result; 
76.  
77.         } 
78.  
79.  
80.         /// <summary> 
81.         /// 使用MySqlBulkLoader批量插入数据 
82.         /// </summary> 
83.         /// <typeparam name="T"></typeparam> 
84.         /// <param name="data"></param> 
85.         /// <returns></returns> 
86.         /// <exception cref="Exception"></exception> 
87.         public int BulkLoaderData<T>(List<T> data) 
88.         { 
89.             if (data.Count <= 0) return 0; 
90.  
91.             var connectString = "数据库连接地址"; 
92.             using (MySqlConnection connection = new MySqlConnection(connectString)) 
93.             { 
94.                 MySqlTransaction sqlTransaction = null; 
95.                 try 
96.                 { 
97.                     if (connection.State == ConnectionState.Closed) 
98.                     { 
99.                         connection.Open(); 
100.                     } 
101.                     sqlTransaction = connection.BeginTransaction(); 
102.  
103.  
104.                     var dt = ListToDataTable<T>(data); //将List转成dataTable 
105.                     string tmpPath = Path.GetTempFileName(); 
106.                     dt.ToCsv(tmpPath); //将DataTable转成CSV文件 
107.                     var insertCount = MySqlHelper.BulkLoad(connection, dt, tmpPath); //使用MySqlBulkLoader插入数据 
108.                     sqlTransaction.Commit(); 
109.  
110.                     try 
111.                     { 
112.                         if (File.Exists(tmpPath)) File.Delete(tmpPath); 
113.                     } 
114.                     catch (Exception) 
115.                     { 
116.                         //删除文件失败 
117.  
118.                     } 
119.                     return insertCount; //返回执行成功的条数 
120.                 } 
121.                 catch (Exception e) 
122.                 { 
123.                     if (sqlTransaction != null) 
124.                     { 
125.                         sqlTransaction.Rollback(); 
126.                     } 
127.                     //执行异常  
128.                     throw e; 
129.                 } 
130.             } 
131.  
132.         } 
133.  
134.  
135.         /// <summary> 
136.         /// 将List转化为DataTable核心方法 
137.         /// </summary> 
138.         /// <returns></returns> 
139.         public DataTable ListToDataTable<T>(List<T> data) 
140.         { 
141.             #region 创建一个DataTable，以实体名称作为DataTable名称 
142.  
143.             var tableName = typeof(T).Name; 
144.             tableName = tableName.ToSnakeCase(); /*实体名称与表名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/ 
145.             DataTable dt = new DataTable 
146.             { 
147.                 TableName = tableName 
148.             }; 
149.  
150.             #endregion 
151.  
152.             #region 拿取列名，以实体的属性名作为列名        
153.  
154.             var properties = typeof(T).GetProperties(); 
155.             foreach (var item in properties) 
156.             { 
157.                 var curFileName = item.Name; 
158.                 curFileName = curFileName.ToSnakeCase();/*列名与字段名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/ 
159.                 dt.Columns.Add(curFileName); 
160.             } 
161.  
162.             #endregion 
163.  
164.             #region 列赋值 
165.             foreach (var item in data) 
166.             { 
167.                 DataRow dr = dt.NewRow(); 
168.                 var columns = dt.Columns; 
169.  
170.                 var curPropertyList = item.GetType().GetProperties(); 
171.                 foreach (var p in curPropertyList) 
172.                 { 
173.                     var name = p.Name; 
174.                     name = name.ToSnakeCase();/*列名与字段名进行转化，主要根据各项目的规定进行转化,不一定就是我这些写的这种转换方式*/ 
175.                     var curValue = p.GetValue(item); 
176.  
177.                     int i = columns.IndexOf(name); 
178.                     dr[i] = curValue; 
179.                 } 
180.  
181.                 dt.Rows.Add(dr); 
182.             } 
183.  
184.             #endregion   
185.  
186.             return dt; 
187.         } 
188.  
189.  
190.     } 
191.  
192.  
193.     /// <summary> 
194.     /// 批量导入mysql帮助类 
195.     /// </summary> 
196.     public static class MySqlHelper 
197.     { 
198.         /// <summary> 
199.         /// MySqlBulkLoader批量导入 
200.         /// </summary> 
201.         /// <param name="_mySqlConnection">数据库连接地址</param> 
202.         /// <param name="table"></param> 
203.         /// <param name="csvName"></param> 
204.         /// <returns></returns> 
205.         public static int BulkLoad(MySqlConnection _mySqlConnection, DataTable table, string csvName) 
206.         { 
207.             var columns = table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList(); 
208.             MySqlBulkLoader bulk = new MySqlBulkLoader(_mySqlConnection) 
209.             { 
210.                 FieldTerminator = "\t", 
211.                 FieldQuotationCharacter = '"', 
212.                 EscapeCharacter = '"', 
213.                 LineTerminator = "\r\n", 
214.                 FileName = csvName, 
215.                 NumberOfLinesToSkip = 0, 
216.                 TableName = table.TableName, 
217.  
218.             }; 
219.  
220.             bulk.Columns.AddRange(columns); 
221.             return bulk.Load(); 
222.         } 
223.     } 
224.  
225.  
226.     /// <summary> 
227.     /// csv扩展 
228.     /// </summary> 
229.     public static class CSVEx 
230.     { 
231.         /// <summary> 
232.         ///将DataTable转换为标准的CSV文件 
233.         /// </summary> 
234.         /// <param name="table">数据表</param> 
235.         /// <param name="tmpPath">文件地址</param> 
236.         /// <returns>返回标准的CSV</returns> 
237.         public static void ToCsv(this DataTable table, string tmpPath) 
238.         { 
239.             //以半角逗号（即,）作分隔符，列为空也要表达其存在。 
240.             //列内容如存在半角逗号（即,）则用半角引号（即""）将该字段值包含起来。 
241.             //列内容如存在半角引号（即"）则应替换成半角双引号（""）转义，并用半角引号（即""）将该字段值包含起来。 
242.             StringBuilder sb = new StringBuilder(); 
243.             DataColumn colum; 
244.             foreach (DataRow row in table.Rows) 
245.             { 
246.                 for (int i = 0; i < table.Columns.Count; i++) 
247.                 { 
248.                     Type _datatype = typeof(DateTime); 
249.                     colum = table.Columns[i]; 
250.                     if (i != 0) sb.Append("\t"); 
251.                     //if (colum.DataType == typeof(string) && row[colum].ToString().Contains(",")) 
252.                     //{ 
253.                     //    sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\""); 
254.                     //} 
255.                     if (colum.DataType == _datatype) 
256.                     { 
257.                         sb.Append(((DateTime)row[colum]).ToString("yyyy/MM/dd HH:mm:ss")); 
258.                     } 
259.                     else sb.Append(row[colum].ToString()); 
260.                 } 
261.                 sb.Append("\r\n"); 
262.             } 
263.             StreamWriter sw = new StreamWriter(tmpPath, false, UTF8Encoding.UTF8); 
264.             sw.Write(sb.ToString()); 
265.             sw.Close(); 
266.         } 
267.  
268.     } 
269.  
270.     /// <summary> 
271.     /// 字符串转化 
272.     /// </summary> 
273.     public static class StringExtensions 
274.     { 
275.         /// <summary> 
276.         /// 转换为 main_keys_id 这种形式的字符串方式 
277.         /// </summary> 
278.         public static string ToSnakeCase(this string input) 
279.         { 
280.             if (string.IsNullOrEmpty(input)) { return input; } 
281.  
282.             var startUnderscores = Regex.Match(input, @"^_+"); 
283.             return startUnderscores + Regex.Replace(input, @"([a-z0-9])([A-Z])", "$1_$2").ToLower(); 
284.         } 
285.     } 
286.  
287.  
288.     /// <summary> 
289.     /// 实体 
290.     /// </summary> 
291.     public class CrmCouponTestDto 
292.     { 
293.         /// <summary> 
294.         /// ID 
295.         /// </summary> 
296.         public long Id { get; set; } 
297.  
298.         /// <summary> 
299.         /// 卡券号 
300.         /// </summary>      
301.         public string CouponCode { get; set; } 
302.  
303.         /// <summary> 
304.         /// 卡券ID 
305.         /// </summary> 
306.         public int CouponId { get; set; } 
307.  
308.         /// <summary> 
309.         /// 会员ID 
310.         /// </summary> 
311.         public int MemberId { get; set; } 
312.  
313.         /// <summary> 
314.         /// 发放时间 
315.         /// </summary>    
316.         public DateTime IssueTime { get; set; } 
317.  
318.         /// <summary> 
319.         /// 使用时间 
320.         /// </summary>       
321.         public DateTime UsageTime { get; set; } 
322.  
323.         /// <summary> 
324.         /// 使用店铺ID 
325.         /// </summary>       
326.  
327.         public int UsageShopId { get; set; } 
328.  
329.         /// <summary> 
330.         /// 使用单号 
331.         /// </summary>       
332.         public string UsageBillNo { get; set; } 
333.  
334.         /// <summary> 
335.         /// 有效开始时间 
336.         /// </summary>       
337.         public DateTime EffectiveStart { get; set; } 
338.  
339.         /// <summary> 
340.         /// 有效结束时间 
341.         /// </summary>       
342.         public DateTime EffectiveEnd { get; set; } 
343.  
344.         /// <summary> 
345.         /// 状态 
346.         /// CouponStatus 卡券状态: 
347.         /// -1：未领用 
348.         /// 0：未使用 
349.         /// 1：已使用 
350.         /// 2：已过期 
351.         ///3：已作废 
352.         ///4：转赠中 
353.         /// </summary> 
354.  
355.         public Int16 Status { get; set; } 
356.     } 
357. }