SQLServer · 最佳实践 · 数据库实现大容量插入的几种方式

本文涉及的产品
云原生数据库 PolarDB 分布式版,标准版 2核8GB
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
简介:

背景

很多用户在使用阿里云云数据库SQL Server时,为了加快插入速度,都尝试使用大容量插入的方式,大家都知道,对于完整恢复模式下的数据库,大容量导入执行的所有行插入操作都会完整地记录在事务日志中。如果使用完整恢复模式,大型数据导入会导致填充事务日志的速度很快。相反,对于简单恢复模式或大容量日志恢复模式,大容量导入操作的按最小方式记录日志减少了大容量导入操作填满日志空间的可能性。另外,按最小方式记录日志的效率也比按完整方式记录日志高 。
但实际上,当大容量导入与数据库镜像共存时,会出现镜像 Suspend的情况,这个情况是由于微软在2008 R2上的BUG导致,微软已经明确表示在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的属性 ,如下图
      1111111

      第二种:直接使用SQL Server Business Intelligence Development Stuidio新建 SSIS包:

      222222

      • 特别说明

      不能在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来实现大容量导入,具体请参见。希望这些对大家有用,特别是阿里云云数据库使用用户。


相关实践学习
使用SQL语句管理索引
本次实验主要介绍如何在RDS-SQLServer数据库中,使用SQL语句管理索引。
SQL Server on Linux入门教程
SQL Server数据库一直只提供Windows下的版本。2016年微软宣布推出可运行在Linux系统下的SQL Server数据库,该版本目前还是早期预览版本。本课程主要介绍SQLServer On Linux的基本知识。 相关的阿里云产品:云数据库RDS SQL Server版 RDS SQL Server不仅拥有高可用架构和任意时间点的数据恢复功能,强力支撑各种企业应用,同时也包含了微软的License费用,减少额外支出。 了解产品详情: https://www.aliyun.com/product/rds/sqlserver
目录
相关文章
|
4天前
|
数据库 Windows
SqlServer数据恢复—SqlServer数据库所在分区损坏的数据恢复案例
一块硬盘上存放的SqlServer数据库,windows server操作系统+NTFS文件系统。由于误操作导致分区损坏,需要恢复硬盘里的SqlServer数据库数据。
|
2月前
|
SQL 存储 Linux
从配置源到数据库初始化一步步教你在CentOS 7.9上安装SQL Server 2019
【11月更文挑战第16天】本文介绍了在 CentOS 7.9 上安装 SQL Server 2019 的详细步骤,包括配置系统源、安装 SQL Server 2019 软件包以及数据库初始化,确保 SQL Server 正常运行。
|
2月前
|
SQL 存储 Linux
从配置源到数据库初始化一步步教你在CentOS 7.9上安装SQL Server 2019
【11月更文挑战第8天】本文介绍了在 CentOS 7.9 上安装 SQL Server 2019 的详细步骤,包括系统准备、配置安装源、安装 SQL Server 软件包、运行安装程序、初始化数据库以及配置远程连接。通过这些步骤,您可以顺利地在 CentOS 系统上部署和使用 SQL Server 2019。
125 1
|
2月前
|
SQL 存储 Linux
从配置源到数据库初始化一步步教你在CentOS 7.9上安装SQL Server 2019
【11月更文挑战第7天】本文介绍了在 CentOS 7.9 上安装 SQL Server 2019 的详细步骤,包括系统要求检查与准备、配置安装源、安装 SQL Server 2019、配置 SQL Server 以及数据库初始化(可选)。通过这些步骤,你可以成功安装并初步配置 SQL Server 2019,进行简单的数据库操作。
|
2月前
|
存储 Java 关系型数据库
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践,包括连接创建、分配、复用和释放等操作,并通过电商应用实例展示了如何选择合适的连接池库(如HikariCP)和配置参数,实现高效、稳定的数据库连接管理。
73 2
|
3月前
|
存储 数据挖掘 数据库
数据库数据恢复—SQLserver数据库ndf文件大小变为0KB的数据恢复案例
一个运行在存储上的SQLServer数据库,有1000多个文件,大小几十TB。数据库每10天生成一个NDF文件,每个NDF几百GB大小。数据库包含两个LDF文件。 存储损坏,数据库不可用。管理员试图恢复数据库,发现有数个ndf文件大小变为0KB。 虽然NDF文件大小变为0KB,但是NDF文件在磁盘上还可能存在。可以尝试通过扫描&拼接数据库碎片来恢复NDF文件,然后修复数据库。
|
21天前
|
存储 Oracle 关系型数据库
数据库传奇:MySQL创世之父的两千金My、Maria
《数据库传奇:MySQL创世之父的两千金My、Maria》介绍了MySQL的发展历程及其分支MariaDB。MySQL由Michael Widenius等人于1994年创建,现归Oracle所有,广泛应用于阿里巴巴、腾讯等企业。2009年,Widenius因担心Oracle收购影响MySQL的开源性,创建了MariaDB,提供额外功能和改进。维基百科、Google等已逐步替换为MariaDB,以确保更好的性能和社区支持。掌握MariaDB作为备用方案,对未来发展至关重要。
47 3
|
21天前
|
安全 关系型数据库 MySQL
MySQL崩溃保险箱:探秘Redo/Undo日志确保数据库安全无忧!
《MySQL崩溃保险箱:探秘Redo/Undo日志确保数据库安全无忧!》介绍了MySQL中的三种关键日志:二进制日志(Binary Log)、重做日志(Redo Log)和撤销日志(Undo Log)。这些日志确保了数据库的ACID特性,即原子性、一致性、隔离性和持久性。Redo Log记录数据页的物理修改,保证事务持久性;Undo Log记录事务的逆操作,支持回滚和多版本并发控制(MVCC)。文章还详细对比了InnoDB和MyISAM存储引擎在事务支持、锁定机制、并发性等方面的差异,强调了InnoDB在高并发和事务处理中的优势。通过这些机制,MySQL能够在事务执行、崩溃和恢复过程中保持
54 3
|
21天前
|
SQL 关系型数据库 MySQL
数据库灾难应对:MySQL误删除数据的救赎之道,技巧get起来!之binlog
《数据库灾难应对:MySQL误删除数据的救赎之道,技巧get起来!之binlog》介绍了如何利用MySQL的二进制日志(Binlog)恢复误删除的数据。主要内容包括: 1. **启用二进制日志**:在`my.cnf`中配置`log-bin`并重启MySQL服务。 2. **查看二进制日志文件**:使用`SHOW VARIABLES LIKE 'log_%';`和`SHOW MASTER STATUS;`命令获取当前日志文件及位置。 3. **创建数据备份**:确保在恢复前已有备份,以防意外。 4. **导出二进制日志为SQL语句**:使用`mysqlbinlog`
72 2
|
1月前
|
关系型数据库 MySQL 数据库
Python处理数据库:MySQL与SQLite详解 | python小知识
本文详细介绍了如何使用Python操作MySQL和SQLite数据库,包括安装必要的库、连接数据库、执行增删改查等基本操作,适合初学者快速上手。
227 15

相关产品

  • 云数据库 RDS SQL Server 版