如何通过REST API将ADF管道运行数据同步至Snowflake

整体实现思路 #

目前有两种成熟落地路径可选，优先推荐用ADF内置诊断日志+Copy Activity的方案，代码量低、运维成本低；需要近实时同步可以选Azure Function事件触发方案。

方案1：ADF内置诊断日志同步方案（推荐） #

• 步骤1：开启ADF诊断日志导出到Azure存储

操作路径：进入你的ADF实例→左侧菜单「监控」板块→选择「诊断设置」→点击「添加诊断设置」，勾选Pipeline runs日志类别，导出目标选择你提前准备好的Azure Blob存储账户，日志存储格式选择JSON。该日志默认包含你需要的所有核心字段：管道名称（对应字段pipelineName）、运行时长（durationMs，单位毫秒）、启动时间（start，UTC时间）、运行状态、错误信息（error/message字段）。

• 步骤2：在Snowflake中创建日志接收表
  执行以下SQL建表，可根据你的实际字段需求调整：

CREATE TABLE ADF_PIPELINE_RUN_MONITOR (
    PIPELINE_NAME VARCHAR(255) NOT NULL COMMENT '管道名称',
    RUN_ID VARCHAR(36) NOT NULL COMMENT '管道运行唯一ID',
    START_TIME TIMESTAMP_NTZ COMMENT '启动时间',
    END_TIME TIMESTAMP_NTZ COMMENT '结束时间',
    DURATION_MS NUMBER COMMENT '运行时长/毫秒',
    RUN_STATUS VARCHAR(50) COMMENT '运行状态：成功/失败/取消',
    ERROR_DETAIL VARCHAR COMMENT '错误详情',
    LOAD_TIME TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP() COMMENT '数据入库时间'
);

• 步骤3：在ADF中配置同步管道
  • 新建ADF同步管道，按你的监控需求配置定时触发频率（推荐15分钟/1小时触发一次）
  • 添加Copy Activity，源端配置为存储ADF日志的Blob Storage，文件格式选择JSON，配置文件过滤规则只同步上次同步后新增的日志文件
  • 接收端配置Snowflake连接器，连接到你的Snowflake实例，目标表选择上一步创建的ADF_PIPELINE_RUN_MONITOR，按字段含义完成JSON字段到Snowflake表字段的映射
• 步骤4：配置增量同步避免重复数据
  你可以在Snowflake中新增一张同步进度表，每次同步前先查询表中记录的上次同步最大结束时间，拉取大于该时间的日志进行同步，同步完成后更新进度表中的时间戳即可实现无重复增量同步。

方案2：事件触发实时同步方案 #

如果需要秒级/分钟级的近实时同步，可以用该方案：

• 配置ADF运行事件上报到Azure Event Grid
• 编写Azure Function触发器，监听Event Grid推送的ADF管道运行结束事件，解析事件中的字段后直接调用Snowflake连接器写入目标表

该方案可以复用你之前对接ADF REST API导出数据到Power BI时的字段解析逻辑，减少适配成本。

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