📚 课程目录第4章 Data Factory进阶编排

4. Data Factory进阶编排

当数据工厂的管道从“单次手工触发”走向“每天定时、批量、可重跑”的生产级调度时，编排能力就成为决定可维护性的关键。本章围绕参数化、变量与表达式、ForEach/If 控制流、错误处理与重试、依赖关系以及增量加载模式，给出一套可直接落地的进阶编排方法。

一、参数化管道：把“写死”变成“可传入”

参数（Parameter）是管道对外暴露的输入接口，在管道运行前由调用方（手动触发、调度触发器、父管道 Execute Pipeline 活动）赋值，运行期间只读。典型参数包括 windowStart、windowEnd、sourceContainer、tableList。与之相对，变量（Variable）在管道内部通过“设置变量”活动读写，用于累积状态，例如循环中拼接已处理表名、记录失败计数。

一个常见的误区是把所有东西都做成参数。判断标准很简单：由外部决定的值用参数，由管道内部计算或累积的值用变量。例如增量加载的起止时间由调度器传入，属于参数；而“本次共处理了多少张表”属于变量。

二、表达式：动态内容的计算引擎

表达式语言用于在字段中生成动态值，常用函数包括 @pipeline().parameters.windowStart、@formatDateTime(...,'yyyy-MM-dd')、@addDays(utcNow(),-1)、@concat()、@if(equals(...),a,b)、@activity('Lookup').output.firstRow。表达式在“动态内容”编辑器中编写，注意字符串拼接时函数返回值需用引号包裹，例如 @concat('raw/', pipeline().parameters.folder, '/data.csv')。

下面这张图展示了参数、变量与表达式在一次增量加载管道中的流转关系：参数从触发器进入，表达式在活动字段中被求值，变量在循环中累积，最终写入目标并回传结果。

三、ForEach 与 If：控制流的两把钥匙

ForEach 活动接收一个数组，对每个元素并行或串行执行内部活动。关键参数是“顺序”开关：勾选顺序执行可避免并发写入冲突，但吞吐下降；不勾选则默认并行（最多 50 并发，可调）。内部活动通过 @item() 引用当前元素，例如 @item().tableName。

If 条件活动根据布尔表达式选择“真”或“假”分支。常见用法是判断 Lookup 返回的行数：@greater(activity('LookupWatermark').output.count, 0)，为真则执行 Copy 并更新水位，为假则跳过并记录日志。注意 If 的两个分支都必须有活动，否则设计器会报错。

四、错误处理与重试：让失败可控

每个活动都有“重试”策略：重试次数（默认 0，建议 2~3 次）与重试间隔（秒）。对于网络抖动、临时限流，重试能显著降低误报。但重试不是万能药——数据格式错误、权限不足重试多少次都会失败，此时应依赖“失败时”路径。

在管道级别，可通过“依赖条件”把活动连成成功、失败、完成、跳过四种关系。典型模式是：Copy 活动失败时，连接一个“记录错误”的存储过程或 Web 活动，把表名、错误信息写入日志表，再让管道以失败状态结束，便于告警。若希望部分失败不阻断整体，可把失败分支设为“完成”依赖，让后续汇总活动继续执行。

重试间隔建议设置为 30 秒以上，避免对下游系统造成瞬时压力；对幂等写入（如按分区覆盖）可放心重试，对追加写入需先确认是否会产生重复数据。

五、依赖关系：编排的骨架

依赖关系决定活动执行顺序。除了默认的“成功”依赖，还有“失败”“完成”“跳过”。一个实用技巧是：用“完成”依赖做清理活动，无论上游成功与否都执行；用“跳过”依赖处理条件分支中未走到的路径。依赖关系画错会导致活动不执行或死锁，建议在复杂管道中先用小规模数据验证依赖拓扑。

六、增量加载模式实战

增量加载的核心是“水位线”（watermark）：记录上次成功加载的时间点或自增 ID，下次只取大于该值的数据。下面给出一个可参考的配置样例，假设源为 Azure SQL，目标为 ADLS Parquet，按天分区。

管道参数：
  windowStart  = @trigger().scheduledTime
  windowEnd    = @addDays(trigger().scheduledTime, 1)
  tableList    = ["orders","customers","products"]

活动 1  LookupWatermark（查询控制表）
  SELECT lastLoadTime FROM etl_watermark WHERE tableName = 'orders'

活动 2  ForEach（items = @pipeline().parameters.tableList，顺序执行）
  内部：
    CopyData
      source.query = @concat(
        'SELECT * FROM ', item().tableName,
        ' WHERE modifiedAt > ''', activity('LookupWatermark').output.firstRow.lastLoadTime, '''',
        ' AND modifiedAt <= ''', formatDateTime(pipeline().parameters.windowEnd,'yyyy-MM-dd HH:mm:ss'), '''')
      sink.folder  = @concat('raw/', item().tableName, '/', formatDateTime(pipeline().parameters.windowStart,'yyyy-MM-dd'))
    SetVariable processedCount = @add(int(variables('processedCount')), 1)

活动 3  If（条件：@greater(activity('CopyData').output.rowsWritten, 0)）
  真分支：UpdateWatermark
    UPDATE etl_watermark SET lastLoadTime = @pipeline().parameters.windowEnd
    WHERE tableName = @item().tableName
  假分支：LogNoChange（写入日志表，标记无新数据）

运行后如何判断结果？三个观察点：一是 Copy 活动的 rowsWritten 是否大于 0；二是控制表 etl_watermark 的 lastLoadTime 是否推进到本次窗口结束时间；三是 ADLS 对应日期分区下是否生成了 Parquet 文件。若 rowsWritten 为 0 且水位未推进，说明该窗口确实无新数据，属于正常；若 rowsWritten 大于 0 但水位未更新，则要检查 If 条件或 UpdateWatermark 是否被跳过。

增量加载必须保证“先写数据、后更新水位”。若顺序颠倒，一旦写入失败而水位已推进，下次将永久丢失这段数据。建议把 UpdateWatermark 放在 Copy 成功依赖之后，并开启重试。

七、编排检查清单

检查项	判断标准
参数与变量职责	外部输入用参数，内部累积用变量，无混用
表达式可读性	复杂拼接拆分为多个 SetVariable，避免单行超长
ForEach 并发	写同一目标时勾选顺序执行，避免冲突
重试策略	临时故障类活动重试 2~3 次，间隔 ≥30 秒
失败路径	关键活动有失败分支，错误写入日志表
水位更新顺序	数据写入成功后才更新水位

掌握以上六项能力后，管道就从“能跑”升级为“可调度、可重跑、可观测”。下一阶段可以进一步引入 Mapping Data Flow 做转换，或使用托管 Airflow 做跨系统编排，但参数化、控制流、错误处理与增量水位这四块，始终是数据工厂编排的地基。