📚 课程目录第4章 Data Factory管道编排核心

4. Data Factory管道编排核心

在 Azure Data Factory（ADF）中，管道（Pipeline）是编排的最小逻辑单元，而真正让管道具备“批处理调度能力”的，是活动（Activity）、依赖关系（Dependency）、控制流（Control Flow）以及参数与变量这四类要素的组合。理解它们之间的关系，比记住某个活动的配置界面更重要：活动决定“做什么”，依赖关系决定“按什么顺序做”，控制流决定“按什么条件重复或分支”，参数与变量决定“每次运行时用什么值做”。

一、活动类型：数据移动、转换与外部动作

ADF 的活动大致可分为三类。第一类是数据移动活动，最典型的是 Copy Activity，它负责在源与接收器之间搬运数据，核心配置是 Source、Sink、Mapping 与并行度。第二类是数据转换活动，例如 Data Flow、Databricks Notebook、Stored Procedure、HDInsight 等，它们把计算下推到 Spark、SQL 或自定义代码中执行。第三类是控制与外部动作活动，包括 ForEach、If Condition、Until、Wait、Web、Lookup、Get Metadata、Set Variable、Append Variable、Execute Pipeline 等，它们本身通常不搬运数据，而是驱动编排逻辑。

一个常见的误区是把 Lookup 当成“取数活动”。Lookup 的定位是读取一份小结果集（默认上限 5000 行、2MB），用于给后续活动提供配置值或判断依据，例如读取一张控制表得到待处理日期列表，再交给 ForEach 逐项处理。Get Metadata 则用于获取文件或表的结构信息，如子项列表、是否存在、最后修改时间，常与 If Condition 搭配做“文件到达即处理”的判断。

二、依赖关系：成功、失败、完成与跳过

活动之间的连线称为依赖关系，它决定了执行顺序与触发条件。ADF 支持四种依赖条件：Succeeded（上游成功才执行）、Failed（上游失败才执行，常用于告警或补偿）、Completed（无论成功失败都执行，常用于清理）、Skipped（上游被跳过才执行）。默认是 Succeeded。

依赖关系可以形成链式、扇出（一个上游连多个下游）和汇聚（多个上游连一个下游）。汇聚时，下游活动会等待所有入边满足各自条件后才启动。需要特别注意的是：依赖关系只表达“先后与条件”，不表达数据传递。如果下游需要上游的结果，必须通过 Lookup 输出、Set Variable 或活动输出（@activity('X').output）显式引用。

三、控制流活动：ForEach、If Condition、Until

ForEach 用于对集合逐项执行内部活动。它的 items 属性接收一个数组表达式，例如 @activity('LookupControl').output.value 或 @variables('v_fileList')。默认并行度为 20，意味着最多同时处理 20 个迭代；若下游是数据库写入且需要控制并发，可勾选 Sequential 改为串行。ForEach 内部活动通过 @item() 引用当前元素，例如 @item().fileName。

If Condition 接收一个布尔表达式，例如 @greater(length(activity('GetMetadata').output.childItems), 0)，然后走 True 或 False 分支。它适合做“有文件才处理、无文件则跳过”的判断。注意 If Condition 的两个分支是互斥的，且分支内部可以继续嵌套 ForEach 或 Until。

Until 用于“重复执行直到条件成立”，类似 do-while。它有一个表达式（如 @equals(variables('v_status'),'Done')）和超时（timeout，默认 7 天）。Until 内部通常放 Wait、Lookup 或 Web 活动，用于轮询外部任务状态。必须设置合理的 timeout 和内部 Wait 间隔，否则可能长时间空转或永不退出。

四、参数与变量：运行期输入与管道内状态

参数（Parameter）是管道或数据集在运行期接收的输入，定义后不可在管道内修改，只能由触发器等外部传入。变量（Variable）是管道内部可读写的状态，通过 Set Variable 赋值、Append Variable 追加。两者的表达式引用方式不同：参数用 @pipeline().parameters.xxx，变量用 @variables('xxx')。

一个典型组合是：管道参数 p_windowStart、p_windowEnd 由调度触发器传入；Lookup 读取控制表得到待处理文件列表；Set Variable 把结果写入 v_fileList；ForEach 遍历该变量；每次迭代内用 Copy 搬运文件，并用 Append Variable 把行数追加到 v_rowCount；最后 If Condition 判断总行数是否大于 0，决定是否发送通知。

参数适合“每次运行都不同、由外部决定”的值，如日期窗口、环境名；变量适合“管道内部逐步累积”的值，如已处理文件数、累计行数。不要用变量去替代参数，否则触发器无法在启动时注入值。

五、一个可参考的编排骨架

下面给出一个批处理管道的参数与变量定义示例，可直接作为模板调整：

管道参数：
  p_windowStart   String   2024-06-01T00:00:00Z
  p_windowEnd     String   2024-06-02T00:00:00Z
  p_env           String   dev

管道变量：
  v_fileList      Array    []
  v_rowCount      Integer  0
  v_status        String   Pending

活动顺序：
  Lookup(控制表) → Set Variable(v_fileList)
  → ForEach(v_fileList)
       └ If Condition(@equals(item().type,'csv'))
            ├ True : Copy(源→接收器) → Append Variable(v_rowCount)
            └ False: Wait(1s)
  → If Condition(@greater(variables('v_rowCount'),0))
       ├ True : Web(发送成功通知)
       └ False: Web(发送空窗口告警)

调试时，优先用 Debug 运行并查看每个活动的 Output：Lookup 的 output.value 是否是预期数组、ForEach 的迭代次数是否与数组长度一致、If Condition 的表达式求值结果是否为 True。若 ForEach 报“items 不是数组”，通常是 Lookup 返回了对象而非数组，需要在表达式后加 .value 或 .arrayValue。若 Until 长时间不退出，检查条件表达式是否引用了会被内部活动更新的变量，以及 timeout 是否设置过短导致提前失败。

依赖关系中的 Failed 分支不会自动重试上游；重试策略（retry、retryIntervalInSeconds）是活动自身的属性，与依赖条件无关。设计补偿逻辑时，应把重试放在活动级，把告警与清理放在依赖级。

掌握这四类要素后，管道编排就从“拖控件”变成了可推理的结构：先确定活动类型与数据流向，再用依赖关系固定顺序与条件，用 ForEach/If/Until 表达重复与分支，最后用参数接收外部输入、用变量承载内部状态。这样设计出的管道，既能在调试时快速定位问题，也能在调度运行时保持可预测的行为。