如何利用MySqlBulkLoader实现批量插入数据

这篇文章主要介绍“如何利用MySqlBulkLoader实现批量插入数据”，在日常操作中，相信很多人在如何利用MysqlBulkLoader实现批量插入数据问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”如何利用mysqlBulkLoader实现批量插入数据”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

介绍

MysqlBulkLoader主要的实现方式：将需要插入的数据转成DataTable，DataTable转成一个CSV文件，将CSV文件使用批量导入的形式导入到数据库里面去。

注意：

1）.数据库连接地址需要添加配置AllowLoadLocalInfile=true,允许本地文件导入；

Data Source = 数据库地址; Port = 端口; Initial Catalog = 数据库名; User Id = 用户名; PassWord = 密码;AllowLoadLocalInfile=true;

2）.插入的时候会返回插入行数，但是检查所有的数据都正确，也没有报异常，却返回了插入数量为0，可以检查表是否有唯一索引，插入的数据是否违反了唯一索引。

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; }    }}

到此，关于“如何利用MySqlBulkLoader实现批量插入数据”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注编程网网站，小编会继续努力为大家带来更多实用的文章！