编排与调度

这一课你将能做到

1. 说清楚编排器比 crontab 多解决的三件事：依赖管理、失败重试、历史回填
2. 用 Prefect 把「取数 → dvc repro → 门禁晋升」串成带重试的 flow，并注册成定时任务
3. 写一个等价的 Airflow DAG，理解 schedule、catchup、幂等参数 ds 的含义

为什么要学

到目前为止，你的管道都是手动 dvc repro 触发的。生产要求它每天凌晨自动跑：拉昨日判题日志、更新特征、重训、过门禁、晋升——中途任何一步失败要有重试和告警，停机一周后恢复还要能把欠的几天补跑回来。

crontab 只会「到点执行一条命令」，没有依赖、没有重试状态、没有补跑。编排器（Orchestrator）就是为此而生的工作流引擎。人话：把管道从「手动挡」换成「自动挡 + 倒车影像」，松油门（挂了）自动补一脚（重试），漏跑了还能倒回去补（backfill）。

先建立图像

每日重训流水线（DAG = 有向无环图）

  [02:00 触发]
      │
  ① fetch_logs(ds)      拉取 ds 当天的判题日志 → data/raw/2026-10-06.jsonl
      │ 失败：重试 2 次，间隔 5 分钟
  ② dvc_repro(ds)       重放数据管道 + 冒烟训练（幂等：按 ds 分区写）
      │
  ③ gate(ds)            指标门禁：达标 → challenger
      │
  ④ promote(ds)         人工/自动晋升 champion

ds = 逻辑日期（数据所属日），≠ 实际运行时间 → 幂等的关键

跟着做

步骤 1：用 Prefect 写第一条 flow

Prefect 是 Python 原生的现代编排器，上手最快：

pip install prefect

# flows/retrain.py
import subprocess
from prefect import flow, task

@task(retries=2, retry_delay_seconds=30, log_prints=True)
def fetch_logs(ds: str) -> str:
    """拉取 ds 当天判题日志（示例用本地脚本代替下载）。"""
    subprocess.run(["python", "src/make_dataset.py"], check=True)
    return f"data/raw/judge_logs.jsonl@{ds}"

@task(retries=1, retry_delay_seconds=60)
def rebuild_features(ds: str):
    """重放数据管道：dvc repro 保证依赖与增量正确。"""
    subprocess.run(["dvc", "repro"], check=True)

@task(retries=1)
def smoke_train_and_gate(ds: str):
    """冒烟训练 + 指标门禁，不达标直接抛异常让流程失败告警。"""
    subprocess.run(["python", "src/train.py", "--smoke"], check=True)
    import mlflow
    from mlflow.tracking import MlflowClient
    exp = mlflow.get_experiment_by_name("timeout-prediction")
    run = MlflowClient().search_runs([exp.experiment_id],
                                     order_by=["attributes.start_time DESC"])[0]
    if run.data.metrics.get("roc_auc", 0) < 0.6:
        raise ValueError(f"门禁不通过: {run.data.metrics}")

@flow(name="daily-retrain")
def daily_retrain(ds: str | None = None):
    from datetime import date, timedelta
    ds = ds or (date.today() - timedelta(days=1)).isoformat()
    path = fetch_logs(ds)
    rebuild_features(ds)
    smoke_train_and_gate(ds)
    print(f"{ds} 重训流水线完成，产物: {path}")

if __name__ == "__main__":
    daily_retrain()

python flows/retrain.py          # 立即手动跑通一次

步骤 2：接上 UI 与定时调度

prefect server start           # http://127.0.0.1:4200 看运行历史、日志、重试
# 另一个终端，注册为每日 02:00 的定时部署：
prefect deploy flows/retrain.py:daily_retrain \
  --name daily-0200 --cron "0 2 * * *"

UI 里能看到每次运行的状态树、每个 task 的重试记录。补跑历史某天：python flows/retrain.py "2026-10-01"——因为 ds 是参数，直接重放。

步骤 3：等价的 Airflow DAG

Airflow 是编排界的老牌标准，语法不同、概念相通：

# dags/retrain_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="daily_retrain",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,                     # True 则停机期间的日期会依次补跑
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["mlops"],
) as dag:
    fetch = BashOperator(
        task_id="fetch_logs",
        bash_command="python /opt/airflow/src/make_dataset.py --date {{ ds }}",
    )
    repro = BashOperator(
        task_id="dvc_repro",
        bash_command="cd /opt/airflow && dvc repro",
    )
    gate = BashOperator(
        task_id="gate",
        bash_command="python /opt/airflow/src/train.py --smoke",
    )
    fetch >> repro >> gate            # 用 >> 声明依赖

{{ ds }} 是 Airflow 模板变量，注入逻辑日期（如 2026-10-06）——任务因此只依赖日期参数，重跑任意历史日期结果一致，这就是幂等。本地体验：

pip install "apache-airflow==2.10.5" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.5/constraints-3.11.txt"
airflow standalone          # 一键起 webserver+scheduler，UI 在 8080
airflow dags test daily_retrain 2026-10-06   # 不等调度，单测式跑一天

讲清楚要点

1. 编排器的三大增值：依赖管理（DAG 声明先后）、失败恢复（retries + 告警 + 部分重跑）、历史回填（backfill/catchup）。crontab 三者全无。
2. 幂等靠「逻辑日期」实现：所有任务只读写 data/.../{ds}/ 这样的日期分区，不用当前系统时间。同一个 ds 重跑一百次，结果一致、不产生双份数据。
3. 任务粒度 = 失败爆炸半径：把「取数/特征/训练/门禁」拆成独立 task，失败时只重跑坏的那环。全塞一个 task 里，网络抖动一次就要全量重来。
4. 编排 ≠ CI：CI 由代码提交触发、管质量门禁；编排由时间/事件触发、管生产节奏。两者都会跑训练脚本，但职责不同（第 7 课对比）。
5. 选型直觉：Python 团队、快速起步、云原生 → Prefect/Dagster；重调度、多团队共享、生态成熟 → Airflow；K8s 深度用户 → Argo Workflows；Kubeflow Pipelines 适合已all-in K8s 的 ML 平台。核心思想一致，学会一个迁移不难。

你可能会懵的地方

• 为什么说任务不幂等是编排的头号反模式？ 重试和补跑都建立在「重跑安全」之上。如果 fetch 任务每次追加写同一张表，重试一次数据就翻倍；改成按 ds 分区覆盖写（df.to_parquet(f"...{ds}.parquet")），重试才变得无害。
• catchup=True 是什么行为？什么时候开？ Airflow 默认会把 start_date 到当前之间的所有调度周期补跑一遍。历史数据初始化时开，日常业务关掉（False），避免停机一周后突然涌入 7 个补跑任务打爆资源。
• 重试会不会把上游重复执行？ 不会。重试只作用于失败的 task 本身；上游成功过的 task 直接复用状态。想要「从某个节点重新开始」用 clear/重跑子图功能。
• 单个 task 要处理的数据太大，内存爆了怎么办？ 三条路：task 内部分块流式处理；把重活下沉到专门引擎（Spark/SQL），编排器只提交作业并轮询状态；用动态映射（Prefect .map、Airflow dynamic task mapping）按分区并行。
• Airflow scheduler 和 worker 什么关系？ scheduler 解析 DAG、到点把 task 实例排进队列；worker（LocalExecutor 单机进程/Celery/K8s Executor 每 Pod 一个）领取并执行。DAG 文件只描述「做什么」，执行环境在 worker，所以依赖要装进 worker 镜像。

错法对照

错法	问题	正解
task 里用 datetime.now() 取日期	重跑/补跑结果不一致	用调度器传入的 ds 逻辑日期
整条流水线写进一个 task	任何失败全量重来	按「取数/特征/训练/门禁」拆 task
BashOperator 命令里硬编码绝对路径	换环境即崩	用环境变量/配置注入路径
失败无告警，靠用户报障发现	故障平均发现时间以天计	task 失败回调接钉钉/飞书/Slack
每天全量重算所有历史特征	成本随历史线性增长	按 ds 增量分区 + 定期月度全量校准

卡点急救

1. Prefect deploy 找不到 flow：确认 @flow 装饰的函数在该文件顶层可导入；命令里 文件路径:函数名 的冒号格式正确。
2. Airflow DAG 在 UI 里不出现：airflow dags list-import-errors 看解析报错，多半是 import 或缩进问题；DAG 文件须放在 AIRFLOW_HOME/dags/。
3. dvc repro 在编排器里找不到数据：编排 worker 的工作目录不是项目根目录，BashOperator 里先 cd 到项目路径，或给任务设 cwd。

面试常问

• 设计一个「每日自动重训」流水线，你会怎么拆任务？ 拉数与校验 → 特征重放（dvc repro）→ 训练（记录 MLflow run）→ 自动评估门禁 → 挂 challenger → 通知人工审核或自动晋升。每个 task 幂等、按 ds 分区、带重试与告警；门禁失败要让流程显式失败，绝不静默上线。
• 如何保证调度任务幂等？ 输入输出全部按逻辑日期分区；写操作用覆盖而非追加；副作用（如调外部 API）带去重键；先在小数据上验证「同 ds 重跑两次结果一致」再上生产。
• Airflow 和 Prefect 怎么选？ Airflow：调度语义成熟、社区与集成最全、适合多团队中心化平台，但 Python 动态工作流写起来啰嗦；Prefect：Python 原生、开发体验好、动态并行轻量，适合中小团队快速起步。决策点在团队规模、运维能力和现有基础设施。

自测三问

1. 编排器比 crontab 多解决哪三件事？
2. 「逻辑日期 ds」和「系统当前时间」的区别为什么是幂等的关键？
3. catchup=True 什么时候该开、什么时候必须关？

小结一句话

编排器把管道变成「按 DAG 依赖自动执行、失败重试、按逻辑日期幂等可回填」的生产流程，Prefect 适合快速起步，Airflow 适合规模化调度。

延伸

• Prefect 官方文档——flow/task/deploy 与重试语义的权威说明。
• Airflow 官方文档：DAG 编写最佳实践——幂等、参数化、任务粒度的官方建议。
• 下一课：CI/CD for ML——代码提交自动触发校验、冒烟训练与镜像发布。