【问题标题】:How to properly truncate a staging table in an ETL pipeline?如何正确截断 ETL 管道中的临时表?
【发布时间】:2020-01-31 00:25:43
【问题描述】:

我们有一个 ETL 管道,它针对上传到存储帐户 (Azure) 的每个 CSV 运行。它在 CSV 上运行一些转换并将输出写入另一个位置,也作为 CSV,并调用数据库 (SQL Azure) 上的存储过程,将生成的 CSV 摄取(BULK INSERT)到临时表中。

此管道可以同时执行,因为多个资源可以将文件上传到存储。因此,临时表经常插入数据。

然后,我们有一个预定的 SQL 作业(弹性作业),它触发一个 SP,将数据从暂存表移动到最终表。 此时,我们希望截断/清空 staging 表,以便我们不会在下次执行作业时重新插入它们。

问题是,我们不能确定在从登台表加载到最终表和 truncate 命令之间,没有任何新数据写入登台表,可以在没有先插入到决赛桌。

有没有办法在我们将数据复制到最终表中时锁定临时表,以便尝试写入它的 SP(从 ETL 管道调用)只会等到锁被释放?这是否可以通过使用事务或一些手动锁定命令来实现?

如果没有,最好的处理方法是什么?

【问题讨论】:

  • “截断”与“删除”有点不同。截断需要特殊权限,因为它不会将删除的信息存储在事务日志中。您正在处理多少数据?临时表可能是一种选择。您还可以编写删除和重新创建表格的脚本。
  • 到目前为止的答案已经引入了某种方式“等待”一个文件在另一个文件运行之前完成。 sp_getapplck 解决方案一次只加载一个文件。 “加载”和“阶段”表解决方案仅一次将文件加载的“带宽”扩展为 2 个文件......但仍然一次仅将 1 个文件插入到最终表中。这可以接受吗?阅读您的问题,一次加载单个文件似乎是不可接受的。如果是,那么我不明白为什么每个传入文件的 ID 的系统都不起作用。所有记录都与它们来自的文件相关联。

标签: sql-server locking azure-sql-database etl staging-table


【解决方案1】:

我会建议使用两个相同的临时表的解决方案。让我们将它们命名为 StageLoading 和 StageProcessing。
加载过程将具有以下步骤:
1.一开始两个表都是空的。
2. 我们将一些数据加载到 StageLoading 表中(我假设每次加载都是一个事务)。
3. 当 Elastic 作业启动时,它会:
- ALTER TABLE SWITCH 将所有数据从 StageLoading 移动到 StageProcessing。它将使 StageLoading 为空并为下一次加载做好准备。这是一个元数据操作,所以需要几毫秒,并且它是完全阻塞的,所以将在加载之间完成。
- 将数据从 StageProcessing 加载到最终表格。
- 截断表 StageProcessing。
4. 现在我们为下一个 Elastic 工作做好了准备。

如果我们在 StageProcessing 不为空时尝试进行 SWITCH,ALTER 将失败,这意味着最后一个加载过程失败。

【讨论】:

  • +1 嗯,如果它按预期工作,这当然很有趣。我不知道 ALTER TABLE SWITCH。我会做一些测试并告诉你。
  • 有很好的解释它是如何工作的:stackoverflow.com/questions/41348523/…。一个注释,您可以使用一个完整的非分区表与另一个完整的非分区表进行切换。
  • 这真的允许一次加载多个文件吗?最后,仍然只有 1 个文件被插入到决赛表中。在这种情况下,如果在给定时间仅应用 1 个文件,那么在“加载/暂存”表集中拥有 2 个文件有什么好处。
  • 它允许您将一个或多个进程加载到阶段表中,同时弹性作业消耗阶段表中的数据。因此加载过程不必等待最终处理完成。至于同时插入暂存文件的多个进程,这取决于暂存表的构建方式(堆/行存储或列存储)。尽管如此,您只能从阶段表中使用一个进程,您仍然可以扩展解决方案以拥有多个消费者。
  • 这很好用。它允许多个进程加载到临时表中,然后,一个简单的 ALTER TABLE SWITCH 命令将所有数据移动到将用于填充最终表的另一个表,同时仍然允许其他进程继续加载到临时表中.谢谢!
【解决方案2】:

我喜欢sp_getapplock,并在少数地方自己使用此方法,因为它具有灵活性,并且您可以完全控制锁定逻辑和等待时间。

我看到的唯一问题是,在您的情况下,并发进程并不完全相同。

您拥有将数据从临时表移动到主表的 SP1。您的系统从不尝试运行此 SP 的多个实例。

另一个将数据插入临时表的 SP2可以同时运行多次,这样做很好。

很容易实现阻止任何 SP1 或 SP2 组合的任何并发运行的锁定。换句话说,如果 SP1 和 SP2 的锁定逻辑相同并且它们被视为相同,则很容易。但是,您不能同时运行多个 SP2 实例。

目前尚不清楚如何实现锁定以防止 SP1 和 SP2 并发运行,同时允许多个 SP2 实例同时运行。


还有另一种方法不尝试阻止 SP 的并发运行,但接受并期望同时运行是可能的。

一种方法是将IDENTITY 列添加到临时表。或者一个自动填充的日期时间,如果你可以保证它是唯一的并且永远不会减少,这可能会很棘手。或rowversion 专栏。

SP2 中将数据插入临时表的逻辑不会改变。

SP1 内部将数据从暂存表移动到主表的逻辑需要使用这些标识值。

首先从暂存表中读取当前标识的最大值并将其记住在一个变量中,例如@MaxID。该 SP1 中临时表中的所有后续 SELECT、UPDATE 和 DELETE 都应包含过滤器 WHERE ID <= @MaxID

这将确保如果在 SP1 运行时恰好有新行添加到暂存表中,则该行不会被处理,并将保留在暂存表中,直到 SP1 的下一次运行。

这种方式的缺点是不能使用TRUNCATE,需要DELETEWHERE ID <= @MaxID一起使用。


如果您同意多个 SP2 实例相互等待(和 SP1),那么您可以使用类似于以下内容的sp_getapplock。我的存储过程中有这段代码。您应该将此逻辑放入 SP1 和 SP2。

这里我没有显式调用sp_releaseapplock,因为锁的拥有者设置为事务并且引擎会在事务结束时自动释放锁。

您不必将重试逻辑放在存储过程中,它可以在运行这些存储过程的外部代码中。无论如何,您的代码应该可以重试了。

CREATE PROCEDURE SP2  -- or SP1
AS
BEGIN
    SET NOCOUNT ON;
    SET XACT_ABORT ON;

    BEGIN TRANSACTION;
    BEGIN TRY
        -- Maximum number of retries
        DECLARE @VarCount int = 10;

        WHILE (@VarCount > 0)
        BEGIN
            SET @VarCount = @VarCount - 1;

            DECLARE @VarLockResult int;
            EXEC @VarLockResult = sp_getapplock
                @Resource = 'StagingTable_app_lock',
                -- this resource name should be the same in SP1 and SP2
                @LockMode = 'Exclusive',
                @LockOwner = 'Transaction',
                @LockTimeout = 60000,
                -- I'd set this timeout to be about twice the time
                -- you expect SP to run normally
                @DbPrincipal = 'public';

            IF @VarLockResult >= 0
            BEGIN
                -- Acquired the lock

                -- for SP2
                -- INSERT INTO StagingTable ...

                -- for SP1
                -- SELECT FROM StagingTable ...
                -- TRUNCATE StagingTable ...

                -- don't retry any more
                BREAK;
            END ELSE BEGIN
                -- wait for 5 seconds and retry
                WAITFOR DELAY '00:00:05';
            END;
        END;

        COMMIT TRANSACTION;
    END TRY
    BEGIN CATCH
        ROLLBACK TRANSACTION;
        -- log error
    END CATCH;

END

此代码保证在任何给定时刻只有一个过程在使用临时表。没有并发。所有其他实例将等待。

显然,如果您尝试访问暂存表,而不是通过这些 SP1 或 SP2(它们首先尝试获取锁),那么这种访问不会被阻止。

【讨论】:

  • 这是一个不错的选择。不幸的是,暂存表和最终表都是具有数百万行的聚集列存储表。 DELETE FROM 暂存表与即时 TRUNCATE 相比需要很长时间。将数据插入临时表的 SP 实际上非常快,不到一秒。所以我想如果这些不能同时运行,我们就可以了。他们只会等待锁被释放,然后一个接一个地运行,对吧?
  • @emzero,是的,请看我添加到答案中的代码示例
【解决方案3】:

有没有办法在我们将数据复制到最终表中时锁定临时表,以便尝试写入它的 SP(从 ETL 管道调用)将等待直到锁被释放?这是否可以通过使用事务或一些手动锁定命令来实现?

看起来您正在寻找一种比事务级别更广泛的机制。 SQL Server/Azure SQL DB 有一个,称为应用程序锁

sp_getapplock

锁定应用程序资源。

放置在资源上的锁与当前事务或当前会话相关联。与当前事务关联的锁在事务提交或回滚时被释放。与会话关联的锁在会话注销时被释放。当服务器因任何原因关闭时,所有的锁都会被释放。

可以使用 sp_releaseapplock 显式释放锁。当应用程序为同一个锁资源多次调用 sp_getapplock 时,必须调用相同次数的 sp_releaseapplock 才能释放锁。当使用事务锁所有者打开锁时,该锁会在事务提交或回滚时释放。

这基本上意味着您的 ETL 工具应该打开单个会话到 DB,获取锁并在完成时释放。其他会话在尝试做任何事情之前应该尝试获取锁(它们不能,因为它已经被占用了),等到它释放并继续工作。

【讨论】:

    【解决方案4】:

    假设您有一个外派工作

    • 将 OutboundProcessing BIT DEFAULT 0 添加到表中
    • 在作业中,SET OutboundProcessing = 1 WHERE OutboundProcessing = 0(声明行)
    • 对于 ETL,将 WHERE OutboundProcessing = 1 合并到获取数据的查询中(传输行)
    • 在 ETL 之后,DELETE FROM TABLE WHERE OutboundProcessing = 1(删除您传输的行)
    • 如果 ETL 失败,SET OutboundProcessing = 0 WHERE OutboundProcessing = 1

    【讨论】:

    • 我已经考虑过这一点,但我宁愿使用带有 TRUNCATE 而不是 DELETE FROM 的解决方案。后者需要很长时间。
    • 您可能已经成为过早性能优化的牺牲品。删除运行时,数据已经加载到目的地。
    • 为什么“删除”需要太长时间?看起来文件很小,我假设流程表单文件到最终表的生命周期很快......那么为什么加载表中会有数百万条记录?它们应该被删除,所以我预计加载表中的记录在任何时候都会很少。
    【解决方案5】:

    我总是喜欢对收到的每个文件进行“标识”。如果可以做到这一点,您可以在整个加载过程中关联给定文件中的记录。你没有说需要这个,但只是说。

    但是,每个文件都有一个标识(应该只是一个 int/bigint 标识值),然后您可以从“模板”加载表动态创建任意数量的加载表。

    1. 当文件到达时,创建一个以文件 ID 命名的新加载表。
    2. 处理从加载到最终表格的数据。
    3. 删除正在处理的文件的加载表。

    这有点类似于关于使用 2 个表(加载和暂存)的其他解决方案,但即使在该解决方案中,您仍然仅限于“加载”2 个文件(尽管您仍然只将一个文件应用于最终表?)

    最后,尚不清楚您的“弹性作业”是与实际的“负载”管道/处理分离还是包含在内。作为一个工作,我假设它不包括在内,如果一个工作,你一次只能运行一个实例?因此,如果您一次只能将一个文件从加载移动到最终文件,那么为什么一次加载多个文件很重要呢?为什么急于加载文件?

    【讨论】:

    • 我认为这行不通,因为将数据从暂存表加载到最终表的作业与管道“分离”。它只知道它需要将数据从临时表移动到最终表。它不知道记录属于哪个文件。暂存表中的所有记录都可以来自多个不同的源文件。
    猜你喜欢
    • 2018-12-29
    • 2012-06-10
    • 2012-01-28
    • 2021-08-23
    • 1970-01-01
    • 2017-11-29
    • 1970-01-01
    • 2020-06-02
    • 1970-01-01
    相关资源
    最近更新 更多