编排与调度

这一课你将能做到

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()