【问题标题】:How can I load a large flat file into a database table using SSIS?如何使用 SSIS 将大型平面文件加载到数据库表中?
【发布时间】:2011-08-28 23:58:10
【问题描述】:

我不确定它是如何工作的,所以我正在寻找合适的解决方案。我认为 SSIS 是正确的方法,但我以前从未使用过它

场景:

每天早上,我都会收到一个包含 800K 记录的制表符分隔文件。我需要将它加载到我的数据库中:

  1. 从 ftp 或本地获取文件
  2. 首先,我需要从数据库中删除新文件中不存在的文件;
    • 如何比较 tsql 中的数据
    • 我应该在哪里加载制表符分隔文件中的数据以便与文件进行比较?我应该使用临时表吗? ItemID 是表中的唯一列。
  3. 其次,我只需要将新记录插入数据库。
  4. 当然,它应该是自动化的。
  5. 这应该是一种不会过热 SQL 数据库的有效方法

不要忘记该文件包含 80 万条记录。

示例平面文件数据:

ID  ItemID  ItemName  ItemType
--  ------  --------  --------
 1  2345    Apple     Fruit
 2  4578    Banana    Fruit

我该如何解决这个问题?

【问题讨论】:

    标签: sql-server-2008 ssis


    【解决方案1】:

    是的,SSIS 可以执行您在问题中指定的要求。以下示例应该让您了解如何完成它。示例使用 SQL Server 作为后端。下面提供了在包上执行的一些基本测试场景。很抱歉回答冗长。

    分步过程:

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

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

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

    4. 如屏幕截图 #6 所示配置控制流任务。配置任务Pre LoadPost LoadLoop 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 SplitDestination SplitStaging 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.
    
    1. 程序包在 64 位计算机上执行,具有 Xeon 单核 CPU 2.5GHz 和 3.00 GB RAM。

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

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

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

    希望对您有所帮助。

    测试查询: .

    --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:

    【讨论】:

    • 这个答案几乎让我见过的所有其他 SSIS 答案都感到羞耻。干得好。
    【解决方案2】:

    假设您使用的是 SQL 代理(或类似的调度程序)

    要求 1/4) 我将有一个前导步骤来处理 FTP 和/或文件复制步骤。如果可以避免的话,我不喜欢用文件操作来弄乱我的包。

    要求 2/3) 在控制流级别,包设计看起来像一个连接到数据流的执行 SQL 任务,该数据流连接到另一个执行 SQL 任务。正如@AllenG 所指出的,最好通过数据流任务加载到临时表中。第一个执行 SQL 任务将清除暂存表中的所有行 (TRUNCATE TABLE dbo.DAILY_STAGE)

    大致的表格设计如下所示。 MICHAEL_BORN 表是您现有的表,而 DAILY_STAGE 是您的数据流所在的位置。

    CREATE TABLE DBO.MICHAEL_BORN
    (
        ID int identity(1,1) NOT NULL PRIMARY KEY CLUSTERED
    ,   ItemID int NOT NULL
    ,   ItemName varchar(20) NOT NULL
    ,   ItemType varchar(20) NOT NULL
    )
    CREATE TABLE dbo.DAILY_STAGE
    (
        ItemID int NOT NULL PRIMARY KEY CLUSTERED
    ,   ItemName varchar(20) NOT NULL
    ,   ItemType varchar(20) NOT NULL
    )
    

    出于演示目的,我将通过 TSQL 将示例数据加载到上述表格中

    -- Original data
    INSERT INTO
        dbo.MICHAEL_BORN
    VALUES
        (2345,'Apple','Fruit')
    ,   (4578, 'Bannana','Fruit')
    
    
    -- Daily load runs
    -- Adds a new fruit (pear), corrects misspelling of banana, eliminates apple
    INSERT INTO
        dbo.DAILY_STAGE
    VALUES
        (7721,'Pear','Fruit')
    ,   (4578, 'Banana','Fruit')
    

    执行 SQL 任务将利用 SQL Server 2008+ 版本中提供的MERGE 语句。请注意,结尾的分号是 MERGE 语句的一部分。不包含它会导致“A MERGE 语句必须以分号 (;) 终止”的错误。

    -- MERGE statement
    -- http://technet.microsoft.com/en-us/library/bb510625.aspx
    -- Given the above scenario, this script will
    -- 1)  Update the matched (4578 bannana/banana) row
    -- 2)  Add the new (pear) row
    -- 3)  Remove the unmatched (apple) row
    
    MERGE
        dbo.[MICHAEL_BORN] AS T
    USING
    (
        SELECT
            ItemID
        ,   ItemName
        ,   ItemType
        FROM
            dbo.DAILY_STAGE
    
    ) AS S
    ON T.ItemID = S.ItemID
    WHEN
        MATCHED THEN
        UPDATE
        SET
            T.ItemName = S.ItemName
        ,   T.ItemType = S.ItemType
    WHEN
        NOT MATCHED THEN
        INSERT
        (
            ItemID
        ,   ItemName
        ,   ItemType
        )
        VALUES
        (
            ItemID
        ,   ItemName
        ,   ItemType
        )
    WHEN
        NOT MATCHED BY SOURCE THEN
        DELETE
        ;
    

    要求 5) 效率完全取决于您的数据以及行的宽度,但应该不会很糟糕。

    -- Performance testing
    -- Assumes you have a similar fast row number generator function
    -- http://billfellows.blogspot.com/2009/11/fast-number-generator.html
    
    TRUNCATE TABLE dbo.MICHAEL_BORN
    TRUNCATE TABLE dbo.DAILY_STAGE
    
    -- load initial rows
    -- 20ish seconds
    INSERT INTO
        dbo.MICHAEL_BORN
    SELECT
        N.number AS ItemID
    ,   'Spam & eggs ' + CAST(N.number AS varchar(10)) AS ItemName
    ,   'SPAM' AS ItemType
    --, CASE N.number % 2 WHEN 0 THEN N.number + 1000000 ELSE N.number END AS UpTheEvens
    FROM
        dbo.GenerateNumbers(1000000) N
    
    
    -- Load staging table
    -- Odds get item type switched out
    -- Evens get delete and new ones created
    -- 20ish seconds
    INSERT INTO
        dbo.DAILY_STAGE
    SELECT
        CASE N.number % 2 WHEN 0 THEN N.number + 1000000 ELSE N.number END AS ItemID
    ,   'Spam & eggs ' + CAST(N.number AS varchar(10)) AS ItemName
    ,   CASE N.number % 2 WHEN 0 THEN 'SPAM' ELSE 'Not much spam' END AS ItemType
    FROM
        dbo.GenerateNumbers(1000000) N
    
    
    -- Run MERGE statement, 32 seconds 1.5M rows upserted
    -- Probably fast enough for you
    

    【讨论】:

      【解决方案3】:

      我只是想为下一个可能通过这个问题的人提供我的想法。因此,我将为每个场景提出我的想法。
      1. 从 FTP 或本地获取文件。
      我建议您使用 Drop box、Google Drive 或您选择的任何其他文件同步云服务,请参阅this 链接了解详细信息。
      2.我建议按照您的建议将所有平面文件数据加载到暂存表然后通过在您的暂存表和您的唯一列(ID)上的目标表之间使用 MERGE 轻松比较数据。您可以查看this 链接了解如何使用合并脚本。如果您使用 MERGE 脚本,将解决第 2 和第 3 种情况。
      对于最后两种情况,我建议您使用 SQL JOB 自动运行程序包并在下班时间或服务器不忙的时间安排它。请查看链接以获取有关如何使用 SQL 运行程序包的详细信息Server Agent Job 只需在您最喜欢的搜索引擎上输入它,您就会发现大量的博客展示了它是如何完成的。

      【讨论】:

        【解决方案4】:

        SSIS 听起来不错。我之前看到您处理的问题类型的方式是使用临时表。新文档加载到暂存表中 - 然后比较暂存和生产 - 从生产中归档(不只是删除)过时的记录,更新具有一些更改的现有行(再次,在某处归档原始数据),并插入新行。

        注意:您对“过时”的定义需要非常、非常精确。例如:是否应该仅仅因为您最近的文件中不存在匹配的行而将某些内容归档?如果它出现在后续文件中,是否应该保留 X 时间?应该考虑这些和其他问题。

        几乎所有标准 SSIS 教程都应该为您指明如何执行每个步骤的正确路径。

        【讨论】:

          【解决方案5】:

          我会给 Merge 一个机会。确保您最终在两个表的 ItemID 上都有索引。

          Merge [dbo].[ItemInfo] as target
          using
          (
              SELECT stg.ItemID, stg.ItemName, stg.ItemType
              FROM [dbo].[ItemInfo_Staging] stg
              LEFT OUTER JOIN [dbo].[ItemInfo] final
                  on stg.ItemId = final.ItemId
          ) as SOURCE
          ON SOURCE.ItemID = target.ItemID
          
          WHEN MATCHED THEN
              Update SET
                  target.ItemID = SOURCE.ItemID
                  , target.ItemName = SOURCE.ItemName
                  , target.ItemType = SOURCE.ItemType
          
          WHEN NOT MATCHED BY TARGET THEN
              INSERT (ItemID, ItemName, ItemType )
                  VALUES (SOURCE.ItemID, SOURCE.ItemName, SOURCE.ItemType ) 
          
          WHEN NOT MATCHED BY SOURCE THEN
              DELETE
          ;
          

          【讨论】:

            猜你喜欢
            • 1970-01-01
            • 2011-01-07
            • 1970-01-01
            • 2017-03-10
            • 1970-01-01
            • 1970-01-01
            • 2011-09-08
            • 2020-11-17
            • 2017-12-08
            相关资源
            最近更新 更多