如何使用Matillion ETL实现Azure Blob Storage到Snowflake的增量加载

以下是三种经过验证的落地实现方案，你可以根据自己的业务场景选择： #

方案一：基于增量识别字段的自定义Upsert逻辑（适配你当前使用的Table Update组件） #

该方案无需额外组件，复用你现有配置即可实现，操作步骤如下：

• 提前为源数据和Snowflake目标表约定统一的增量识别字段，优先选择数据更新时间戳update_time、自增主键ID、ETL批次号这类可排序、无重复的字段，用于区分增量数据范围。
• 在Matillion作业中新建一个作业变量，命名为last_sync_max_value，首次运行时赋值为'1970-01-01'（时间戳场景）或者0（自增ID场景）作为同步起始值，该变量后续会存储每次同步完成后的最大增量字段值，持久化保存在Matillion内置元数据表或专门的Snowflake配置表中，下次作业启动时自动读取。
• 调用Azure Blob Storage读取组件拉取源文件，过滤出增量识别字段值大于last_sync_max_value的数据集，结构化文件可直接在组件过滤配置中写条件，非结构化文件可后续接入Calculator组件做字段处理后再过滤。
• 对接你正在使用的Table Update组件，配置参数参考如下，组件配置界面见下图：
  

配置说明：

• 连接选择已完成鉴权的Snowflake连接器
• 目标表选择要写入的Snowflake正式表
• Update Method选择Upsert (Update if exists, Insert if not)
• Match Keys勾选表的主键字段作为匹配依据
• 字段映射部分确认源字段和目标字段的对应关系准确

• 同步完成后接入SQL查询组件，从Snowflake目标表中查询当前最大的增量识别字段值，将结果赋值给last_sync_max_value变量并持久化存储，作为下一次同步的起始判断值。

方案二：基于文件变动的增量同步（适合Blob文件按规则命名的场景） #

如果你的Azure Blob中的文件是按日期/批次规则命名、不会修改历史文件，该方案逻辑更简单，性能更高：

• 调用Azure Blob Storage List组件，根据文件最后修改时间、或者文件名中的日期/批次后缀，过滤出上次同步之后新增的所有文件。
• 接入Loop组件循环处理每个新增文件，读取文件内容后写入Snowflake的临时 staging 表。
• 接入SQL组件执行Snowflake原生MERGE语句，将staging表数据合并到正式目标表，参考语法如下：

MERGE INTO 正式目标表 t
USING 临时staging表 s
ON t.主键字段 = s.主键字段
WHEN MATCHED THEN UPDATE SET 
  t.字段1 = s.字段1,
  t.字段2 = s.字段2,
  t.update_time = s.update_time
WHEN NOT MATCHED THEN INSERT (主键字段, 字段1, 字段2, update_time)
VALUES (s.主键字段, s.字段1, s.字段2, s.update_time);

• 所有文件处理完成后，记录本次处理的最新文件修改时间或最大批次号，持久化存储作为下次同步的过滤起点。

方案三：基于CDC日志的增量同步（适合有全量数据变更记录的场景） #

如果你的源系统已经生成了CDC（变更数据捕获）日志并存储到Azure Blob中，可以直接读取CDC日志文件，根据日志中标记的操作类型（插入/更新/删除），对应执行Snowflake表的增删改操作，该方案可以实现最精准的增量同步，无需额外做数据冲突判断。 #

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