Snowflake 使用 Streams 跟踪表的更新/删除答案

【问题标题】：Snowflake Using Streams to Track Updates/Deletes to a TableSnowflake 使用 Streams 跟踪表的更新/删除
【发布时间】：2020-08-13 14:06:27
【问题描述】：

我无法理解 Streams 在跟踪更改方面的工作原理。我想创建一个历史表来跟踪每个 UPDATE 和 DELETE 到一个表，但我发现我不明白这是如何工作的。

如果我有带有 Stream 的表 Table1：

 CREATE TABLE Table1
 (
   XID INT IDENTITY PRIMARY KEY,
   FIELD1 INT,
   FIELD2 STRING,
   DATECREATED TIMESTAMP DEFAULT CURRENT_TIMESTAMP::TIMESTAMP
 );

 CREATE STREAM Table1_History ON TABLE Table1;

如果我插入数据：

INSERT INTO Table1 (FIELD1,FIELD2)
VALUES
(101,'String1'),
(102,'String2')
;

然后运行：

SELECT * FROM Table1_History;

它返回以下内容：

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
1   101 String1 2020-08-13 06:52:34.402 INSERT  FALSE   23bc7a4d83522484f4d7e36edf84b4c7986dfa9b
2   102 String2 2020-08-13 06:52:34.402 INSERT  FALSE   5b5e429cf3a174303b2f2192b5d602ed9dedd865

到目前为止一切顺利。

但如果我跑：

UPDATE Table1 SET FIELD1 = 1001 WHERE XID = 1;

然后从Table1_History中选择，我得到：

SELECT * FROM Table1_History;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
1   1001    String1 2020-08-13 06:52:34.402 INSERT  FALSE   23bc7a4d83522484f4d7e36edf84b4c7986dfa9b
2   102 String2 2020-08-13 06:52:34.402 INSERT  FALSE   5b5e429cf3a174303b2f2192b5d602ed9dedd865

METADATA$ACTION 仍为 INSERT，FIELD1 值现在以 1001 的形式存储在流中。不再有任何记录我可以看到该行的值曾经是 101 并且它已更新。

如果我运行以下命令：

DELETE FROM Table1 WHERE XID = 2;

流现在返回：

SELECT * FROM Table1_History;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
1   1001    String1 2020-08-13 06:52:34.402 INSERT  FALSE   23bc7a4d83522484f4d7e36edf84b4c7986dfa9b

现在我可以在数据库中的第二行的流中看到 0 条记录。

我不明白用于跟踪 UPDATES/DELETES 的 Stream 表的意义。这不是使用流吗？

我试过按照这个：Snowflake Streams Made Simple，但我还是不明白。

【问题讨论】：

标签： sql change-tracking snowflake-cloud-data-platform

【解决方案1】：

引用 Snowflake 文档： “流存储表的当前事务版本，并且在大多数情况下是 CDC 记录的适当来源。”

查看 Snowflake 文档中的此示例：https://docs.snowflake.com/en/user-guide/streams.html#example-1

我的理解是，在您推进偏移量之前，流只会保存记录的当前版本。因此，如果您插入一条记录，然后在推进偏移之前对其进行更新，那么它将显示单个插入，但字段将保存最新值。

如果您随后推进偏移量并更新或删除记录，那么这些事件将显示在流中 - 但如果您更新然后删除相同的记录（在推进偏移量之前），流只会显示删除，因为那是该记录的最后一个位置。

更新 1 听起来您正在尝试对表中记录的每项更改实施审计跟踪 - 这不是 Streams 的设计目的，我认为您无法使用 Streams 实施解决方案，保证记录每个更改。

如果您阅读 Streams 文档，它会声明“流可以提供从当前偏移量到源表的当前事务时间的一组更改（即表的当前版本）。流仅维护更改；如果多个 DML 语句更改一行，则流仅包含对该行执行的最新操作。"

CDC 是一个专门与加载数据仓库相关的术语，绝不是用于捕获对记录所做的每项更改的通用术语。

如果您想在 Snowflake 中创建真正的审计功能，那么恐怕我不知道这是否可能。时间旅行功能显示 Snowflake 保留了对记录所做的所有更改（在保留期内），但我不知道有任何方式可以访问这些更改；我认为您只能在某个时间点访问记录的历史记录，而您无法知道在什么时间进行了任何更改

更新 2 刚刚意识到 Snowflake 允许在表上进行更改跟踪，而不必使用 Streams。如果您想捕获对表的所有更改，而不仅仅是最新版本，这可能是一个更好的解决方案。此处记录了该功能： https://docs.snowflake.com/en/sql-reference/constructs/changes.html

【讨论】：

我现在开始明白这一点，但这意味着我必须在运行UPDATE/DELETE 之前将来自Table1_History 的记录INSERT 移到另一个表中以向前移动偏移量。这意味着您需要两个表来跟踪更改：流和流目标，以便在每个可能影响原始表的进程之后转储流记录。我想我将创建两个表和一个流：Table1、Table1_History 和Table1_Stream，其中 History 表实际上永久存储了所有更改。你同意这是必要的设计吗？
我看到了您的修改并将您的答案标记为正确，但我不同意“审计跟踪是不可能的”。审计跟踪只要求您在每次 DML 更改操作后转储流。对于托管表，您可以轻松地将流记录加载到另一个表中。在标准 MS SQL 数据库中，您可以在 UPDATE/DELETE 上使用 TRIGGER 执行此操作。由于 Snowflake 不支持触发器，因此您只需将 INSERT 作为所有 DML 操作的一部分运行到历史记录表中。我将在新的回复中发布该操作。

【解决方案2】：

好吧，正如@NickW 所说，流表更多的是关于跟踪偏移之间的变化。这意味着我仍然可以做我想做的事，但它需要在 DML 操作之间的历史表中显式地添加 INSERT。

首先创建主表、流和历史表：

CREATE TABLE Table1
(
   XID INT IDENTITY PRIMARY KEY,
   FIELD1 INT,
   FIELD2 STRING,
   DATECREATED TIMESTAMP DEFAULT CURRENT_TIMESTAMP::TIMESTAMP
);

CREATE STREAM Table1_Stream ON TABLE Table1;

CREATE TABLE Table1_History
(
   UID INT IDENTITY PRIMARY KEY,
   XID INT,
   FIELD1 INT,
   FIELD2 STRING,
   DATECREATED TIMESTAMP,
   METADATA$ACTION STRING,   --METADATA Column from Stream
   METADATA$ISUPDATE STRING, --METADATA Column from Stream
   DATEINSERTED TIMESTAMP DEFAULT CURRENT_TIMESTAMP::TIMESTAMP
);

然后INSERT 记录：

INSERT INTO Table1 (FIELD1,FIELD2)
VALUES
(101,'String1'),
(102,'String2')
;

流表现在显示：

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
1   101    String1 2020-08-13 06:52:34.402 INSERT  FALSE   23bc7a4d83522484f4d7e36edf84b4c7986dfa9b
2   102 String2 2020-08-13 06:52:34.402 INSERT  FALSE   5b5e429cf3a174303b2f2192b5d602ed9dedd865

然后从 Streams 表到 History 表执行 INSERT：

INSERT INTO Table1_History(XID,FIELD1,FIELD2,DATECREATED,
                           METADATA$ACTION,METADATA$ISUPDATE)
SELECT XID,FIELD1,FIELD2,DATECREATED,
       METADATA$ACTION,UPPER(METADATA$ISUPDATE)
FROM Table1_Stream
WHERE METADATA$ACTION <> 'INSERT'
OR METADATA$ISUPDATE = 'TRUE'

注意WHERE 子句，它只删除INSERT 的记录，除非它是UPDATE 的一部分，DELETEs 然后INSERTs 是一个记录。

现在，即使由于WHERE 子句，流记录实际上并未插入历史记录表，但如果您查询流，您会得到 NULL：

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID

现在，如果您执行UPDATE，则流将显示它：

UPDATE Table1 SET FIELD1 = 1001 WHERE XID = 1;

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
1   1001    String1 2020-08-14 09:11:20.173 INSERT  TRUE    93256f240f338581cc4781c2e79a28075e1b66d7
1   101 String1 2020-08-14 09:11:20.173 DELETE  TRUE    93256f240f338581cc4781c2e79a28075e1b66d7

现在运行插入：

INSERT INTO Table1_History(XID,FIELD1,FIELD2,DATECREATED,
                           METADATA$ACTION,METADATA$ISUPDATE)
SELECT XID,FIELD1,FIELD2,DATECREATED,
       METADATA$ACTION,UPPER(METADATA$ISUPDATE)
FROM Table1_Stream
WHERE METADATA$ACTION <> 'INSERT'
OR METADATA$ISUPDATE = 'TRUE'

SELECT * FROM Table1_History;

UID XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   DATEINSERTED
1   1   1001    String1 2020-08-14 09:11:20.173 INSERT  TRUE    2020-08-14 09:13:41.474
2   1   101 String1 2020-08-14 09:11:20.173 DELETE  TRUE    2020-08-14 09:13:41.474

Stream 再次为 NULL：

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID

如果您运行DELETE，则更改会再次反映在流中：

DELETE FROM Table1 WHERE XID = 2

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID
2   102 String2 2020-08-14 09:11:20.173 DELETE  FALSE   51d1d0f5c5bf9c328d79cbbd54a10bf99f73bcd3

而且可以INSERT进入历史表：

INSERT INTO Table1_History(XID,FIELD1,FIELD2,DATECREATED,
                           METADATA$ACTION,METADATA$ISUPDATE)
SELECT XID,FIELD1,FIELD2,DATECREATED,
       METADATA$ACTION,UPPER(METADATA$ISUPDATE)
FROM Table1_Stream
WHERE METADATA$ACTION <> 'INSERT'
OR METADATA$ISUPDATE = 'TRUE'

SELECT * FROM Table1_History;

UID XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   DATEINSERTED
1   1   1001    String1 2020-08-14 09:11:20.173 INSERT  TRUE    2020-08-14 09:13:41.474
2   1   101 String1 2020-08-14 09:11:20.173 DELETE  TRUE    2020-08-14 09:13:41.474
4   2   102 String2 2020-08-14 09:11:20.173 DELETE  FALSE   2020-08-14 09:17:37.694

Stream 再次为 NULL：

SELECT * FROM Table1_Stream;

XID FIELD1  FIELD2  DATECREATED METADATA$ACTION METADATA$ISUPDATE   METADATA$ROW_ID

我不明白的一件事是为什么Table1_History 上的UID 增加到4 而不是3，但这是一个小问题。

这就是我将跟踪 Snowflake 中所有历史变化的方式。

【讨论】：

嗨 - 如果您每次更新主表时都可以更新历史记录表，那么我不确定使用 Streams 是否会给流程增加任何东西（除了复杂性）。为什么不每次更新主表时都更新历史表？
我喜欢这个，因为它明显地将加载/合并步骤与数据归档步骤分开。此外，如果有人确实进行了临时更新/删除，他们将被自动捕获。
“如果有人确实进行了临时更新/删除，他们将被自动捕获”。如果您在临时更新/删除后立即将数据插入历史记录表，那么它们肯定只会被捕获（以您想要的方式，而不是以 Streams 的设计方式）？如果它是临时的，那么你怎么知道这样做呢？如果您允许临时更改，我不清楚您的方法将如何工作 - 因为您的方法依赖于在每次更改后将流数据写入历史表，即硬编码“INSERT INTO Table1_History ...”之后每个 DML
您可以在MERGE-SERT 前面加上INSERT INTO HISTORY 代码，这样它将运行两次。这意味着流捕获两个偏移量：原始表和MERGE-SERT 之间的增量，然后是最后一个MERGE-SERT 末尾的表之间的增量，然后在再次运行MERGE-SERT 之前（尽管它不会如果同一记录被临时更改多次）。理论上不应该有临时更新，因为所有数据都应该通过MERGE-SERT 过程运行，但实际上有时您必须即时进行更改。
当您说“有人确实进行了临时更新/删除”时，我假设除了您之外还有其他人可能正在执行这些 DML 语句。如果是这种情况，那么您是否仍然依靠他们来了解您概述的流程并与他们正在运行的 DML 一起正确执行它？这对我来说似乎不是一个强大的解决方案，但显然取决于你。我仍然会选择更改跟踪方法（请参阅我的答案中的更新 2），因为这似乎更容易实现，并且不依赖于知道他们必须执行额外的 DML 的人

【解决方案3】：

流不是表格。

流仅测量自您上次使用（或最初定义）流以来的增量。您目前看到的是在通过流拍摄另一个快照之前，您只是不断覆盖数据。初始插入值是自上次捕获数据以来发生的变化。通过将插入值注销到另一个表，创建了一个新的快照来测量变化。

我建议你创建一个表：创建表 Table_history (...)

对 table1 进行初始插入并查看流：（从 Table1_History 中选择 *）；

现在从您定义的 STREAM 中插入历史记录表：

插入到 table_history (...) 从 Table1_History 中选择 * ;

再次检查流：select * from Table1_History 数据应该没有了。什么都没有。

然后进行更新： UPDATE Table1 SET FIELD1 = 1001 WHERE XID = 1;

再次检查流： select * from Table1_History

【讨论】：

是的，这是令人讨厌的事情，没有标准化的命令来创建一个可查询的表并跟踪更改，相反，您必须在您可能已经更改的任何时间之后将流插入另一个表桌子。我认为如果有像 MS SQL 的 SYSTEM_VERSIONING 这样的魔术命令来跟踪更改会很好，因为这是相当标准的。