SQLServer · 最佳實踐 · 數據庫實現大容量插入的幾種方式 · 數據庫內核月報

## 背景很多用戶在使用阿里云云數據庫SQL Server時，為了加快插入速度，都嘗試使用大容量插入的方式，大家都知道，對于完整恢復模式下的數據庫，大容量導入執行的所有行插入操作都會完整地記錄在事務日志中。如果使用完整恢復模式，大型數據導入會導致填充事務日志的速度很快。相反，對于簡單恢復模式或大容量日志恢復模式，大容量導入操作的按最小方式記錄日志減少了大容量導入操作填滿日志空間的可能性。另外，按最小方式記錄日志的效率也比按完整方式記錄日志高。但實際上，當大容量導入與數據庫鏡像共存時，會出現鏡像 Suspend的情況，這個情況是由于[微軟在2008 R2上的BUG](https://support.microsoft.com/en-us/kb/2700641)導致,微軟已經明確表示在2008 R2不會FIXED，那么如何正確在RDS使用大容量導入并避免鏡像異常，下面介紹幾種方式. ## 實現方法 * 通過ADO.NET SQLBulkCopy 方式 ~~~ 只需要將SqlBulkCopy 指定SqlBulkCopyOptions.CheckConstraints就好，即：SqlBulkCopy blkcpy = new SqlBulkCopy(desConnString, SqlBulkCopyOptions.CheckConstraints) 例如：將本地的一個大表通過SQLBulkCopy方式導入到RDS的實例中 static void Main() { string srcConnString = "Data Source=(local);Integrated Security=true; Initial Catalog=testdb"; string desConnString = "Data Source=****.sqlserver.rds.aliyuncs.com,3433; UserID=**;Password=**;Initial Catalog=testdb"; SqlConnection srcConnection = new SqlConnection(); SqlConnection desConnection = new SqlConnection(); SqlCommand sqlcmd = new SqlCommand(); SqlDataAdapter da = new SqlDataAdapter(); DataTable dt = new DataTable(); srcConnection.ConnectionString = srcConnString; desConnection.ConnectionString = desConnString; sqlcmd.Connection = srcConnection; sqlcmd.CommandText = @" SELECT top 1000000 [PersonType],[NameStyle],[Title],[FirstName],[MiddleName], [LastName] ,[Suffix],[EmailPromotion],[AdditionalContactInfo],[Demographics],NULL as rowguid,[ModifiedDate] FROM [testdb].[dbo].[Person]"; sqlcmd.CommandType = CommandType.Text; sqlcmd.Connection.Open(); da.SelectCommand = sqlcmd; da.Fill(dt); using (SqlBulkCopy blkcpy = new SqlBulkCopy(desConnString,SqlBulkCopyOptions.CheckConstraints)) // using (SqlBulkCopy blkcpy = // new SqlBulkCopy(desConnString, SqlBulkCopyOptions.Default)) { blkcpy.BatchSize = 2000; blkcpy.BulkCopyTimeout = 5000; blkcpy.SqlRowsCopied += new SqlRowsCopiedEventHandler(OnSqlRowsCopied); blkcpy.NotifyAfter = 2000; foreach (DataColumn dc in dt.Columns) { blkcpy.ColumnMappings.Add(dc.ColumnName, dc.ColumnName); } try { blkcpy.DestinationTableName = "Person"; blkcpy.WriteToServer(dt); } catch (Exception ex) { Console.WriteLine(ex.Message); } finally { sqlcmd.Clone(); srcConnection.Close(); desConnection.Close(); } } } private static void OnSqlRowsCopied( object sender, SqlRowsCopiedEventArgs e) { Console.WriteLine("Copied {0} so far...", e.RowsCopied); } ~~~ * 通過JDBC SQLServerBulkCopy 方式 ~~~ 同樣的道理，需要在copyOptions指定檢查約束性 SQLServerBulkCopyOptions copyOptions = new SQLServerBulkCopyOptions(); copyOptions.setCheckConstraints(true); 測試時，請用Microsoft JDBC Drivers 6.0 的sqljdbc41.jar，sqljdbc4.jar及更老版本沒有SQLServerBulkCopy 實現。例如: 將本地的一個大表通過SQLServerBulkCopy方式導入到RDS的實例中 import java.sql.*; import com.microsoft.sqlserver.jdbc.SQLServerBulkCopy; import com.microsoft.sqlserver.jdbc.SQLServerBulkCopyOptions; public class Program { public static void main(String[] args) { String sourceConnectionString = "jdbc:sqlserver://localhost:1433;" + "databaseName=testdb;user=****;password=****"; String destConnectionString = "jdbc:sqlserver://*****.sqlserver.rds.aliyuncs.com:3433;" + "databaseName=testdb;user=****;password=**** "; try { Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver"); try (Connection sourceConnection = DriverManager.getConnection(sourceConnectionString)) { try (Statement stmt = sourceConnection.createStatement()) { try (ResultSet rsSourceData = stmt.executeQuery( " SELECT top 1000000 " + "[PersonType],[NameStyle],[Title],[FirstName],[MiddleName],[LastName] ," + "[Suffix],[EmailPromotion],[AdditionalContactInfo]," + "[Demographics],NULL as rowguid,[ModifiedDate] " + "FROM [testdb].[dbo].[Person]")) { try (Connection destinationConnection = DriverManager.getConnection(destConnectionString)) { Statement stmt1 = destinationConnection.createStatement(); long countStart = 0; try (ResultSet rsRowCount = stmt1.executeQuery( "SELECT COUNT(*) FROM dbo.Person;")) { rsRowCount.next(); countStart = rsRowCount.getInt(1); System.out.println("Starting row count = " + countStart); } try (SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(destinationConnection)) { SQLServerBulkCopyOptions copyOptions = new SQLServerBulkCopyOptions(); copyOptions.setKeepIdentity(true); copyOptions.setBatchSize(2000); copyOptions.setBulkCopyTimeout(5000); //this is importance setting copyOptions.setCheckConstraints(true); bulkCopy.setBulkCopyOptions(copyOptions); bulkCopy.setDestinationTableName("dbo.Person"); bulkCopy.addColumnMapping("PersonType", "PersonType"); bulkCopy.addColumnMapping("NameStyle", "NameStyle"); bulkCopy.addColumnMapping("Title", "Title"); bulkCopy.addColumnMapping("FirstName", "FirstName"); bulkCopy.addColumnMapping("MiddleName", "MiddleName"); bulkCopy.addColumnMapping("LastName", "LastName"); bulkCopy.addColumnMapping("Suffix", "Suffix"); bulkCopy.addColumnMapping("EmailPromotion", "EmailPromotion"); bulkCopy.addColumnMapping("AdditionalContactInfo", "AdditionalContactInfo"); bulkCopy.addColumnMapping("Demographics", "Demographics"); bulkCopy.addColumnMapping("rowguid", "rowguid"); bulkCopy.addColumnMapping("ModifiedDate", "ModifiedDate"); try { bulkCopy.writeToServer(rsSourceData); } catch (Exception e) { e.printStackTrace(); } try (ResultSet rsRowCount = stmt1.executeQuery( "SELECT COUNT(*) FROM dbo.Person;")) { rsRowCount.next(); long countEnd = rsRowCount.getInt(1); System.out.println("Ending row count = " + countEnd); System.out.println((countEnd - countStart) + " rows were added."); } } } } } } } catch (Exception e) { e.printStackTrace(); } } } ~~~ * 通過BCP方式第一步：需要將數據BCP到本地 ~~~ BCP testdb.dbo.person Out "bcp_data" /t /N /U **** /P *** /S "****.sqlserver.rds.aliyuncs.com,3433" ~~~ 第二步：將導出的文件直接導入到RDS的實例中,但需要指定提示：/h “CHECK_CONSTRAINTS” ~~~ BCP testdb.dbo.person In "bcp_data" /C /N /q /k /h "CHECK_CONSTRAINTS" /U *** /P *** /b 500 /S "***.sqlserver.rds.aliyuncs.com,3433" ~~~ * 通過DTS/SSIS方式第一種：import/export data方式需要先保存SSIS包，然后修改Connection Manager的屬性，如下圖 ![](https://box.kancloud.cn/2016-07-22_5791a705c59ff.jpg) 第二種：直接使用SQL Server Business Intelligence Development Stuidio新建 SSIS包： ![](https://box.kancloud.cn/2016-07-22_5791a705e7ef4.jpg) * 特別說明不能在RDS通過下列兩種方式進行大容量插入: 原因是基于安全考慮不提供上傳文件到RDS 數據庫服務器。第一種： ~~~ BULK INSERT testdb.dbo.person_in FROM N'D:\trace\bcp.txt' WITH ( CHECK_CONSTRAINTS ); ~~~ 第二種： ~~~ INSERT ... SELECT * FROM OPENROWSET(BULK...) ~~~ * 總結大容量導入數據會帶來更快的插入，解決了用戶在有大量數據導入緩慢困惑，在阿里云數據庫中，你可以使用五種方式來實現業務場景，但是基于鏡像的主備關系，需要特別加入一個檢查約束的選項，這是寫這個最佳實踐的目的，一旦鏡像SUSPEND，不斷有DUMP文件產生，一來需要時間來修正，二來DUMP文件也會不斷占用空間，但不會影響用戶的可用性和可靠性。有兩種方式在RDS中不能實現，另外，還可以通過ODBC來實現大容量導入，具體請[參見](https://msdn.microsoft.com/en-us/library/ms403302.aspx)。希望這些對大家有用，特別是阿里云云數據庫使用用戶。