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

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
RDS PostgreSQL Serverless,0.5-4RCU 50GB 3个月
推荐场景:
对影评进行热评分析
简介: title: SQLServer · 最佳实践 · 数据库实现大容量插入的几种方式 author: 石沫 背景 很多用户在使用阿里云云数据库SQL Server时,为了加快插入速度,都尝试使用大容量插入的方式,大家都知道,对于完整恢复模式下的数据库,大容量导入执行的所有行插入操作都会完整地记录

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

author: 石沫

背景

很多用户在使用阿里云云数据库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的属性 ,如下图
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来实现大容量导入,具体请参见https://msdn.microsoft.com/en-us/library/ms403302.aspx。希望这些对大家有用,特别是阿里云云数据库使用用户。

相关实践学习
使用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
目录
相关文章
|
1月前
|
SQL 开发框架 .NET
ASP.NET连接SQL数据库:详细步骤与最佳实践指南ali01n.xinmi1009fan.com
随着Web开发技术的不断进步,ASP.NET已成为一种非常流行的Web应用程序开发框架。在ASP.NET项目中,我们经常需要与数据库进行交互,特别是SQL数据库。本文将详细介绍如何在ASP.NET项目中连接SQL数据库,并提供最佳实践指南以确保开发过程的稳定性和效率。一、准备工作在开始之前,请确保您
150 3
|
6天前
|
SQL 存储 Linux
从配置源到数据库初始化一步步教你在CentOS 7.9上安装SQL Server 2019
【11月更文挑战第8天】本文介绍了在 CentOS 7.9 上安装 SQL Server 2019 的详细步骤,包括系统准备、配置安装源、安装 SQL Server 软件包、运行安装程序、初始化数据库以及配置远程连接。通过这些步骤,您可以顺利地在 CentOS 系统上部署和使用 SQL Server 2019。
|
7天前
|
SQL 存储 Linux
从配置源到数据库初始化一步步教你在CentOS 7.9上安装SQL Server 2019
【11月更文挑战第7天】本文介绍了在 CentOS 7.9 上安装 SQL Server 2019 的详细步骤,包括系统要求检查与准备、配置安装源、安装 SQL Server 2019、配置 SQL Server 以及数据库初始化(可选)。通过这些步骤,你可以成功安装并初步配置 SQL Server 2019,进行简单的数据库操作。
|
16天前
|
存储 Java 关系型数据库
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践
在Java开发中,数据库连接是应用与数据交互的关键环节。本文通过案例分析,深入探讨Java连接池的原理与最佳实践,包括连接创建、分配、复用和释放等操作,并通过电商应用实例展示了如何选择合适的连接池库(如HikariCP)和配置参数,实现高效、稳定的数据库连接管理。
33 2
|
21天前
|
存储 数据挖掘 数据库
数据库数据恢复—SQLserver数据库ndf文件大小变为0KB的数据恢复案例
一个运行在存储上的SQLServer数据库,有1000多个文件,大小几十TB。数据库每10天生成一个NDF文件,每个NDF几百GB大小。数据库包含两个LDF文件。 存储损坏,数据库不可用。管理员试图恢复数据库,发现有数个ndf文件大小变为0KB。 虽然NDF文件大小变为0KB,但是NDF文件在磁盘上还可能存在。可以尝试通过扫描&拼接数据库碎片来恢复NDF文件,然后修复数据库。
|
1月前
|
关系型数据库 MySQL 数据库
MySQL数据库:基础概念、应用与最佳实践
一、引言随着互联网技术的快速发展,数据库管理系统在现代信息系统中扮演着核心角色。在众多数据库管理系统中,MySQL以其开源、稳定、可靠以及跨平台的特性受到了广泛的关注和应用。本文将详细介绍MySQL数据库的基本概念、特性、应用领域以及最佳实践,帮助读者更好地理解和应用MySQL数据库。二、MySQL
109 5
|
1月前
|
SQL 数据管理 数据库
文章初学者指南:SQL新建数据库详细步骤与最佳实践
引言:在当今数字化的世界,数据库管理已经成为信息技术领域中不可或缺的一部分。作为广泛使用的数据库管理系统,SQL已经成为数据管理和信息检索的标准语言。本文将详细介绍如何使用SQL新建数据库,包括准备工作、具体步骤和最佳实践,帮助初学者快速上手。一、准备工作在开始新建数据库之前,你需要做好以下准备工作
117 3
|
2月前
|
SQL 关系型数据库 MySQL
创建包含MySQL和SQLServer数据库所有字段类型的表的方法
创建一个既包含MySQL又包含SQL Server所有字段类型的表是一个复杂的任务,需要仔细地比较和转换数据类型。通过上述方法,可以在两个数据库系统之间建立起相互兼容的数据结构,为数据迁移和同步提供便利。这一过程不仅要考虑数据类型的直接对应,还要注意特定数据类型在不同系统中的表现差异,确保数据的一致性和完整性。
32 4
|
2月前
|
SQL 数据库
数据库数据恢复—SQL Server数据库报错“错误823”的数据恢复案例
SQL Server附加数据库出现错误823,附加数据库失败。数据库没有备份,无法通过备份恢复数据库。 SQL Server数据库出现823错误的可能原因有:数据库物理页面损坏、数据库物理页面校验值损坏导致无法识别该页面、断电或者文件系统问题导致页面丢失。
101 12
数据库数据恢复—SQL Server数据库报错“错误823”的数据恢复案例
|
2月前
|
SQL 存储 数据管理
SQL Server数据库
SQL Server数据库
58 11

相关产品

  • 云数据库 RDS SQL Server 版