合并 CDC 雪花的数据逻辑

Question

我正在研究基于 product_id 和 extraction_date 将一个 table（1000 多行）中的重复行合并到另一个空 table 中的逻辑。 extraction_date 用于判断哪些数据是最新的。如果 extraction_date 在多行中相同，我的逻辑就会失败。这是一个例子：

create or replace table source(v variant); 
 INSERT INTO source SELECT parse_json('{
  "pd": {
    "extraction_date": "1644471240",
    "product_id": "357946",
   "retailerName": "retailer",
    "productName":"product"
  }
}');
 INSERT INTO source SELECT parse_json('{
  "pd": {
    "extraction_date": "1644471240",
    "product_id": "357946",
   "retailerName": "retailer2",
    "productName":"product2"
  }
}');

//合并逻辑：

create or replace TABLE target AS 
SELECT * from source 
where v:pd:extraction_date not in
(SELECT c1.v:pd:extraction_date 
 FROM source c1, source c2 where 
 c1.v:pd:product_id=c2.v:pd:product_id and
 c1.v:pd:extraction_date>c2.v:pd:extraction_date);

在给定的示例中，目标 table 将有两个重复的行，因为相同的 extraction_date。请建议我一个修改，当 extraction_date 相同时只选择一行。

Answer 1

不要自行加入，只需使用 QUALIFY/ROW_NUMBER 强制每个 product_id 一个值如果有超过 2 行具有相同的“最新”日期，它将随机选择一个且仅一个。

create or replace TABLE target AS 
SELECT s.* 
from source as s 
QUALIFY row_number() over (partition by s.v:pd:product_id order by s.v:pd:extraction_date) = 1;

如果您想对 retailerName 或 productName 之类的额外详细信息进行排序，请将它们放在 ORDER BY 中，这将更加“可预测”，这正是您通常想要的。或者甚至散列 json blob，因此它在多次运行中都选择相同的 .. 这可能有助于以后的调试。

此外，虽然变体数据与原生列“几乎”相同，但如果您的数据“完全相同”，则使用原生列会有所帮助，并且保留 json（并且从不访问它）可以提高性能，同时允许处理数据改变形状并且您需要处理新旧形状的那一天。

合并 CDC 雪花的数据逻辑

Merge data logic for CDC snowflake

sql

snowflake-schema

oracle-cdc

snowflake-cloud-data-platform