如何获取Azure Data Factory的DataFlow流水线元数据并构建性能测试调试流水线

1. 提前创建SQL目标表 #

首先在你的目标SQL库中创建存储监控数据的表，参考建表语句：

CREATE TABLE ADF_DataFlow_Perf_Monitor (
    Sr_No INT IDENTITY(1,1) PRIMARY KEY,
    Pipeline_ID NVARCHAR(100),
    Dataflow_Start_Time NVARCHAR(20),
    Dataflow_End_Time NVARCHAR(20),
    Compute_Acquire_Duration INT,
    Data_Input_Bytes BIGINT,
    Data_Output_Bytes BIGINT,
    Run_Type NVARCHAR(50),
    DataFlow_Status NVARCHAR(20)
)

2. 流水线基础框架搭建 #

整个流水线仅包含3个核心组件，所有指标都从ADF自带的系统变量和活动输出中获取，无需调用额外API：

• 第一步：添加一个DataFlow活动，命名为Test_DataFlow，暂时可以不绑定具体的DataFlow，后续测试时直接替换即可
• 第二步：添加一个Script活动，连接到你提前创建的SQL目标表，设置活动依赖项为「完成」（即无论前面的DataFlow成功/失败都执行）

3. 各监控字段的获取方式 #

所有字段值都可以通过ADF内置表达式直接获取，直接填入Script活动的INSERT语句参数即可：

• pipeline ID：用表达式 @pipeline().RunId，系统内置变量直接返回当前流水线运行ID
• Dataflow Start Time：用表达式 @formatDateTime(activity('Test_DataFlow').output.runStatus.startTime, 'ddMMyy HH:mm')，注意原需求中的HH:SS为笔误，实际时间格式用分mm而非秒SS
• Dataflow end time：用表达式 @formatDateTime(activity('Test_DataFlow').output.runStatus.endTime, 'ddMMyy HH:mm')，规则同上
• Duration to acquire compute：用表达式 @div(activity('Test_DataFlow').output.runStatus.computeAcquisitionDuration, 1000)，原始返回值为毫秒，除以1000得到秒级数值
• Data input：用表达式 @sum(activity('Test_DataFlow').output.runStatus.metrics.sources.*.totalBytesRead)，自动统计所有源读取的总字节数，如需转MB可再除以1048576
• Data output：用表达式 @sum(activity('Test_DataFlow').output.runStatus.metrics.sinks.*.totalBytesWritten)，自动统计所有Sink写入的总字节数
• Run type：用表达式 @pipeline().TriggerType，系统变量直接返回运行类型，调试运行返回Manual，触发器运行返回Triggered
• DataFlow Status：用表达式 @activity('Test_DataFlow').status，直接返回活动运行状态，成功返回Succeeded，失败返回Failed，可按需求替换显示文本

4. 即插即用使用方法 #

后续需要测试任意DataFlow时，仅需两步操作：

1. 点击流水线中的Test_DataFlow活动，选择你要测试的目标DataFlow，配置好DataFlow的源、Sink连接信息
2. 直接运行流水线即可，所有指标会自动写入SQL表，无需修改其他组件配置

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