C# 之 批量插入數據到 SQLServer 中

　　創建一個用來測試的數據庫和表，為了讓插入數據更快，表中主鍵采用的是GUID，表中沒有創建任何索引。GUID必然是比自增長要快。而如果存在索引的情況下，每次插入記錄都會進行索引重建，這是非常耗性能的。如果表中無可避免的存在索引，我們可以通過先刪除索引，然后批量插入，最后再重建索引的方式來提高效率。

create database CarSYS;    
go    
use CarSYS;    
go 
CREATE TABLE Product(
Id UNIQUEIDENTIFIER PRIMARY KEY,
NAME VARCHAR(50) NOT NULL,
Price DECIMAL(18,2) NOT NULL
)

　　通過SQL腳本來插入數據，常見如下四種方式。

　　一、通過 INSERT INTO tname (...) VALUES (...) 逐條插入，性能最差，不建議使用。

　　示例：

#region 方式一
        static void InsertOne()
        {
            Console.WriteLine("采用一條一條插入的方式實現");
            Stopwatch sw = new Stopwatch();

            using (SqlConnection conn = new SqlConnection(StrConnMsg)) //using中會自動Open和Close 連接。
            {
                string sql = "INSERT INTO Product(Id,Name,Price) VALUES(newid(),@p,@d)";
                conn.Open();
                for (int i = 0; i < totalRow; i++)
                {
                    using (SqlCommand cmd = new SqlCommand(sql, conn))
                    {
                        cmd.Parameters.AddWithValue("@p", "商品" + i);
                        cmd.Parameters.AddWithValue("@d", i);
                        sw.Start();
                        cmd.ExecuteNonQuery();
                        Console.WriteLine(string.Format("插入一條記錄，已耗時{0}毫秒", sw.ElapsedMilliseconds));
                    }
                    if (i == getRow)
                    {
                        sw.Stop();
                        break;
                    }
                }
            }
            Console.WriteLine(string.Format("插入{0}條記錄，每{4}條的插入時間是{1}毫秒,預估總得插入時間是{2}毫秒，{3}分鍾",
 totalRow, sw.ElapsedMilliseconds, ((sw.ElapsedMilliseconds / getRow) * totalRow), GetMinute((sw.ElapsedMilliseconds / getRow * totalRow)), getRow));
        }
        static int GetMinute(long l)
        {
            return (Int32)l / 60000;
        } 
        #endregion

　　運行結果如下：

 　　

　　結果顯示插入100w條記錄，預計需要50分鍾時間，每插入一條記錄大概需要3毫秒左右。

　　二、使用 insert bulk

BULK INSERT [ [ 'database_name'.][ 'owner' ].]{ 'table_name' FROM 'data_file' }       
    WITH  (   
            [ BATCHSIZE [ = batch_size ] ],       
            [ CHECK_CONSTRAINTS ],           
            [ CODEPAGE [ = 'ACP' | 'OEM' | 'RAW' | 'code_page' ] ],   
            [ DATAFILETYPE [ = 'char' | 'native'| 'widechar' | 'widenative' ] ],               
            [ FIELDTERMINATOR [ = 'field_terminator' ] ],   
            [ FIRSTROW [ = first_row ] ],   
            [ FIRE_TRIGGERS ],   
            [ FORMATFILE = 'format_file_path' ],   
            [ KEEPIDENTITY ],   
            [ KEEPNULLS ],   
            [ KILOBYTES_PER_BATCH [ = kilobytes_per_batch ] ],      
            [ LASTROW [ = last_row ] ],   
            [ MAXERRORS [ = max_errors ] ],   
            [ ORDER ( { column [ ASC | DESC ] } [ ,...n ] ) ],     
            [ ROWS_PER_BATCH [ = rows_per_batch ] ],   
            [ ROWTERMINATOR [ = 'row_terminator' ] ],             
            [ TABLOCK ],   
    )  

　　相關參數說明：

BULK INSERT    
   [ database_name . [ schema_name ] . | schema_name . ] [ table_name | view_name ]    
      FROM 'data_file'    
     [ WITH    
        (    
   [ [ , ] BATCHSIZE = batch_size ]    --BATCHSIZE指令來設置在單個事務中可以插入到表中的記錄的數量   
   [ [ , ] CHECK_CONSTRAINTS ]     --指定在大容量導入操作期間，必須檢查所有對目標表或視圖的約束。若沒有 CHECK_CONSTRAINTS 選項，則所有 CHECK 和 FOREIGN KEY 約束都將被忽略，並且在此操作之后表的約束將標記為不可信。   
   [ [ , ] CODEPAGE = { 'ACP' | 'OEM' | 'RAW' | 'code_page' } ]  --指定該數據文件中數據的代碼頁   
   [ [ , ] DATAFILETYPE =    
      { 'char' | 'native'| 'widechar' | 'widenative' } ]  --指定 BULK INSERT 使用指定的數據文件類型值執行導入操作。   
   [ [ , ] FIELDTERMINATOR = 'field_terminator' ]  --標識分隔內容的符號   
   [ [ , ] FIRSTROW = first_row ]    --指定要加載的第一行的行號。默認值是指定數據文件中的第一行   
   [ [ , ] FIRE_TRIGGERS ]     --是否啟動觸發器   
   [ [ , ] FORMATFILE = 'format_file_path' ]    
   [ [ , ] KEEPIDENTITY ]   --指定導入數據文件中的標識值用於標識列   
   [ [ , ] KEEPNULLS ]    --指定在大容量導入操作期間空列應保留一個空值，而不插入用於列的任何默認值   
   [ [ , ] KILOBYTES_PER_BATCH = kilobytes_per_batch ]    
   [ [ , ] LASTROW = last_row ]   --指定要加載的最后一行的行號   
   [ [ , ] MAXERRORS = max_errors ]   --指定允許在數據中出現的最多語法錯誤數，超過該數量后將取消大容量導入操作。   
   [ [ , ] ORDER ( { column [ ASC | DESC ] } [ ,...n ] ) ]  --指定數據文件中的數據如何排序   
   [ [ , ] ROWS_PER_BATCH = rows_per_batch ]    
   [ [ , ] ROWTERMINATOR = 'row_terminator' ]   --標識分隔行的符號   
   [ [ , ] TABLOCK ]     --指定為大容量導入操作持續時間獲取一個表級鎖   
   [ [ , ] ERRORFILE = 'file_name' ]   --指定用於收集格式有誤且不能轉換為 OLE DB 行集的行的文件。   
        )]

　　

#region 方式二
        static void InsertTwo()
        {
            Console.WriteLine("使用Bulk插入的實現方式");
            Stopwatch sw = new Stopwatch();
            DataTable dt = GetTableSchema(); 

            using (SqlConnection conn = new SqlConnection(StrConnMsg))
            {
                SqlBulkCopy bulkCopy = new SqlBulkCopy(conn);
                bulkCopy.DestinationTableName = "Product";
                bulkCopy.BatchSize = dt.Rows.Count;
                conn.Open();
                sw.Start();

                for (int i = 0; i < totalRow;i++ )
                {
                    DataRow dr = dt.NewRow();
                    dr[0] = Guid.NewGuid();
                    dr[1] = string.Format("商品", i);
                    dr[2] = (decimal)i;
                    dt.Rows.Add(dr);
                }
                    if (dt != null && dt.Rows.Count != 0)
                    {
                        bulkCopy.WriteToServer(dt);
                        sw.Stop();
                    }
                    Console.WriteLine(string.Format("插入{0}條記錄共花費{1}毫秒，{2}分鍾", totalRow, sw.ElapsedMilliseconds, GetMinute(sw.ElapsedMilliseconds)));
            }
        }
        static DataTable GetTableSchema()
        {
            DataTable dt = new DataTable();
            dt.Columns.AddRange(new DataColumn[] {   
        new DataColumn("Id",typeof(Guid)),  
        new DataColumn("Name",typeof(string)),  
        new DataColumn("Price",typeof(decimal))});
            return dt;
        }
        #endregion

　　運行結果如下：

　　

　　插入100w條記錄才8s多，推薦。

　　打開Sqlserver Profiler跟蹤，會發現執行的是如下語句：

insert bulk Product ([Id] UniqueIdentifier, [NAME] VarChar(50) COLLATE Chinese_PRC_CI_AS, [Price] Decimal(18,2))

　　三、使用 INSERT INTO xx select...

INSERT INTO Product(Id,Name,Price)
 SELECT NEWID(),'牛欄1段',160 
 UNION ALL 
 SELECT NEWID(),'牛欄2段',180
 UNION ALL
......

　　從sqlserver 2008起開始支持TVPs。創建緩存表ProductTemp ，執行如下SQL：

CREATE TYPE ProductTemp AS  TABLE(
Id UNIQUEIDENTIFIER PRIMARY KEY,
NAME VARCHAR(50) NOT NULL,
Price DECIMAL(18,2) NOT NULL
)

　　執行完成之后，會發現在數據庫CarSYS下面多了一張緩存表ProductTemp

　　

　　

　　可見插入100w條記錄共花費了11秒多，此方式麻煩，一次性可使用。　　

　　四、拼接SQL

INSERT INTO Product(Id,Name,Price) VALUES
(newid(),'牛欄1段',160)
,(newid(),'牛欄2段',260)
......

 　　示例：此種方法在C#中有限制，一次性只能批量插入1000條，所以就得分段進行插入。

#region 方式四
        static void InsertFour()
        {
            Console.WriteLine("采用拼接批量SQL插入的方式實現");
            Stopwatch sw = new Stopwatch();

            using (SqlConnection conn = new SqlConnection(StrConnMsg)) //using中會自動Open和Close 連接。
            {
                conn.Open();
                sw.Start();
                for (int j = 0; j < totalRow / getRow;j++ )
                {
                    StringBuilder sb = new StringBuilder();
                    sb.Append("INSERT INTO Product(Id,Name,Price) VALUES");
                    using (SqlCommand cmd = new SqlCommand())
                    {
                       
                        for (int i = 0; i < getRow; i++)
                        {
                            sb.AppendFormat("(newid(),'商品{0}',{0}),", j*i+i);
                        }
                        cmd.Connection = conn;
                        cmd.CommandText = sb.ToString().TrimEnd(',');
                        cmd.ExecuteNonQuery();
                    }
                }
                sw.Stop();
                Console.WriteLine(string.Format("插入{0}條記錄，共耗時{1}毫秒",totalRow,sw.ElapsedMilliseconds));
            }
        }
        #endregion

　　運行結果如下：

　　

　　可以看到大概花費了10分鍾。雖然在方式一的基礎上，性能有了較大的提升，但是顯然還是不夠快。

　　總結：大數據批量插入方式一和方式四盡量避免使用，而方式二和方式三都是非常高效的批量插入數據方式。其都是通過構建DataTable的方式插入的，而我們知道DataTable是存在內存中的，所以當數據量特別特別大，大到內存中無法一次性存儲的時候，可以分段插入。比如需要插入9千萬條數據，可以分成9段進行插入，一次插入1千萬條。而在for循環中直接進行數據庫操作，我們是應該盡量避免的。每一次數據庫的連接、打開和關閉都是比較耗時的，雖然在C#中存在數據庫連接池，也就是當我們使用using或者conn.Close()，進行釋放連接時，其實並沒有真正關閉數據庫連接，它只是讓連接以類似於休眠的方式存在，當再次操作的時候，會從連接池中找一個休眠狀態的連接，喚醒它，這樣可以有效的提高並發能力，減少連接損耗。而連接池中的連接數，我們都是可以配置的。