如何在使用 SSIS 包导入大型数据集时执行查找？

Question

我有一个问题 - 我希望通过 SSIS (Microsoft ETL) 下载客户数据。对于我导入的每个客户记录，我必须检查目标数据库中的现有客户记录。如果记录存在，我必须更新记录，如果不存在，我必须插入它。最佳性能的策略是什么？我们说的是 20-30 百万行。

我想到了触发器（INSERT 上的触发器 - 检查记录是否存在）、查找、游标，但我确信有更好的方法，可能使用 SSIS 中的数据流转换之一？

还是将其导入临时表然后在本地进行查找更好？

Answer 1

这是一种可能的选项，您可以应用该选项将数据加载到目标表中，无论它是否涉及插入/更新。下面给出的示例在 3 分钟内将包含 100 万行的文本文件加载到 SQL 表中。如果 SQL 表中不存在记录，则将其插入到目标表中，否则将记录插入到临时表中，然后在存储过程的帮助下，该表将用于更新目标表。存储过程还将删除任何可能存在于 SQL 表中但不存在于平面文件中的记录。该示例使用带有 SQL Server 2008 R2 后端的 SSIS 2008 R2。

分步过程：

1. 在 SQL Server 数据库中，创建两个表，即dbo.ItemInfo 和dbo.Staging。 脚本部分下提供了创建表查询。这些表的结构显示在屏幕截图 #1 中。 ItemInfo 将保存实际数据，Staging 表将保存暂存数据以比较和更新实际记录。这两个表中的Id 列是自动生成的唯一标识列。表ItemInfo 中的IsProcessed 列将用于识别和删除不再有效的记录。

2. 创建一个 SSIS 包并创建 5 个变量，如屏幕截图 #2 所示。我为制表符分隔的文件使用了.txt 扩展名，因此变量FileExtension 中的值*.txt。 FilePath 变量将在运行时赋值。 FolderLocation 变量表示文件所在的位置。 SQLPostLoad 和SQLPreLoad 变量表示在预加载和加载后操作期间使用的存储过程。 脚本部分提供了这些存储过程的脚本。

3. 创建一个指向 SQL Server 数据库的 OLE DB 连接。创建一个平面文件连接，如屏幕截图 #3 和 #4 所示。 平面文件连接列部分包含列级信息。屏幕截图 #5 显示了列数据预览。

4. 如屏幕截图 #6 所示配置控制流任务。配置任务Pre Load、Post Load 和Loop Files，如屏幕截图#7 - #10 所示。 Pre Load 将截断 staging 表并将 ItemInfo 表中所有行的 IsProcessed 标志设置为 false。 Post Load 将更新更改并删除数据库中未在文件中找到的行。请参阅这些任务中使用的存储过程，以了解这些 Execute SQL 任务中正在执行的操作。

5. 双击 Load Items 数据流任务并按照屏幕截图 #11 所示进行配置。 Read File 是配置为使用平面文件连接的平面文件源。 Row Count 是派生列转换，其配置显示在屏幕截图 #12 中。 Check Exist 是一个查找转换，其配置显示在屏幕截图 #13 - #15 中。 查找不匹配输出被重定向到左侧的Destination Split。 查找匹配输出被重定向到左侧的Staging Split。 Destination Split 和 Staging Split 具有完全相同的配置，如屏幕截图 #16 所示。目的地和临时表有 9 个不同目的地的原因是为了提高包的性能。

6. 所有目标任务 0 - 8 都配置为将数据插入表 dbo.ItemInfo，如屏幕截图 #17 所示。如屏幕截图 #18 所示，所有暂存任务 0 - 8 都配置为将数据插入到 dbo.Staging。

7. 在平面文件连接管理器上，将 ConnectionString 属性设置为使用变量 FilePath，如屏幕截图 #19 所示。这将使包在遍历文件夹中的每个文件时能够使用变量中设置的值。

测试场景：

Test results may vary from machine to machine. 
In this scenario, file was located locally on the machine. 
Files on network might perform slower. 
This is provided just to give you an idea. 
So, please take these results with grain of salt.

8. 程序包在 64 位计算机上执行，具有 Xeon 单核 CPU 2.5GHz 和 3.00 GB RAM。

9. 加载了带有1 million rows 的平面文件。包在大约 2 分 47 秒内执行。请参阅截图 #20 和 #21。

10. 使用测试查询部分下提供的查询来修改数据，以在包的第二次运行期间模拟更新、删除和创建新记录。

11. 在数据库中执行以下查询后，加载了包含 1 million rows 的同一文件。包在大约 1 分 35 秒内执行。请参阅屏幕截图 #22 和 #23。请注意屏幕截图 #22 中重定向到目标和暂存表的行数。

以上答案也是提供给this Stack Overflow问题的答案的副本。

希望对您有所帮助。

测试查询： .

--These records will be deleted during next run 
--because item ids won't match with file data.
--(111111 row(s) affected)
UPDATE dbo.ItemInfo SET ItemId = 'DEL_' + ItemId WHERE Id % 9 IN (3)

--These records will be modified to their original item type of 'General'
--because that is the data present in the file.
--(222222 row(s) affected)
UPDATE dbo.ItemInfo SET ItemType = 'Testing' + ItemId WHERE Id % 9 IN (2,6)

--These records will be reloaded into the table from the file.
--(111111 row(s) affected)
DELETE FROM dbo.ItemInfo WHERE Id % 9 IN (5,9)

平面文件连接列 .

Name        InputColumnWidth     DataType          OutputColumnWidth
----------  ----------------     ---------------   -----------------
Id          8                    string [DT_STR]   8
ItemId      11                   string [DT_STR]   11
ItemName    21                   string [DT_STR]   21
ItemType    9                    string [DT_STR]   9

脚本： （用于创建表和存储过程） .

CREATE TABLE [dbo].[ItemInfo](
    [Id] [int] IDENTITY(1,1) NOT NULL,
    [ItemId] [varchar](255) NOT NULL,
    [ItemName] [varchar](255) NOT NULL,
    [ItemType] [varchar](255) NOT NULL,
    [IsProcessed] [bit] NULL,
    CONSTRAINT [PK_ItemInfo] PRIMARY KEY CLUSTERED ([Id] ASC),
    CONSTRAINT [UK_ItemInfo_ItemId] UNIQUE NONCLUSTERED ([ItemId] ASC)) ON [PRIMARY]
GO

CREATE TABLE [dbo].[Staging](
    [Id] [int] IDENTITY(1,1) NOT NULL,
    [ItemId] [varchar](255) NOT NULL,
    [ItemName] [varchar](255) NOT NULL,
    [ItemType] [varchar](255) NOT NULL,
 CONSTRAINT [PK_Staging] PRIMARY KEY CLUSTERED ([Id] ASC)) ON [PRIMARY]
GO

CREATE PROCEDURE [dbo].[PostLoad]
AS
BEGIN
    SET NOCOUNT ON;

    UPDATE      ITM
    SET         ITM.ItemName    = STG.ItemName
            ,   ITM.ItemType    = STG.ItemType 
            ,   ITM.IsProcessed = 1
    FROM        dbo.ItemInfo    ITM
    INNER JOIN  dbo.Staging     STG
    ON          ITM.ItemId      = STG.ItemId;

    DELETE FROM dbo.ItemInfo
    WHERE       IsProcessed = 0;
END
GO

CREATE PROCEDURE [dbo].[PreLoad]
AS
BEGIN
    SET NOCOUNT ON;

    TRUNCATE TABLE dbo.Staging;     

    UPDATE  dbo.ItemInfo 
    SET     IsProcessed = 0;
END
GO

屏幕截图 #1：

屏幕截图 #2：

截图#3：

屏幕截图 #4：

屏幕截图 #5：

屏幕截图 #6：

截图#7：

截图#8：

屏幕截图 #9：

屏幕截图 #10：

屏幕截图 #11：

屏幕截图 #12：

屏幕截图 #13：

屏幕截图 #14：

屏幕截图 #15：

屏幕截图 #16：

屏幕截图 #17：

屏幕截图 #18：

屏幕截图 #19：

屏幕截图 #20：

屏幕截图 #21：

屏幕截图 #22：

屏幕截图 #23：