如何将含可变更新列的CDC数据合并至Snowflake？

我有从Oracle系统以JSON格式传入S3 Bucket的CDC数据，插入记录格式如下：

{
  "operation": "I",
  "position": "00000000000010911785",
  "foo": 1,
  "bar": 678594,
  "cat_date": "2019-10-15 21:50:31.000000000",
  "dogs": 1388,
  "elephants": 72
}

但更新记录仅包含更新列、主键（bar）及CDC元数据：

{
  "operation": "U",
  "position": "00000000000010911999",
  "bar": 678594,
  "dogs": 2500,
  "elephants": null
}

在Snowflake中处理这类更新时，不仅要应对空值与非空值互转的情况，还因更新字段不全，目前只能按position顺序逐行动态生成MERGE语句：

MERGE INTO db.schema.target t
    USING src_db.src_schema.src_tbl s ON t.bar = s.bar
            WHEN MATCHED and s.operation = 'U' THEN UPDATE SET
            {dynamic list of columns}

请问是否有更优的处理方案？

以下几种方案可替代逐行生成动态MERGE的方式，提升处理效率与可维护性：

1. 利用半结构化函数实现统一MERGE语句 #

借助Snowflake的半结构化数据处理能力，将CDC记录转为对象后，结合GET和COALESCE实现字段的智能更新：

MERGE INTO db.schema.target t
USING (
    SELECT 
        bar,
        operation,
        -- 若源表存储原始JSON字符串，需用PARSE_JSON转换：PARSE_JSON(raw_json_col) AS cdc_obj
        OBJECT_CONSTRUCT(*) AS cdc_obj
    FROM src_db.src_schema.src_tbl
    WHERE operation IN ('I', 'U')
    ORDER BY position -- 保证CDC事件顺序
) s
ON t.bar = s.bar
WHEN MATCHED AND s.operation = 'U' THEN UPDATE SET
    foo = COALESCE(GET(s.cdc_obj, 'foo'), t.foo),
    cat_date = COALESCE(GET(s.cdc_obj, 'cat_date'), t.cat_date),
    dogs = COALESCE(GET(s.cdc_obj, 'dogs'), t.dogs),
    elephants = GET(s.cdc_obj, 'elephants') -- 显式null直接覆盖原字段，无此字段则保留原值
WHEN NOT MATCHED AND s.operation = 'I' THEN INSERT (bar, foo, cat_date, dogs, elephants)
    VALUES (s.bar, GET(s.cdc_obj, 'foo'), GET(s.cdc_obj, 'cat_date'), GET(s.cdc_obj, 'dogs'), GET(s.cdc_obj, 'elephants'));

核心逻辑：

• 无更新字段时GET返回null，COALESCE自动保留目标表原值；
• 显式传入null的字段直接赋值，实现空值与非空值的互转处理。

2. 基于字段存在性的条件赋值更新 #

通过OBJECT_KEYS提取更新记录的字段列表，精准判断哪些字段需要更新，避免无意义的赋值操作：

MERGE INTO db.schema.target t
USING (
    SELECT 
        bar,
        operation,
        foo,
        cat_date,
        dogs,
        elephants,
        OBJECT_KEYS(PARSE_JSON(raw_json_col)) AS cdc_keys
    FROM src_db.src_schema.src_tbl
    WHERE operation IN ('I', 'U')
    ORDER BY position
) s
ON t.bar = s.bar
WHEN MATCHED AND s.operation = 'U' THEN UPDATE SET
    foo = CASE WHEN 'foo' IN (SELECT VALUE FROM TABLE(FLATTEN(input => s.cdc_keys))) THEN s.foo ELSE t.foo END,
    cat_date = CASE WHEN 'cat_date' IN (SELECT VALUE FROM TABLE(FLATTEN(input => s.cdc_keys))) THEN s.cat_date ELSE t.cat_date END,
    dogs = CASE WHEN 'dogs' IN (SELECT VALUE FROM TABLE(FLATTEN(input => s.cdc_keys))) THEN s.dogs ELSE t.dogs END,
    elephants = CASE WHEN 'elephants' IN (SELECT VALUE FROM TABLE(FLATTEN(input => s.cdc_keys))) THEN s.elephants ELSE t.elephants END
WHEN NOT MATCHED AND s.operation = 'I' THEN INSERT (bar, foo, cat_date, dogs, elephants)
    VALUES (s.bar, s.foo, s.cat_date, s.dogs, s.elephants);

核心逻辑：

• 用OBJECT_KEYS+FLATTEN解析出更新记录的所有字段名；
• 仅当字段存在于更新记录中时才执行赋值，严格区分“字段不存在”和“显式传null”的场景。

3. 结合Streams & Tasks实现自动化增量处理 #

如果CDC数据持续流入Snowflake，可通过流和任务实现全自动化的增量CDC处理，无需手动维护position顺序：

1. 创建流跟踪未处理的CDC数据：

CREATE OR REPLACE STREAM src_db.src_schema.cdc_stream
ON TABLE src_db.src_schema.src_tbl
APPEND_ONLY = TRUE; -- 适配CDC数据追加写入的特性

2. 创建定时任务自动执行MERGE：

CREATE OR REPLACE TASK src_db.src_schema.process_cdc_task
WAREHOUSE = your_warehouse_name
SCHEDULE = 'USING CRON 0 * * * * UTC' -- 可根据业务调整执行频率
WHEN SYSTEM$STREAM_HAS_DATA('src_db.src_schema.cdc_stream')
AS
MERGE INTO db.schema.target t
USING (
    SELECT 
        bar,
        operation,
        OBJECT_CONSTRUCT(*) AS cdc_obj
    FROM src_db.src_schema.cdc_stream
    ORDER BY position
) s
ON t.bar = s.bar
WHEN MATCHED AND s.operation = 'U' THEN UPDATE SET
    foo = COALESCE(GET(s.cdc_obj, 'foo'), t.foo),
    cat_date = COALESCE(GET(s.cdc_obj, 'cat_date'), t.cat_date),
    dogs = COALESCE(GET(s.cdc_obj, 'dogs'), t.dogs),
    elephants = GET(s.cdc_obj, 'elephants')
WHEN NOT MATCHED AND s.operation = 'I' THEN INSERT (bar, foo, cat_date, dogs, elephants)
    VALUES (s.bar, GET(s.cdc_obj, 'foo'), GET(s.cdc_obj, 'cat_date'), GET(s.cdc_obj, 'dogs'), GET(s.cdc_obj, 'elephants'));

核心逻辑：

• 流自动记录未处理的CDC数据，任务仅在流中有数据时触发；
• 全程自动化，无需人工介入排序和执行。

内容的提问来源于stack exchange，提问作者lpack