最近在專案中遇到插入資料瓶頸，幾萬、幾十萬、幾百萬的資料儲存到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
/// <summary> /// 將List轉化為DataTable /// </summary> /// <returns></returns> public DataTable ListToDataTable<T>(List<T> data) { #region 建立一個DataTable，以實體名稱作為DataTable名稱 var tableName = typeof(T).Name; tableName = tableName.ToSnakeCase(); /*實體名稱與表名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ DataTable dt = new DataTable { TableName = tableName }; #endregion #region 拿取列名，以實體的屬性名作為列名 var properties = typeof(T).GetProperties(); foreach (var item in properties) { var curFileName = item.Name; curFileName = curFileName.ToSnakeCase();/*列名與欄位名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ dt.Columns.Add(curFileName); } #endregion #region 列賦值 foreach (var item in data) { DataRow dr = dt.NewRow(); var columns = dt.Columns; var curPropertyList = item.GetType().GetProperties(); foreach (var p in curPropertyList) { var name = p.Name; name = name.ToSnakeCase();/*列名與欄位名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ var curValue = p.GetValue(item); int i = columns.IndexOf(name); dr[i] = curValue; } dt.Rows.Add(dr); } #endregion return dt; }
2.將DataTable轉換為標準的CSV檔案
/// <summary> /// csv擴充套件 /// </summary> public static class CSVEx { /// <summary> ///將DataTable轉換為標準的CSV檔案 /// </summary> /// <param name="table">資料表</param> /// <param name="tmpPath">檔案地址</param> /// <returns>返回標準的CSV</returns> public static void ToCsv(this DataTable table, string tmpPath) { //以半形逗號（即,）作分隔符，列為空也要表達其存在。 //列內容如存在半形逗號（即,）則用半形引號（即""）將該欄位值包含起來。 //列內容如存在半形引號（即"）則應替換成半形雙引號（""）跳脫，並用半形引號（即""）將該欄位值包含起來。 StringBuilder sb = new StringBuilder(); DataColumn colum; foreach (DataRow row in table.Rows) { for (int i = 0; i < table.Columns.Count; i++) { Type _datatype = typeof(DateTime); colum = table.Columns[i]; if (i != 0) sb.Append("\t"); //if (colum.DataType == typeof(string) && row[colum].ToString().Contains(",")) //{ // sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\""); //} if (colum.DataType == _datatype) { sb.Append(((DateTime)row[colum]).ToString("yyyy/MM/dd HH:mm:ss")); } else sb.Append(row[colum].ToString()); } sb.Append("\r\n"); } StreamWriter sw = new StreamWriter(tmpPath, false, UTF8Encoding.UTF8); sw.Write(sb.ToString()); sw.Close(); } }
3.CSV檔案匯入資料到資料庫
/// <summary> /// 批次匯入mysql幫助類 /// </summary> public static class MySqlHelper { /// <summary> /// MySqlBulkLoader批次匯入 /// </summary> /// <param name="_mySqlConnection">資料庫連線地址</param> /// <param name="table"></param> /// <param name="csvName"></param> /// <returns></returns> public static int BulkLoad(MySqlConnection _mySqlConnection, DataTable table, string csvName) { var columns = table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList(); MySqlBulkLoader bulk = new MySqlBulkLoader(_mySqlConnection) { FieldTerminator = "\t", FieldQuotationCharacter = '"', EscapeCharacter = '"', LineTerminator = "\r\n", FileName = csvName, NumberOfLinesToSkip = 0, TableName = table.TableName, }; bulk.Columns.AddRange(columns); return bulk.Load(); } }
4.使用MySqlBulkLoader批次插入資料
/// <summary> /// 使用MySqlBulkLoader批次插入資料 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="data"></param> /// <returns></returns> /// <exception cref="Exception"></exception> public int BulkLoaderData<T>(List<T> data) { if (data.Count <= 0) return 0; var connectString = "資料庫連線地址"; using (MySqlConnection connection = new MySqlConnection(connectString)) { MySqlTransaction sqlTransaction = null; try { if (connection.State == ConnectionState.Closed) { connection.Open(); } sqlTransaction = connection.BeginTransaction(); var dt = ListToDataTable<T>(data); //將List轉成dataTable string tmpPath = Path.GetTempFileName(); dt.ToCsv(tmpPath); //將DataTable轉成CSV檔案 var insertCount = MySqlHelper.BulkLoad(connection, dt, tmpPath); //使用MySqlBulkLoader插入資料 sqlTransaction.Commit(); try { if (File.Exists(tmpPath)) File.Delete(tmpPath); } catch (Exception) { //刪除檔案失敗 } return insertCount; //返回執行成功的條數 } catch (Exception e) { if (sqlTransaction != null) { sqlTransaction.Rollback(); } //執行異常 throw e; } } }
5.完整的程式碼：
namespace WebApplication1.BrantchInsert { /// <summary> /// 批次插入 /// </summary> public class BulkLoader { /// <summary> /// 測試批次插入入口 /// </summary> /// <returns></returns> public int BrantchDataTest() { #region 模擬資料 var data = new List<CrmCouponTestDto>() { new CrmCouponTestDto { Id=1, CouponCode="test001", CouponId = 1, MemberId=100, IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), UsageShopId=0, UsageBillNo="", EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), Status=0 }, new CrmCouponTestDto { Id=2, CouponCode="test002", CouponId = 1, MemberId=101, IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), UsageTime=Convert.ToDateTime("2022-06-27 14:30:00"), UsageShopId=2, UsageBillNo="CS202206271430001", EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), Status=1 }, new CrmCouponTestDto { Id=3, CouponCode="test003", CouponId = 1, MemberId=102, IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), UsageShopId=0, UsageBillNo="", EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), Status=0 }, new CrmCouponTestDto { Id=4, CouponCode="test004", CouponId = 1, MemberId=103, IssueTime=Convert.ToDateTime("2022-06-27 14:00:00"), UsageTime=Convert.ToDateTime("3000-12-31 00:00:00"), UsageShopId=0, UsageBillNo="", EffectiveStart=Convert.ToDateTime("2022-06-27 14:00:00"), EffectiveEnd=Convert.ToDateTime("2023-06-27 14:00:00"), Status=0 } }; #endregion var result = BulkLoaderData<CrmCouponTestDto>(data); return result; } /// <summary> /// 使用MySqlBulkLoader批次插入資料 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="data"></param> /// <returns></returns> /// <exception cref="Exception"></exception> public int BulkLoaderData<T>(List<T> data) { if (data.Count <= 0) return 0; var connectString = "資料庫連線地址"; using (MySqlConnection connection = new MySqlConnection(connectString)) { MySqlTransaction sqlTransaction = null; try { if (connection.State == ConnectionState.Closed) { connection.Open(); } sqlTransaction = connection.BeginTransaction(); var dt = ListToDataTable<T>(data); //將List轉成dataTable string tmpPath = Path.GetTempFileName(); dt.ToCsv(tmpPath); //將DataTable轉成CSV檔案 var insertCount = MySqlHelper.BulkLoad(connection, dt, tmpPath); //使用MySqlBulkLoader插入資料 sqlTransaction.Commit(); try { if (File.Exists(tmpPath)) File.Delete(tmpPath); } catch (Exception) { //刪除檔案失敗 } return insertCount; //返回執行成功的條數 } catch (Exception e) { if (sqlTransaction != null) { sqlTransaction.Rollback(); } //執行異常 throw e; } } } /// <summary> /// 將List轉化為DataTable核心方法 /// </summary> /// <returns></returns> public DataTable ListToDataTable<T>(List<T> data) { #region 建立一個DataTable，以實體名稱作為DataTable名稱 var tableName = typeof(T).Name; tableName = tableName.ToSnakeCase(); /*實體名稱與表名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ DataTable dt = new DataTable { TableName = tableName }; #endregion #region 拿取列名，以實體的屬性名作為列名 var properties = typeof(T).GetProperties(); foreach (var item in properties) { var curFileName = item.Name; curFileName = curFileName.ToSnakeCase();/*列名與欄位名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ dt.Columns.Add(curFileName); } #endregion #region 列賦值 foreach (var item in data) { DataRow dr = dt.NewRow(); var columns = dt.Columns; var curPropertyList = item.GetType().GetProperties(); foreach (var p in curPropertyList) { var name = p.Name; name = name.ToSnakeCase();/*列名與欄位名進行轉化，主要根據各專案的規定進行轉化,不一定就是我這些寫的這種轉換方式*/ var curValue = p.GetValue(item); int i = columns.IndexOf(name); dr[i] = curValue; } dt.Rows.Add(dr); } #endregion return dt; } } /// <summary> /// 批次匯入mysql幫助類 /// </summary> public static class MySqlHelper { /// <summary> /// MySqlBulkLoader批次匯入 /// </summary> /// <param name="_mySqlConnection">資料庫連線地址</param> /// <param name="table"></param> /// <param name="csvName"></param> /// <returns></returns> public static int BulkLoad(MySqlConnection _mySqlConnection, DataTable table, string csvName) { var columns = table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList(); MySqlBulkLoader bulk = new MySqlBulkLoader(_mySqlConnection) { FieldTerminator = "\t", FieldQuotationCharacter = '"', EscapeCharacter = '"', LineTerminator = "\r\n", FileName = csvName, NumberOfLinesToSkip = 0, TableName = table.TableName, }; bulk.Columns.AddRange(columns); return bulk.Load(); } } /// <summary> /// csv擴充套件 /// </summary> public static class CSVEx { /// <summary> ///將DataTable轉換為標準的CSV檔案 /// </summary> /// <param name="table">資料表</param> /// <param name="tmpPath">檔案地址</param> /// <returns>返回標準的CSV</returns> public static void ToCsv(this DataTable table, string tmpPath) { //以半形逗號（即,）作分隔符，列為空也要表達其存在。 //列內容如存在半形逗號（即,）則用半形引號（即""）將該欄位值包含起來。 //列內容如存在半形引號（即"）則應替換成半形雙引號（""）跳脫，並用半形引號（即""）將該欄位值包含起來。 StringBuilder sb = new StringBuilder(); DataColumn colum; foreach (DataRow row in table.Rows) { for (int i = 0; i < table.Columns.Count; i++) { Type _datatype = typeof(DateTime); colum = table.Columns[i]; if (i != 0) sb.Append("\t"); //if (colum.DataType == typeof(string) && row[colum].ToString().Contains(",")) //{ // sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\""); //} if (colum.DataType == _datatype) { sb.Append(((DateTime)row[colum]).ToString("yyyy/MM/dd HH:mm:ss")); } else sb.Append(row[colum].ToString()); } sb.Append("\r\n"); } StreamWriter sw = new StreamWriter(tmpPath, false, UTF8Encoding.UTF8); sw.Write(sb.ToString()); sw.Close(); } } /// <summary> /// 字串轉化 /// </summary> public static class StringExtensions { /// <summary> /// 轉換為 main_keys_id 這種形式的字串方式 /// </summary> public static string ToSnakeCase(this string input) { if (string.IsNullOrEmpty(input)) { return input; } var startUnderscores = Regex.Match(input, @"^_+"); return startUnderscores + Regex.Replace(input, @"([a-z0-9])([A-Z])", "$1_$2").ToLower(); } } /// <summary> /// 實體 /// </summary> public class CrmCouponTestDto { /// <summary> /// ID /// </summary> public long Id { get; set; } /// <summary> /// 卡券號 /// </summary> public string CouponCode { get; set; } /// <summary> /// 卡券ID /// </summary> public int CouponId { get; set; } /// <summary> /// 會員ID /// </summary> public int MemberId { get; set; } /// <summary> /// 發放時間 /// </summary> public DateTime IssueTime { get; set; } /// <summary> /// 使用時間 /// </summary> public DateTime UsageTime { get; set; } /// <summary> /// 使用店鋪ID /// </summary> public int UsageShopId { get; set; } /// <summary> /// 使用單號 /// </summary> public string UsageBillNo { get; set; } /// <summary> /// 有效開始時間 /// </summary> public DateTime EffectiveStart { get; set; } /// <summary> /// 有效結束時間 /// </summary> public DateTime EffectiveEnd { get; set; } /// <summary> /// 狀態 /// CouponStatus 卡券狀態: /// -1：未領用 /// 0：未使用 /// 1：已使用 /// 2：已過期 ///3：已作廢 ///4：轉贈中 /// </summary> public Int16 Status { get; set; } } }