使用SqlBulkCopy進行大批量數據遷移

下面咱們先經過一個簡單的例子說明這個類的使用：   

DateTime startTime; protected void Button1_Click(object sender, EventArgs e) { startTime = DateTime.Now; string SrcConString; string DesConString; SqlConnection SrcCon = new SqlConnection(); SqlConnection DesCon = new SqlConnection(); SqlCommand SrcCom = new SqlCommand(); SqlDataAdapter SrcAdapter = new SqlDataAdapter(); DataTable dt = new DataTable(); SrcConString = ConfigurationManager.ConnectionStrings["SrcDBConnectionString"].ConnectionString; DesConString = ConfigurationManager.ConnectionStrings["DesDBConnectionString"].ConnectionString; SrcCon.ConnectionString = SrcConString; SrcCom.Connection = SrcCon; SrcCom.CommandText = " SELECT * From [SrcTable]"; SrcCom.CommandType = CommandType.Text; SrcCom.Connection.Open(); SrcAdapter.SelectCommand = SrcCom; SrcAdapter.Fill(dt); SqlBulkCopy DesBulkOp; DesBulkOp = new SqlBulkCopy(DesConString, SqlBulkCopyOptions.UseInternalTransaction); DesBulkOp.BulkCopyTimeout = 500000000; DesBulkOp.SqlRowsCopied += new SqlRowsCopiedEventHandler(OnRowsCopied); DesBulkOp.NotifyAfter = dt.Rows.Count; try { DesBulkOp.DestinationTableName = "SrcTable"; DesBulkOp.WriteToServer(dt); } catch (Exception ex) { } finally { SrcCon.Close(); DesCon.Close(); } } private void OnRowsCopied(object sender, SqlRowsCopiedEventArgs args) { lblCounter.Text += args.RowsCopied.ToString() + " rows are copied "; TimeSpan copyTime = DateTime.Now - startTime; lblCounter.Text += "Copy Time:" + copyTime.Seconds.ToString() + "." + copyTime.Milliseconds.ToString() + " seconds"; }

    

接着具體分析這幾行代碼：

SqlBulkCopy DesBulkOp; DesBulkOp = new SqlBulkCopy(DesConString, SqlBulkCopyOptions.UseInternalTransaction);

先生成 SqlBulkCopy實例，構造函數指定了目標數據庫，使用 SqlBulkCopyOptions.UseInternalTransaction是指遷移動做指定在一個Transaction當中，若是數據遷移中產生錯誤或異常將發生回滾。其餘選項請參考MSDN。

DesBulkOp.BulkCopyTimeout = 500000000;

指定操做完成的Timeout時間

DesBulkOp.SqlRowsCopied += new SqlRowsCopiedEventHandler(OnRowsCopied); DesBulkOp.NotifyAfter = dt.Rows.Count; try { DesBulkOp.DestinationTableName = "SrcTable"; DesBulkOp.WriteToServer(dt); }

NotifyAfter屬性指定通知事件前處理的數據行數，在這裏指定爲表的行數，並添加SqlRowsCopied事件輸出整個遷移過程的時間。WriteToServer方法就是將數據源拷備到目標數據庫。在使用WriteToServer方法以前必須先指定DestinationTableName屬性，也就是目標數據庫的表名，咱們還能夠本身定義一個Transaction，例如：

SqlTransaction Transaction; Transaction = SrcCom.Connection.BeginTransaction(); SqlBulkCopy DesBulkOp; DesBulkOp = new SqlBulkCopy(new SqlConnection(DesConString), SqlBulkCopyOptions.Default, Transaction); try { //.. } catch{} finally { Transaction.Commit(); }

另外還有一個 SqlBulkCopyColumnMapping類，能夠讓數據源字段映射到目標數據中命名不一樣的字段上。也就是說若是目標數據和源數據的列名不一樣時，能夠用這個類進行映射：

SqlBulkCopyColumnMapping ColMap = new SqlBulkCopyColumnMapping("SrcCol", "DesCol"); DesBulkOp.ColumnMappings.Add(ColMap);

或者能夠直接添加映射：

DesBulkOp.ColumnMappings.Add("SrcCol", "DesCol");

性能問題：

我使用上面的例子測試，遷移了2萬條左右的記錄，花的時間不到一秒，應該說性能仍是不錯的。另外，使用SQL Profile監視遷移事件，能夠看見請求記錄很是少，只有幾條而已。聽說使用 SqlBulkCopy能夠大大減小數據遷移的時間。