BigQuery 中的行级原子合并替换

Question

对于我的用例，我正在处理可通过源中唯一键识别的数据，这些数据会分解为 n（非确定性）数量的目标条目，这些条目已加载到 BigQuery 表中以进行分析。

构建此 ETL 以使用 Mongo 最近的更改流功能我想删除 BigQuery 中的所有条目，然后自动加载新条目。

探索 BigQuery DML 我看到支持 MERGE 操作，但只有 WHEN MATCHED THEN DELETE 或 WHEN MATCHED THEN UPDATE 是可能的。

我对 WHEN MATCHED THEN DELETE，然后是 INSERT 操作感兴趣。

我如何在 BigQuery 中实现此类 ETL，同时在数据可用性和正确性方面尽可能保持原子性或最终一致。

---

编辑1：我想提供一个具体的例子来详细说明。

我在这个数据集上拥有的最小唯一性粒度是user_id。行不是唯一可识别的。

示例

1.

从 mongo 更改流接收的更新用户对象：

user={_id: "3", name="max", registered="2018-07-05" q=["a", "b", "c"]}

2.

当前 BigQuery.user_q 保留

| user_id | q |
...
|    3    | a |
|    3    | b |
...

3.

转换代码将修改后的用户对象加载到 BigQuery.user_q_incoming

| user_id | q |
|    3    | a |
|    3    | b |
|    3    | c |

4.

user_q 和 user_q_incoming 之间的合并：

1. user_q 中属于 user_id 3 的 2 行被删除
2. user_q_incoming 中属于 user_id 3 的 3 行已插入。
3. user_q 中的其余数据 (...) 保留在原处，未修改。

5.

BigQuery.user_q 保留

| user_id | q |
...
|    3    | a |
|    3    | b |
|    3    | c |
...

例如，用户可能会从他的个人资料中删除一个问题。将剩余的行保留为q=["a", "c"]。我也需要将其转化为 BigQuery 的世界观。

Answer 1

BigQuery DML 支持 INSERT

MERGE 语句是一种 DML 语句，它可以将 INSERT、UPDATE 和 DELETE 操作组合到一个语句中并以原子方式执行这些操作。

例如

MERGE dataset.Inventory T
USING dataset.NewArrivals S
ON FALSE
WHEN NOT MATCHED AND product LIKE '%washer%' THEN
  INSERT (product, quantity) VALUES(product, quantity)
WHEN NOT MATCHED BY SOURCE AND product LIKE '%washer%' THEN
  DELETE   

所以，你应该很好地使用你的 ETL

根据添加到问题的更具体细节进行编辑

好的，我明白了 - 我认为在这种情况下 MERGE 将不适用，因为 INSERT 只能用于 NOT MATCH 子句。有人可能会想出如何欺骗 MERGE 在这种情况下工作，但与此同时，下面的解决方案可以实现您想要实现的目标 - 我认为是这样 :o)

CREATE OR REPLACE TABLE `project.dataset.user_q` (user_id INT64, q STRING) AS
SELECT * FROM `project.dataset.user_q`
WHERE NOT user_id IN (SELECT DISTINCT user_id FROM `project.dataset.user_q_incoming`)
UNION ALL
SELECT * FROM `project.dataset.user_q_incoming`
WHERE user_id IN (SELECT DISTINCT user_id FROM `project.dataset.user_q`)

Answer 2

有一个类似的问题和一个让 MERGE 工作的方法 (https://issuetracker.google.com/issues/35905927#comment9)。

基本上，像下面这样的东西应该可以工作，

MERGE `project.dataset.user_q` T
USING (
  SELECT *, false AS is_insert FROM `project.dataset.user_q_incoming`
UNION ALL
  SELECT *, true AS is_insert FROM `project.dataset.user_q_incoming`
) S
ON T.user_id = S.user_id and NOT is_insert
WHEN MATCHED THEN
  DELETE
WHEN NOT MATCHED AND is_insert THEN
  INSERT(user_id, q) VALUES(user_id, q)

理想情况下，以下是您需要的，但尚不支持。

MERGE `project.dataset.user_q`
USING `project.dataset.user_q_incoming`
ON FALSE
WHEN NOT MATCHED BY TARGET THEN
  INSERT(user_id, q) VALUES(user_id, q)
WHEN NOT MATCHED BY SOURCE AND user_id in (SELECT user_id FROM `project.dataset.user_q_incoming`) THEN
  DELETE