批量導入/更新 數據的幾種方法

在工作中經常遇到要導入數據的場景。

1 導入全新的數據

2 導入的數據中庫中可能已經存在，已經存在的數據不能覆蓋，不能變更

3 導入的數據中庫中可能已經存在，已經存在的數據主鍵等不能變化，同時需要更新這些數據的一些字段（比如：積分字段）

 

第一種最簡單，最坑爹的是第三種了。

新手碰到這種問題往往一籌莫展，能想到的最直接的辦法就是 一條條獲取 ，然后判斷，然后 update。這種方式的效率在少量數據的時候還沒太大問題，當達到上千上萬條時候問題就會出現了。

這種處理方式的資源開銷和效率簡直“慘絕人寰”。我也經歷過這種方式，並漸漸的了解是應用了更高效的方式。

一、一條條的插入/獲取判斷和更新

這種方式上文已經說了效率和資源開銷都是最大的，沒啥好講的，做程序的都會

二、一次插入多條/一次更新多條

插入多條：即一個connection.open()  后執行多個 command命令 。參考代碼如下：

　　

 public static void ExecuteSqlTran(List<string> SQLStringList)//SQLs
        {
            using (SqlConnection conn = new SqlConnection(connectionString))
            {
                conn.Open();
                SqlCommand cmd = new SqlCommand();
                cmd.Connection = conn;
                SqlTransaction tx = conn.BeginTransaction();
                cmd.Transaction = tx;
                try
                {
                    for (int n = 0; n < SQLStringList.Count; n++)
                    {
                        string strsql = SQLStringList[n];
                        if (strsql.Trim().Length > 1)
                        {
                            cmd.CommandText = strsql;
                            cmd.ExecuteNonQuery();
                        }
                    }
                    tx.Commit();
                }
                catch (System.Data.SqlClient.SqlException E)
                {
                    tx.Rollback();
                    throw new Exception(E.Message);
                }
            }
        }

這種的方式效率會比  一  高很多，但是資源開銷和執行效率還是難以讓人忍受，在一次執行語句數量達到萬級時候就表現的很明顯。

我工作中的一次案例數據庫中數據有2300w條記錄，要對比的數據有20w條記錄時候  純update語句 1w條數據時候需要等待100s左右，當要對比的數據為40w條時候需要等待時間是600s左右。這種方式在效率上是相當讓人無法忍受的。

三、使用DataSet更新記錄

參考代碼如下：

　　

/// <summary>
        /// 利用dataset批量更新數據
        /// </summary>
        /// <param name="sqlString"></param>

        public void BatchUpDataForDataset(string sqlString)
        {
            using (SqlConnection connection = new SqlConnection(connectionString))
            {


                using (SqlDataAdapter da = new SqlDataAdapter(sqlString, connection))
                {
                    DataSet ds = new DataSet();
                    try
                    {
                        da.Fill(ds);
                        //交給委托處理
                        if (DataUpWork != null)
                        {
                            DataUpWork(ds);
                        }

                        SqlCommandBuilder scb = new SqlCommandBuilder(da);
                        //執行更新
                        da.InsertCommand = scb.GetUpdateCommand();
                        da.Update(ds);
                        //使DataTable保存更新
                        ds.AcceptChanges();

                    }
                    catch (System.Data.SqlClient.SqlException ex)
                    {
                        throw new Exception(ex.Message);
                    }
                }
            }
        }

在代碼中我使用了委托DataUpWork來處理dataset中的數據，隨后提交。

這種方式寫代碼可能會輕松很多，但是效率也是不盡人意，好像最終也是生成update語句批量執行的，具體沒有細細研究，請知道的大神指出這種方式的工作原理。

這種方式同樣不能同時執行插入數據的操作（個人沒測試過，不知道對DataSet新增的數據會不會插入到數據庫中。(/ □ \)）

四、 使用SqlBulkCopy批量插入全新的數據

這個批量插入效率真的極高，插入10w也是瞬間完成，參考代碼

  /// <summary>
        /// 使用原列列表、目標列表和目標表明及數據記錄數在指定的時間將數據批量導入到數據庫
        /// </summary>
        /// <param name="sourceColumnName">源列name</param>
        /// <param name="dbTableColumnName">目標列name</param>
        /// <param name="tableName">目標表名</param>
        /// <param name="sourceDt">數據源</param>
        /// <param name="timeOut">指定時間</param>
        public static void BatchInput(List<string> sourceColumnName, List<string> dbTableColumnName, string tableName, DataTable sourceDt, int timeOut = 300 ,string  exSql="")
        {
            if (sourceColumnName.Count != dbTableColumnName.Count)
            {
                throw new Exception("傳遞的量表數據數不匹配");
            }


            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;

            try
            {
                sqlConn.Open();
                if (!string.IsNullOrEmpty(exSql))
                {
                    SqlCommand cmd = new SqlCommand(exSql, sqlConn);
                    cmd.ExecuteNonQuery();
                }

                if (bulkCopy.BatchSize != 0)
                {
                    bulkCopy.WriteToServer(sourceDt);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {

                //釋放資源
              
                if (bulkCopy != null)
                {
                    bulkCopy.Close();
                }

                sqlConn.Close();
            }


        }

但是該方法只能插入全新的記錄，對於已經存在的數據無能為力了，不會去除已經存在的記錄項，更不能更新已經存在的記錄項 😂

五、使用臨時表作為中介來進行導入和更新

這種方式效率極高，資源開銷我只能說不是很大，開銷主要在數據庫上相對（本人技術菜不會監測開銷情況），我工作中的一個案例：數據庫表有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 管理工具進行導入  /捂臉