SqlBulkCopy 批量复制数据到数据表

使用 SqlBulkCopy 类只能向 SQL Server 表写入数据。但是，数据源不限于 SQL Server；可以使用任何数据源，只要数据可加载到 DataTable 实例或可使用 IDataReader 实例读取数据

1. 使用Datatable作为数据源的方式：

下面的代码使用到了ColumnMappings，因为目标表和数据源Datatable的结构不一致，需要这么一个映射来指定对应关系

  public string SaveJHCData(LzShopBasicData[] datas){var result = new AResult();SqlConnection con = new SqlConnection(System.Configuration.ConfigurationManager.ConnectionStrings["**"].ConnectionString);con.Open();foreach (var item in datas){Logger.Info("数据更新处理，店铺名称：" + item.ShopName + "数据日期" + item.SellDate);try{using (TransactionScope scope = new TransactionScope()){DataTable JHCOrderItemsdt = SaveJHCOrderItemsData(item);SqlBulkCopy JHCOrderItemscopy = new SqlBulkCopy(con);JHCOrderItemscopy.ColumnMappings.Add("orderId", "orderId");JHCOrderItemscopy.ColumnMappings.Add("auctionId", "auctionId");JHCOrderItemscopy.ColumnMappings.Add("itemTitle", "itemTitle");JHCOrderItemscopy.ColumnMappings.Add("tradeAmt", "tradeAmt");JHCOrderItemscopy.ColumnMappings.Add("alipayNum", "alipayNum");JHCOrderItemscopy.ColumnMappings.Add("tradeTime", "tradeTime");JHCOrderItemscopy.ColumnMappings.Add("uv", "uv");JHCOrderItemscopy.ColumnMappings.Add("srcId", "srcId");JHCOrderItemscopy.ColumnMappings.Add("srcName", "srcName");JHCOrderItemscopy.ColumnMappings.Add("DataType", "DataType");JHCOrderItemscopy.ColumnMappings.Add("DataDate", "DataDate");JHCOrderItemscopy.ColumnMappings.Add("OrderSourceID", "OrderSourceID");JHCOrderItemscopy.ColumnMappings.Add("ShopName", "ShopName");JHCOrderItemscopy.DestinationTableName = "JHCOrderItems";JHCOrderItemscopy.WriteToServer(JHCOrderItemsdt);result.Updatedata += 1;result.UpdatedataText += item.SellDate + ",";scope.Complete();Logger.Info(item.SellDate + "事务提交");}}catch (Exception ex){Logger.Error(ex.ToString());continue;}}con.Close();return result.ToSerializeObject();}

 2.使用IDataReader作为数据源的方式，这种方式个人认为用的很少，首先目标表和来源表两个数据库连接你都需要拿到，如果两个都可以拿到，一般直接操作sql就可以解决：

这里是直接拷贝的MSDN的代码，

用到的AdventureWorks 数据库可以直接在网上下载到，

using System.Data.SqlClient;class Program
{static void Main(){string connectionString = GetConnectionString();// Open a sourceConnection to the AdventureWorks database.using (SqlConnection sourceConnection =new SqlConnection(connectionString)){sourceConnection.Open();// Perform an initial count on the destination table.SqlCommand commandRowCount = new SqlCommand("SELECT COUNT(*) FROM " +"dbo.BulkCopyDemoMatchingColumns;",sourceConnection);long countStart = System.Convert.ToInt32(commandRowCount.ExecuteScalar());Console.WriteLine("Starting row count = {0}", countStart);// Get data from the source table as a SqlDataReader.SqlCommand commandSourceData = new SqlCommand("SELECT ProductID, Name, " +"ProductNumber " +"FROM Production.Product;", sourceConnection);SqlDataReader reader =commandSourceData.ExecuteReader();// Open the destination connection. In the real world you would // not use SqlBulkCopy to move data from one table to the other // in the same database. This is for demonstration purposes only.using (SqlConnection destinationConnection =new SqlConnection(connectionString)){destinationConnection.Open();// Set up the bulk copy object. // Note that the column positions in the source// data reader match the column positions in // the destination table so there is no need to// map columns.using (SqlBulkCopy bulkCopy =new SqlBulkCopy(destinationConnection)){bulkCopy.DestinationTableName ="dbo.BulkCopyDemoMatchingColumns";try{// Write from the source to the destination.
                        bulkCopy.WriteToServer(reader);}catch (Exception ex){Console.WriteLine(ex.Message);}finally{// Close the SqlDataReader. The SqlBulkCopy// object is automatically closed at the end// of the using block.
                        reader.Close();}}// Perform a final count on the destination // table to see how many rows were added.long countEnd = System.Convert.ToInt32(commandRowCount.ExecuteScalar());Console.WriteLine("Ending row count = {0}", countEnd);Console.WriteLine("{0} rows were added.", countEnd - countStart);Console.WriteLine("Press Enter to finish.");Console.ReadLine();}}}private static string GetConnectionString()// To avoid storing the sourceConnection string in your code, // you can retrieve it from a configuration file. 
    {return "Data Source=(local); " +" Integrated Security=true;" +"Initial Catalog=AdventureWorks;";}
}

 

实战：借助类型反射动态构建Datatable数据源,通过SqlBulkCopy批量保存入库

1.获取一张空的Datatable

var dt = bisdal.From<TopBrand>(TopBrand._.ID == -1, OrderByClip.Default).ToDataTable();

2.填充DataTable，这里是通过遍历外部的集合，把属性属性逐一赋值填充到目标Datatable

         foreach (var item in brandselldataitems){try{TopBrand topbrand = new TopBrand{BrandIndex = item.mk,BrandName = item.c58,Date = date,WinnerAmt = item.c60,WinnerPeople = item.c62,WinnerProNum = item.c61,HotTaobaoCategoryID = cid};CreateDtByItem<TopBrand>(topbrand, dt);}catch (Exception ex){Logger.Error(ex.ToString());continue;}}

 这里借助反射，遍历实体属性集合，动态构建DataTableRow对象

       private void CreateDtByItem<T>(T item, DataTable dt){System.Reflection.PropertyInfo[] properties = item.GetType().GetProperties(System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.Public);var newrow = dt.NewRow();foreach (System.Reflection.PropertyInfo pitem in properties){string name = pitem.Name;if (name == "children"){continue;}object value = pitem.GetValue(item, null);newrow[name] = value == null ? DBNull.Value : value;}dt.Rows.Add(newrow);}

3.保存入库，

     BulkWriteToServer(con, "TopBrand", dt);

 这里因为目标表和数据源的Datatable数据结构一致，所以省去了ColumnMappings列映射的操作，可以直接WriteToServer保存

     private void BulkWriteToServer(SqlConnection con, string destinationtablename, DataTable sourcedt){try{if (con.State == ConnectionState.Closed){con.Open();}SqlBulkCopy topbranddtcopy = new SqlBulkCopy(con);topbranddtcopy.DestinationTableName = destinationtablename;topbranddtcopy.WriteToServer(sourcedt);con.Close();}catch (Exception ex){Logger.Error("批量新增数据:" + destinationtablename + "," + ex.ToString());}}

 

完整调用代码：

 private void CreateTopBrandData(int date, int cid, List<BrandSellDataItem> brandselldataitems){try{var dt = bisdal.From<TopBrand>(TopBrand._.ID == -1, OrderByClip.Default).ToDataTable();foreach (var item in brandselldataitems){try{TopBrand topbrand = new TopBrand{BrandIndex = item.mk,BrandName = item.c58,Date = date,WinnerAmt = item.c60,WinnerPeople = item.c62,WinnerProNum = item.c61,HotTaobaoCategoryID = cid};CreateDtByItem<TopBrand>(topbrand, dt);}catch (Exception ex){Logger.Error(ex.ToString());continue;}}BulkWriteToServer(con, "TopBrand", dt);}catch (Exception ex){throw new Exception("CreateTopBrandData:" + ex.ToString());}}