批量导入/更新 数据的几种方法
Posted Myisqq
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了批量导入/更新 数据的几种方法相关的知识，希望对你有一定的参考价值。
在工作中经常遇到要导入数据的场景。
1 导入全新的数据
2 导入的数据中库中可能已经存在，已经存在的数据不能覆盖，不能变更
3 导入的数据中库中可能已经存在，已经存在的数据主键等不能变化，同时需要更新这些数据的一些字段（比如：积分字段）
第一种最简单，最坑爹的是第三种了。
新手碰到这种问题往往一筹莫展，能想到的最直接的办法就是 一条条获取 ，然后判断，然后 update。这种方式的效率在少量数据的时候还没太大问题，当达到上千上万条时候问题就会出现了。
这种处理方式的资源开销和效率简直“惨绝人寰”。我也经历过这种方式，并渐渐的了解是应用了更高效的方式。
一、一条条的插入/获取判断和更新
这种方式上文已经说了效率和资源开销都是最大的，没啥好讲的，做程序的都会
二、一次插入多条/一次更新多条
插入多条：即一个connection.open() 后执行多个 command命令 。参考代码如下：
1 public static void ExecuteSqlTran(List<string> SQLStringList)//SQLs 2 { 3 using (SqlConnection conn = new SqlConnection(connectionString)) 4 { 5 conn.Open(); 6 SqlCommand cmd = new SqlCommand(); 7 cmd.Connection = conn; 8 SqlTransaction tx = conn.BeginTransaction(); 9 cmd.Transaction = tx; 10 try 11 { 12 for (int n = 0; n < SQLStringList.Count; n++) 13 { 14 string strsql = SQLStringList[n]; 15 if (strsql.Trim().Length > 1) 16 { 17 cmd.CommandText = strsql; 18 cmd.ExecuteNonQuery(); 19 } 20 } 21 tx.Commit(); 22 } 23 catch (System.Data.SqlClient.SqlException E) 24 { 25 tx.Rollback(); 26 throw new Exception(E.Message); 27 } 28 } 29 }
这种的方式效率会比 一 高很多，但是资源开销和执行效率还是难以让人忍受，在一次执行语句数量达到万级时候就表现的很明显。
我工作中的一次案例数据库中数据有2300w条记录，要对比的数据有20w条记录时候 纯update语句 1w条数据时候需要等待100s左右，当要对比的数据为40w条时候需要等待时间是600s左右。这种方式在效率上是相当让人无法忍受的。
三、使用DataSet更新记录
参考代码如下：
1 /// <summary> 2 /// 利用dataset批量更新数据 3 /// </summary> 4 /// <param name="sqlString"></param> 5 6 public void BatchUpDataForDataset(string sqlString) 7 { 8 using (SqlConnection connection = new SqlConnection(connectionString)) 9 { 10 11 12 using (SqlDataAdapter da = new SqlDataAdapter(sqlString, connection)) 13 { 14 DataSet ds = new DataSet(); 15 try 16 { 17 da.Fill(ds); 18 //交给委托处理 19 if (DataUpWork != null) 20 { 21 DataUpWork(ds); 22 } 23 24 SqlCommandBuilder scb = new SqlCommandBuilder(da); 25 //执行更新 26 da.InsertCommand = scb.GetUpdateCommand(); 27 da.Update(ds); 28 //使DataTable保存更新 29 ds.AcceptChanges(); 30 31 } 32 catch (System.Data.SqlClient.SqlException ex) 33 { 34 throw new Exception(ex.Message); 35 } 36 } 37 } 38 }
在代码中我使用了委托DataUpWork来处理dataset中的数据，随后提交。
这种方式写代码可能会轻松很多，但是效率也是不尽人意，好像最终也是生成update语句批量执行的，具体没有细细研究，请知道的大神指出这种方式的工作原理。
这种方式同样不能同时执行插入数据的操作（个人没测试过，不知道对DataSet新增的数据会不会插入到数据库中。(/ □ \)）
四、 使用SqlBulkCopy批量插入全新的数据
这个批量插入效率真的极高，插入10w也是瞬间完成，参考代码
1 /// <summary> 2 /// 使用原列列表、目标列表和目标表明及数据记录数在指定的时间将数据批量导入到数据库 3 /// </summary> 4 /// <param name="sourceColumnName">源列name</param> 5 /// <param name="dbTableColumnName">目标列name</param> 6 /// <param name="tableName">目标表名</param> 7 /// <param name="sourceDt">数据源</param> 8 /// <param name="timeOut">指定时间</param> 9 public static void BatchInput(List<string> sourceColumnName, List<string> dbTableColumnName, string tableName, DataTable sourceDt, int timeOut = 300 ,string exSql="") 10 { 11 if (sourceColumnName.Count != dbTableColumnName.Count) 12 { 13 throw new Exception("传递的量表数据数不匹配"); 14 } 15 16 17 SqlConnection sqlConn = new SqlConnection(connectionString); 18 SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn); 19 bulkCopy.BulkCopyTimeout = timeOut; 20 21 //目标表 22 if (string.IsNullOrEmpty(tableName)) 23 { 24 throw new ArgumentNullException(" 无效的目标表名：" + tableName); 25 } 26 27 bulkCopy.DestinationTableName = tableName; 28 29 30 31 for (int i = 0; i < sourceColumnName.Count; i++) 32 { 33 //原列和目标列 34 bulkCopy.ColumnMappings.Add(sourceColumnName[i], dbTableColumnName[i]); 35 36 } 37 //导入的数据量 38 bulkCopy.BatchSize = sourceDt.Rows.Count; 39 40 try 41 { 42 sqlConn.Open(); 43 if (!string.IsNullOrEmpty(exSql)) 44 { 45 SqlCommand cmd = new SqlCommand(exSql, sqlConn); 46 cmd.ExecuteNonQuery(); 47 } 48 49 if (bulkCopy.BatchSize != 0) 50 { 51 bulkCopy.WriteToServer(sourceDt); 52 } 53 } 54 catch (Exception ex) 55 { 56 throw ex; 57 } 58 finally 59 { 60 61 //释放资源 62 63 if (bulkCopy != null) 64 { 65 bulkCopy.Close(); 66 } 67 68 sqlConn.Close(); 69 } 70 71 72 }
但是该方法只能插入全新的记录，对于已经存在的数据无能为力了，不会去除已经存在的记录项，更不能更新已经存在的记录项 ??
五、使用临时表作为中介来进行导入和更新
这种方式效率极高，资源开销我只能说不是很大，开销主要在数据库上相对（本人技术菜不会监测开销情况），我工作中的一个案例：数据库表有2500w数据，需要导入处理的有15w数据，要对比的有45w历史数据。执行时间大约在30s内，就导入了新记录和更新了历史记录。
参考代码如下：
/// <summary> /// 使用原列列表、目标列表和目标表明及数据记录数在指定的时间将数据批量导入到数据库 /// </summary> /// <param name="sourceColumnName">源列name</param> /// <param name="dbTableColumnName">目标列name</param> /// <param name="tableName">目标表名</param> /// <param name="sourceDt">数据源</param> /// <param name="timeOut">指定时间</param> void BatchInput(List<string> sourceColumnName, List<string> dbTableColumnName, string tableName, DataTable sourceDt, out int olduser, out int newuser, int timeOut = 300, string exSql = "") { olduser = 0; newuser = 0; if (sourceColumnName.Count != dbTableColumnName.Count) { throw new Exception("传递的量表数据数不匹配"); } string connectionString = ConfigurationManager.ConnectionStrings["sqlconn"].ToString(); SqlConnection sqlConn = new SqlConnection(connectionString); SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn); bulkCopy.BulkCopyTimeout = timeOut; //目标表 if (string.IsNullOrEmpty(tableName)) { throw new ArgumentNullException(" 无效的目标表名：" + tableName); } bulkCopy.DestinationTableName = tableName; for (int i = 0; i < sourceColumnName.Count; i++) { //原列和目标列 bulkCopy.ColumnMappings.Add(sourceColumnName[i], dbTableColumnName[i]); } //导入的数据量 bulkCopy.BatchSize = sourceDt.Rows.Count; sqlConn.Open(); SqlTransaction transaction = null; try { SqlCommand cmd = sqlConn.CreateCommand(); cmd.CommandText = create table #a( mobile varchar(50),a int , idno int , b int ,c int ,d int )"; //创建临时表 cmd.ExecuteNonQuery(); if (bulkCopy.BatchSize != 0) { bulkCopy.WriteToServer(sourceDt); } transaction = sqlConn.BeginTransaction("SampleTransaction"); cmd.Transaction = transaction; //设置事务 //更新老用户 cmd.CommandText = " select #a.*, UserTable.UserID into #cz from #a left join UserTable on UserTable.a=4 and UserTable.b=" + ActivityId + " and UserTable.c=#a.mobile where UserID is not null"; //创建老用户数据表 cmd.ExecuteNonQuery(); cmd.CommandText = " update UserTable set b+=b,c=#cz.c from #cz ,UserTable where UserTable.a=4 and UserTable.b=" + ActivityId + " and #cz.userid=UserTable.UserID "; //更新老用户数据 cmd.ExecuteNonQuery(); cmd.CommandText = " select #a.*, UserTable.UserID into #newuser from #a left join UserTable on UserTable.a=4 and UserTable.b=" + ActivityId + " and UserTable.c=#a.mobile where UserID is null "; //创建新用户数据表 cmd.ExecuteNonQuery(); cmd.CommandText = " insert into UserTable (a,b,[c],d,e,f) select a,b,[c],d,e,f from #newuser "; //插入老用户 cmd.ExecuteNonQuery(); cmd.CommandText = " select count(1) from #newuser "; //新用户数量 newuser = (int)(cmd.ExecuteScalar()); cmd.CommandText = " select count(1) from #cz "; //老用户数量 olduser = (int)(cmd.ExecuteScalar()); transaction.Commit(); WriteMessage("导入完毕，共导入" + sourceDt.Rows.Count + "条记录,其中新增用户：" + newuser + "条，老用户：" + olduser + "条"); } catch (Exception ex) { bulkCopy.Close(); if (transaction != null) { transaction.Rollback(); } WriteMessage("反生错误，导入失败：" + ex.Message + "\r\n" + ex.StackTrace); } finally { sqlConn.Close(); } }
原理： 首先创建临时表，然后用SqlBulkCopy 将数据批量导入到临时表中，但后用临时表 联合 数据库存表查询 得到老用户 数据，存储到临时表 #cz中，然后更新老用户的相关字段
在联合库存表查询出新用户存储到临时表 #newuser 中 ，然后将数据 insert into select 库存表中
六、大杀器
如果只有全新的记录并且你又能接触到数据库那么请用excel等文件，直接用sql server 管理工具进行导入 /捂脸