将 200 万行快速插入 SQL Server

Posted 2023-04-12

【中文标题】将 200 万行快速插入 SQL Server

【英文标题】：Insert 2 million rows into SQL Server quickly

【发布时间】：2012-11-23 04:57:59

【问题描述】：

我必须从一个文本文件中插入大约 200 万行。

通过插入，我必须创建一些主表。

将如此庞大的数据集插入 SQL Server 的最佳且快速的方法是什么？

【参考方案1】：

我认为最好在DataSet中读取文本文件的数据

试用 SqlBulkCopy - Bulk Insert into SQL from C# App

// connect to SQL
using (SqlConnection connection = new SqlConnection(connString))

    // make sure to enable triggers
    // more on triggers in next post
    SqlBulkCopy bulkCopy = new SqlBulkCopy(
        connection, 
        SqlBulkCopyOptions.TableLock | 
        SqlBulkCopyOptions.FireTriggers | 
        SqlBulkCopyOptions.UseInternalTransaction,
        null
        );

    // set the destination table name
    bulkCopy.DestinationTableName = this.tableName;
    connection.Open();

    // write the data in the "dataTable"
    bulkCopy.WriteToServer(dataTable);
    connection.Close();

// reset
this.dataTable.Clear();

或

在顶部执行第 1 步之后

从数据集创建 XML 将 XML 传递到数据库并进行批量插入

您可以查看这篇文章了解详细信息：Bulk Insertion of Data Using C# DataTable and SQL server OpenXML function

但它没有用 200 万条记录测试，它会消耗机器上的内存，因为你必须加载 200 万条记录并插入它。

【讨论】：

我知道这已经很晚了，但是对于大约 200 万行（或更多），如果有足够的列（25+），几乎不可避免的是代码会在某些时候生成OutOfMemoryException点，填充数据集/数据表时。

您可以设置缓冲区以避免内存不足异常。对于文本文件，我使用了 File.ReadLines(file).Skip(X).Take(100000).ToList()。在每 100k 之后，我重置并通过下一个 100k。效果很好，速度很快。

如果源数据来自 Sql Server 表怎么办。假设该表有 3000 万行，我们还可以使用批量复制吗？一个简单的Insert into table1 Select * from table2 不是更快吗？

【参考方案2】：

你可以试试SqlBulkCopy类。

让您可以高效地批量加载 SQL Server 表，其中包含来自 另一个来源。

有一个很酷的blog post 告诉你如何使用它。

【讨论】：

另请注意（在SQLBulCopy 链接中也提到）If the source and destination tables are in the same SQL Server instance, it is easier and faster to use a Transact-SQL INSERT … SELECT statement to copy the data.

【参考方案3】：

重新解决SqlBulkCopy：

我使用 StreamReader 来转换和处理文本文件。结果是我的对象列表。

我创建了一个类而不是 Datatable 或 List<T> 和缓冲区大小 (CommitBatchSize)。它将使用扩展名（在第二类中）将列表转换为数据表。

它的工作速度非常快。在我的 PC 上，我可以在 10 秒内插入超过 1000 万条复杂记录。

这是课程：

using System;
using System.Collections;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DAL


public class BulkUploadToSql<T>

    public IList<T> InternalStore  get; set; 
    public string TableName  get; set; 
    public int CommitBatchSize  get; set; =1000;
    public string ConnectionString  get; set; 

    public void Commit()
    
        if (InternalStore.Count>0)
        
            DataTable dt;
            int numberOfPages = (InternalStore.Count / CommitBatchSize)  + (InternalStore.Count % CommitBatchSize == 0 ? 0 : 1);
            for (int pageIndex = 0; pageIndex < numberOfPages; pageIndex++)
                
                    dt= InternalStore.Skip(pageIndex * CommitBatchSize).Take(CommitBatchSize).ToDataTable();
                BulkInsert(dt);
                
         
    

    public void BulkInsert(DataTable dt)
    
        using (SqlConnection connection = new SqlConnection(ConnectionString))
        
            // make sure to enable triggers
            // more on triggers in next post
            SqlBulkCopy bulkCopy =
                new SqlBulkCopy
                (
                connection,
                SqlBulkCopyOptions.TableLock |
                SqlBulkCopyOptions.FireTriggers |
                SqlBulkCopyOptions.UseInternalTransaction,
                null
                );

            // set the destination table name
            bulkCopy.DestinationTableName = TableName;
            connection.Open();

            // write the data in the "dataTable"
            bulkCopy.WriteToServer(dt);
            connection.Close();
        
        // reset
        //this.dataTable.Clear();
    



public static class BulkUploadToSqlHelper

    public static DataTable ToDataTable<T>(this IEnumerable<T> data)
    
        PropertyDescriptorCollection properties =
            TypeDescriptor.GetProperties(typeof(T));
        DataTable table = new DataTable();
        foreach (PropertyDescriptor prop in properties)
            table.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
        foreach (T item in data)
        
            DataRow row = table.NewRow();
            foreach (PropertyDescriptor prop in properties)
                row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
            table.Rows.Add(row);
        
        return table;
    

这是一个示例，当我想插入我的自定义对象 List<PuckDetection> (ListDetections) 的列表时：

var objBulk = new BulkUploadToSql<PuckDetection>()

        InternalStore = ListDetections,
        TableName= "PuckDetections",
        CommitBatchSize=1000,
        ConnectionString="ENTER YOU CONNECTION STRING"
;
objBulk.Commit();

如果需要，可以修改BulkInsert 类以添加列映射。例如，您有一个 Identity 键作为第一列。（假设数据表中的列名与数据库相同）

//ADD COLUMN MAPPING
foreach (DataColumn col in dt.Columns)

        bulkCopy.ColumnMappings.Add(col.ColumnName, col.ColumnName);

【参考方案4】：

我使用 bcp 实用程序。 （批量复制程序） 我每月加载大约 150 万条文本记录。 每个文本记录的宽度为 800 个字符。 在我的服务器上，将 150 万条文本记录添加到 SQL Server 表中大约需要 30 秒。

bcp 的说明在http://msdn.microsoft.com/en-us/library/ms162802.aspx

【讨论】：

我也会推荐 bcp 实用程序。我什么都不知道。

【参考方案5】：

我最近遇到了这种情况（超过 700 万行）并通过 powershell 使用 sqlcmd（在将原始数据解析为 SQL 插入语句之后）一次 5,000 段（SQL 无法处理 700 万行）一个批量作业甚至 500,000 行，除非它分解成更小的 5K 片段。然后您可以一个接一个地运行每个 5K 脚本。）因为我需要利用 SQL Server 2012 Enterprise 中的新序列命令。我找不到使用上述序列命令快速有效地插入 700 万行数据的编程方式。

其次，一次插入一百万行或更多数据时要注意的一件事是插入过程中的 CPU 和内存消耗（主要是内存）。 SQL 将在不释放上述进程的情况下完成如此规模的工作占用内存/CPU。不用说，如果您的服务器上没有足够的处理能力或内存，您很容易在短时间内崩溃（我发现很难）。如果你的内存消耗超过 70-75%，只需重新启动服务器，进程就会恢复正常。

在我真正制定最终执行计划之前，我必须运行大量试错测试来查看我的服务器的限制是什么（考虑到可使用的 CPU/内存资源有限）。我建议您在将其推广到生产环境之前在测试环境中执行相同的操作。

【讨论】：

7M 行花了多长时间？我有大约 30M 行要插入。现在我正在通过存储过程和 DataTable 推动它们。

同时运行小批量需要 5 到 6 个小时。请记住，我只是直接执行了 T-SQL 插入命令，因为我利用了 SQL 2012 中的新 SEQUENCE 命令，并且找不到有关如何在 T-SQL 之外自动化此过程的信息。

【参考方案6】：

我尝试了这种方法，它显着减少了我的数据库插入执行时间。

List<string> toinsert = new List<string>();
StringBuilder insertCmd = new StringBuilder("INSERT INTO tabblename (col1, col2, col3) VALUES ");

foreach (var row in rows)

      // the point here is to keep values quoted and avoid SQL injection
      var first = row.First.Replace("'", "''")
      var second = row.Second.Replace("'", "''")
      var third = row.Third.Replace("'", "''")

      toinsert.Add(string.Format("( '0', '1', '2' )", first, second, third));

if (toinsert.Count != 0)

      insertCmd.Append(string.Join(",", toinsert));
      insertCmd.Append(";");

using (mysqlCommand myCmd = new MySqlCommand(insertCmd.ToString(), SQLconnectionObject))

      myCmd.CommandType = CommandType.Text;
      myCmd.ExecuteNonQuery();

*创建 SQL 连接对象并将其替换为我编写 SQLconnectionObject 的位置。

【讨论】：

小心！这可以通过 SQL 注入来利用

反对 1) 不能避免 sql 注入攻击。 （见***.com/questions/15537368/…）

【参考方案7】：

我遇到了一个应该与 ADO、Entity 和 Dapper 一起使用的解决方案的问题，所以我做了一个 this lib;它以以下形式生成批次：

    IEnumerable<(string SqlQuery, IEnumerable<SqlParameter> SqlParameters)>  
    IEnumerable<(string SqlQuery, DynamicParameters DapperDynamicParameters)> 

this link 包含说明。它对 SQL 注入是安全的，因为使用参数而不是连接；如果需要，您也可以通过可选参数将标识插入设置为 ON。

与 ADO.NET 一起使用：

using MsSqlHelpers;
// ...
var mapper = new MapperBuilder<Person>()
    .SetTableName("People")
    .AddMapping(person => person.FirstName, columnName: "Name")
    .AddMapping(person => person.LastName, columnName: "Surename")
    .AddMapping(person => person.DateOfBirth, columnName: "Birthday")
    .Build();
var people = new List<Person>()
 
    new Person()
    
        FirstName = "John", 
        LastName = "Lennon", 
        DateOfBirth = new DateTime(1940, 10, 9) 
    ,
    new Person()
    
        FirstName = "Paul", 
        LastName = "McCartney", 
        DateOfBirth = new DateTime(1942, 6, 18) 
    ,
;
var connectionString = "Server=SERVER_ADDRESS;Database=DATABASE_NAME;User Id=USERNAME;Password=PASSWORD;";
var sqlQueriesAndParameters = new MsSqlQueryGenerator()
    .GenerateParametrizedBulkInserts(mapper, people);

using (var sqlConnection = new SqlConnection(connectionString))

    sqlConnection.Open();
    
    // Default batch size: 1000 rows or (2100-1) parameters per insert.
    foreach (var (SqlQuery, SqlParameters) in sqlQueriesAndParameters)
    
        using (SqlCommand sqlCommand = new SqlCommand(SqlQuery, sqlConnection))
        
            sqlCommand.Parameters.AddRange(SqlParameters.ToArray());
            sqlCommand.ExecuteNonQuery();
        
    

与 Dapper 一起使用：

using MsSqlHelpers;
// ...
var mapper = new MapperBuilder<Person>()
    .SetTableName("People")
    .AddMapping(person => person.FirstName, columnName: "Name")
    .AddMapping(person => person.LastName, columnName: "Surename")
    .AddMapping(person => person.DateOfBirth, columnName: "Birthday")
    .Build();
var people = new List<Person>()
 
    new Person()
    
        FirstName = "John", 
        LastName = "Lennon", 
        DateOfBirth = new DateTime(1940, 10, 9) 
    ,
    new Person()
     
        FirstName = "Paul", 
        LastName = "McCartney", 
        DateOfBirth = new DateTime(1942, 6, 18) 
    ,
;
var connectionString = "Server=SERVER_ADDRESS;Database=DATABASE_NAME;User Id=USERNAME;Password=PASSWORD;";
var sqlQueriesAndDapperParameters = new MsSqlQueryGenerator()
    .GenerateDapperParametrizedBulkInserts(mapper, people);

using (var sqlConnection = new SqlConnection(connectionString))

    // Default batch size: 1000 rows or (2100-1) parameters per insert.
    foreach (var (SqlQuery, DapperDynamicParameters) in sqlQueriesAndDapperParameters)
    
        sqlConnection.Execute(SqlQuery, DapperDynamicParameters);
    

与实体框架一起使用：

using MsSqlHelpers;
// ...
var mapper = new MapperBuilder<Person>()
    .SetTableName("People")
    .AddMapping(person => person.FirstName, columnName: "Name")
    .AddMapping(person => person.LastName, columnName: "Surename")
    .AddMapping(person => person.DateOfBirth, columnName: "Birthday")
    .Build();
var people = new List<Person>()
 
    new Person() 
     
        FirstName = "John", 
        LastName = "Lennon", 
        DateOfBirth = new DateTime(1940, 10, 9) 
    ,
    new Person()
     
        FirstName = "Paul", 
        LastName = "McCartney", 
        DateOfBirth = new DateTime(1942, 6, 18) 
    ,
;
var sqlQueriesAndParameters = new MsSqlQueryGenerator()
    .GenerateParametrizedBulkInserts(mapper, people);

// Default batch size: 1000 rows or (2100-1) parameters per insert.
foreach (var (SqlQuery, SqlParameters) in sqlQueriesAndParameters)

    _context.Database.ExecuteSqlRaw(SqlQuery, SqlParameters);
    // Depracated but still works: _context.Database.ExecuteSqlCommand(SqlQuery, SqlParameters);