如何将一个巨大的表数据复制到SQL Server中的另一个表中

Question

我有一个包含 340 万行的表。我想将整个数据复制到另一个表中。

我正在使用以下查询执行此任务：

select * 
into new_items 
from productDB.dbo.items

我需要知道执行此任务的最佳方法。

Answer 1

我遇到了同样的问题，除了我有一个包含 20 亿行的表，因此如果我这样做，即使将恢复模式设置为批量日志记录，日志文件也会无限增长：

insert into newtable select * from oldtable

所以我对数据块进行操作。这样，如果传输中断，您只需重新启动即可。此外，您不需要与表一样大的日志文件。您似乎也获得了更少的 tempdb I/O，不知道为什么。

set identity_insert newtable on
DECLARE @StartID bigint, @LastID bigint, @EndID bigint
select @StartID = isNull(max(id),0) + 1
from newtable

select @LastID = max(ID)
from oldtable

while @StartID <= @LastID
begin
    set @EndID = @StartID + 1000000

    insert into newtable (FIELDS,GO,HERE)
    select FIELDS,GO,HERE from oldtable (NOLOCK)
    where id BETWEEN @StartID AND @EndId
            
    set @StartID = @EndID + 1
end
set identity_insert newtable off
go

您可能需要更改处理 ID 的方式，如果您的表按 ID 进行集群，则此方法效果最佳。

Answer 2

如果您要复制到新表中，最快的方法可能就是您问题中的方法，除非您的行非常大。

如果您的行非常大，您可能需要使用 SQL Server 中的批量插入功能。我认为你可以从 C# 调用它们。

或者，您可以先将该数据下载到文本文件中，然后批量复制 (bcp)。这样做的额外好处是允许您忽略键、索引等。

还可以尝试 SQL Management Studio 附带的导入/导出实用程序；不确定它是否会像直接批量复制一样快，但它应该允许您跳过写为平面文件的中间步骤，而直接复制表到表，这可能比您的 SELECT INTO 语句。

Answer 3

我一直在与 DBA 合作将一个包含 240M 行的审计表复制到另一个数据库。

使用简单的选择/插入创建了一个巨大的 tempdb 文件。

使用导入/导出向导有效，但在 10 分钟内复制了 800 万行

创建自定义 SSIS 包并调整设置在 10 分钟内复制了 3000 万行

来说是最快、最有效的

SSIS 包对于我们的目的

Answer 4

这是转移大表的另一种方法。我刚刚使用它在两台服务器之间传输了 1.05 亿行。也相当快。

1. 右键单击数据库并选择“任务/导出数据”。
2. 向导将引导您完成这些步骤，但您选择 SQL Server 客户端作为数据源和目标将允许您选择要传输的数据库和表。

有关详细信息，请参阅 https:// /www.mssqltips.com/sqlservertutorial/202/simple-way-to-export-data-from-sql-server/

Answer 5

如果是 1 次导入，SSMS 中的导入/导出实用程序可能会最简单、最快。 SSIS 似乎也比直接 INSERT 更适合导入大型数据集。

BULK INSERT 或 BCP 也可用于导入大型记录集。

另一种选择是暂时删除要导入的表上的所有索引和约束，并在导入过程完成后将它们添加回来。以前失败的直接 INSERT 在这些情况下可能会起作用。

如果您在直接从一个数据库转到另一个数据库时遇到超时或锁定/阻塞问题，您可能会考虑从一个数据库进入 TEMPDB，然后从 TEMPDB 进入另一个数据库，因为这样可以最大限度地减少锁定和阻塞进程对数据库的影响。任一侧。 TempDB 不会阻止或锁定源，也不会阻止目标。

这些是一些可以尝试的选择。

——埃里克·艾萨克斯

Answer 6

我喜欢@Mathieu Longtin 的解决方案，批量复制，从而最大限度地减少日志文件问题，并按照@CervEd 的建议创建了带有 OFFSET FETCH 的版本。

其他人建议使用导入/导出向导或 SSIS 包，但这并不总是可行。

对于许多人来说，这可能有点过分了，但我的解决方案还包括对记录计数和输出进度的一些检查。

USE [MyDB]
GO

SET NOCOUNT ON;
DECLARE @intStart int = 1;
DECLARE @intCount int;
DECLARE @intFetch int = 10000;
DECLARE @strStatus VARCHAR(200);
DECLARE @intCopied int = 0;

SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Getting count of HISTORY records currently in MyTable...';
RAISERROR (@strStatus, 10, 1) WITH NOWAIT;
SELECT @intCount = COUNT(*) FROM [dbo].MyTable WHERE IsHistory = 1;
SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Count of HISTORY records currently in MyTable: ' + CONVERT(VARCHAR(20), @intCount);
RAISERROR (@strStatus, 10, 1) WITH NOWAIT;  --(note: PRINT resets @@ROWCOUNT to 0 so using RAISERROR instead)
SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Starting copy...';
RAISERROR (@strStatus, 10, 1) WITH NOWAIT;

WHILE @intStart < @intCount
BEGIN

    INSERT INTO [dbo].[MyTable_History] (
        [PK1], [PK2], [PK3], [Data1], [Data2])
    SELECT
        [PK1], [PK2], [PK3], [Data1], [Data2]
    FROM [MyDB].[dbo].[MyTable]
    WHERE IsHistory = 1
    ORDER BY 
        [PK1], [PK2], [PK3]
        OFFSET @intStart - 1 ROWS 
        FETCH NEXT @intFetch ROWS ONLY;

    SET @intCopied = @intCopied + @@ROWCOUNT;
    SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Records copied so far: ' + CONVERT(VARCHAR(20), @intCopied); 
    RAISERROR (@strStatus, 10, 1) WITH NOWAIT;

    SET @intStart = @intStart + @intFetch;

END

--Check the record count is correct.
IF @intCopied = @intCount
    BEGIN
        SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Correct record count.'; 
        RAISERROR (@strStatus, 10, 1) WITH NOWAIT;
    END
ELSE
    BEGIN
        SET @strStatus = CONVERT(VARCHAR(30), GETDATE()) + ' Only ' + CONVERT(VARCHAR(20), @intCopied) + ' records were copied, expected: ' + CONVERT(VARCHAR(20), @intCount);
        RAISERROR (@strStatus, 10, 1) WITH NOWAIT;
    END


GO

Answer 7

简单插入/选择 sp 的工作效果很好，直到行数超过 1 百万。我见过 tempdb 文件在尝试插入/选择 2000 万行以上时发生爆炸。最简单的解决方案是 SSIS 将批处理行大小缓冲区设置为 5000，并将提交大小缓冲区设置为 1000。

Answer 8

我知道这已经晚了，但是如果您遇到信号量超时，那么您可以使用 row_number 来设置插入的增量，使用类似

INSERT INTO DestinationTable (column1, column2, etc) 
 FROM ( 
SELECT ROW_NUMBER() OVER (ORDER BY ID) AS RN , column1, column2, etc
FROM SourceTable ) AS A
WHERE A.RN >= 1 AND A.RN <= 10000 )

日志文件的大小将会增长的内容，因此需要解决这个问题。如果在插入现有表时禁用约束和索引，您将获得更好的性能。然后启用约束并在插入完成后为插入的表重建索引。

Answer 9

如果您的重点是归档 (DW) 并且正在处理具有 100 多个分区表的 VLDB，并且您希望将大部分资源密集型工作隔离在非生产服务器 (OLTP) 上，这里有一个建议 (OLTP -> DW)
1）使用备份/恢复将数据保存到存档服务器上（所以现在，在存档或DW上您将拥有阶段和目标数据库）
2）Stage数据库：使用分区开关将数据移动到对应的Stage表
3）使用SSIS将双方每个暂存表的数据从暂存数据库传输到目标数据库
4）目标数据库：使用目标数据库上的分区开关将数据从阶段移动到基表
希望这有帮助。

Answer 10

我正在回答这篇 12 年前的帖子，因为它仍然具有相关性，并且参考我的博客文章可能会有用。这就是将大表/数据分成“块”并一次加载一个块或并行加载这些块。如果出现故障，我们只需重置保存“块”的状态表并从中断的地方继续。

我关于复制大型表/数据集的博客文章

使用块创建状态表的关键查询是以下查询：

    --++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
--Author: Jana Sattainathan (@SQLJana - https://sqljana.wordpress.com)
--Sourcecode: https://sqljana.wordpress.com/?p=7980
--++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
--## **** Script to split any table into chunks based on a column to chunk it by and create a status tracking table out of the chunk data **** 
--@TableName could be any schema.table 
--@ChunkByColumn can be of any data type but prefer an indexed PK ID type column for performance!
--@ChunkSize decides how big each chunk is when table is broken up into chunks (ranges of ChunkByColumn values)
 
DECLARE @StatusTrackingTable VARCHAR(50) = '[dbo].[JanasTableStatusTracker]';
DECLARE @SourceServerName VARCHAR(50) = @@SERVERNAME;
DECLARE @SourceDBName VARCHAR(50) = DB_NAME();
DECLARE @SourceTableName VARCHAR(50) = '[Sales].[SalesOrderHeader]';
DECLARE @ChunkByColumn VARCHAR(50) = '[SalesOrderID]';
DECLARE @ChunkSize INT = 100;
DECLARE @TargetServerName VARCHAR(50) = @SourceServerName;
DECLARE @TargetDBName VARCHAR(50) = 'DBAT';
DECLARE @TargetTableName VARCHAR(50) = '[dbo].[SalesOrderHeader_Jana]' --CASE WHEN @TargetServerName IS NOT NULL THEN @SourceTableName ELSE NULL END;
 
DECLARE @Batch BIGINT = CAST(FORMAT(GETDATE(),'yyyyMMddHHmmssffff') AS BIGINT);
DECLARE @BatchString VARCHAR(18) = LTRIM(STR(@Batch,18));
DECLARE @SQL NVARCHAR(1600); 
DECLARE @CreateInsertToStatusTableSQL NVARCHAR(1600); 
 
BEGIN
    SET @SQL = 'SELECT          
            [Batch], ' +
            '''' + @SourceServerName + ''' AS SourceServerName, ' +
            '''' + @SourceDBName + ''' AS SourceDBName, ' +
            '''' + @SourceTableName + ''' AS SourceTableName, ' +
            '''' + @ChunkByColumn + ''' AS ChunkByColumn, ' +
            '''' + COALESCE(@TargetServerName,'') + ''' AS TargetServerName, ' +
            '''' + COALESCE(@TargetDBName,'')  + ''' AS TargetDBName, ' +
            '''' + COALESCE(@TargetTableName,'')  + ''' AS TargetTableName, 
            ChunkNumber,
            MIN(Id) AS MinValue,
            MAX(Id) AS MaxValue,
            CAST(COUNT(1) AS BIGINT) AS ChunkRowCount, 
            /*'' SELECT *  FROM ' + @SourceDBName + '.' + @SourceTableName + ' WHERE ' + @ChunkByColumn + ' BETWEEN ''+' + 'LTRIM(STR(MIN(Id)))' + '+'' AND ''+' + 'LTRIM(STR(MAX(Id)))' + ' + '''' AS SimpleChunkSelectSQLForNumIds, */
            '' SELECT *  FROM ' + @SourceDBName + '.' + @SourceTableName + ' WHERE ' + @ChunkByColumn + ' BETWEEN (SELECT MinValue FROM ' + @SourceDBName + '.' + @StatusTrackingTable + ' WHERE [Batch] = ' + @BatchString + ' AND ChunkNumber = ''+' + 'LTRIM(STR(ChunkNumber))' + '+'') AND (SELECT MaxValue FROM ' + @SourceDBName + '.' + @StatusTrackingTable + ' WHERE [Batch] = ' + @BatchString + ' AND ChunkNumber = ''+' + 'LTRIM(STR(ChunkNumber))' + '+'')' + ' '' AS ChunkSelectSQL, ' +
            CASE WHEN (LEN(COALESCE(@StatusTrackingTable,'')) > 0)
                THEN ' CONVERT(VARCHAR(20), NULL) AS Status, 0 AS TargetBatchRowCount, CONVERT(DATETIME, NULL) AS StartDateTime,  CONVERT(DATETIME, NULL) AS EndDateTime'
                ELSE ''
            END + '
        /*INTO STATUSTABLE*/
        FROM
            (
                SELECT ' +
                    @BatchString + ' AS [Batch],
                    ' + @ChunkByColumn + '  AS Id,
                    ROW_NUMBER() OVER (ORDER BY ' + @ChunkByColumn + ') RowNum,
                    CEILING(ROW_NUMBER() OVER (ORDER BY ' + @ChunkByColumn + ')/' + LTRIM(STR(@ChunkSize)) + '.0 /*INT to REAL*/) AS ChunkNumber
                FROM 
                    ' + @SourceTableName + '
            ) splits
        GROUP BY
            [Batch], ChunkNumber
        ORDER BY
            ChunkNumber';
 
    --Create or insert the data into the status tracking table if a name is provided
    IF (LEN(COALESCE(@StatusTrackingTable,'')) = 0)
    BEGIN
        PRINT @SQL;
        EXECUTE(@SQL);
    END
    ELSE
    BEGIN
        IF OBJECT_ID(@StatusTrackingTable, 'U') IS NOT NULL
        BEGIN
            SET @CreateInsertToStatusTableSQL = 'INSERT INTO ' + @StatusTrackingTable + ' ' + @SQL;
        END;
        ELSE
        BEGIN
            SET @CreateInsertToStatusTableSQL = REPLACE(@SQL, '/*INTO STATUSTABLE*/', (' INTO ' + @StatusTrackingTable + ' '));         
        END;
 
        PRINT @CreateInsertToStatusTableSQL;
        EXECUTE(@CreateInsertToStatusTableSQL);
    END;
END;

Answer 11

对于任何想要完成此任务并随时删除源记录的人，请保持批处理，然后使用 OUTPUT 子句将记录移动到新表。

WHILE (SELECT TOP 1 1 FROM SourceTable) IS NOT NULL
BEGIN
    DELETE TOP (100000)
    FROM SourceTable
    OUTPUT DELETED.* --Either * or list columns
    INTO TargetTable
END;

如果您不想删除，请尝试使用游标从源表中填充表变量，在游标中设置检查，以便当表变量达到约 100K 行时，它会执行插入，然后截断表变量。 （抱歉，我现在没有时间深入探讨这个问题。）

“我能说什么，除了……”

Answer 12

从productDB.dbo.items中选择*到new_items

差不多就是这样了。这是最有效的方法。