Airflow实现自动重训的关键是将训练逻辑封装为可重复执行、状态可控、失败可追溯的独立任务单元。需确保训练脚本幂等、路径可配置、输出带时间戳、支持PythonOperator调用；触发依赖数据就绪而非固定时间；上线前必须验证指标并设人工卡点。
Airflow 本身不负责模型训练，它只调度任务；想让模型自动重训，关键不是“用 Airflow 写个 DAG 就完事”，而是把训练逻辑封装成可重复执行、状态可控、失败可追溯的独立任务单元。
训练脚本必须能单独运行且幂等
很多人写完 train.py 直接扔进 Airflow，结果调度一次就报错：路径不对、数据没更新、GPU 占满、上次训练残留锁文件……根本原因在于脚本没脱离“本地调试态”。
- 所有路径用
os.path.join(os.path.dirname(__file__), "data")或配置文件读取，禁止硬编码"C:/project/data" - 训练前加检查：用
os.path.exists("models/latest.pth")判断是否已存在有效模型，有则跳过或强制覆盖（由参数控制） - 输出必须带时间戳：模型保存为
f"model_{int(time.time())}.pth"，日志写入logs/train_$(date +%Y%m%d_%H%M%S).log - 入口加
if __name__ == "__main__":，确保可被 Airflow 的PythonOperator直接调用
Airflow DAG 中别用 BashOperator 调 Python 脚本
用 BashOperator 执行 python train.py 看似简单，但会丢失异常堆栈、无法传参、环境隔离差——尤其当 Airflow worker 和 webserver Python 版本/包版本不一致时，ImportError 频发。
- 改用
PythonOperator，直接导入函数：task = PythonOperator(task_id="train", python_callable=train_main, op_kwargs={"epochs": 50}) - 确保
train_main函数返回明确状态（如return {"status": "success", "model_path": path}），后续任务可依赖该返回值 - 若训练必须用 conda 环境或特定 CUDA 版本，别在 DAG 里切环境，改用
KubernetesPodOperator或自定义 Docker 镜像启动隔离容器
重训触发逻辑不能只靠 schedule_interval
定时 ≠ 自动重训。每月 1 号跑一次，但数据源上周就断更了，模型还是拿脏数据训；或者数据更新了三次，你却只训了一次——这都不是“自动重训”，只是“自动跑脚本”。
python-script-generator
下载
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
立即学习“Python免费学习笔记（深入）”；
- 用
ExternalTaskSensor监听上游 ETL DAG 的完成信号，比如等etl_daily_data任务成功后再触发训练 - 用
FileSensor检查 HDFS/S3 上是否有新数据文件：filepath="s3://my-bucket/data/{{ ds }}/features.parquet" - 关键数据表加元数据校验：在训练任务开头查
SELECT MAX(updated_at) FROM raw_events，若距当前超 24h 则直接raise AirflowSkipException
模型上线前必须加验证和人工卡点
自动训练最危险的环节是自动上线。一个在验证集上 AUC 下降 0.05 的模型，如果直接覆盖线上服务，故障就是秒级的。
- 训练后必须跑评估脚本：
evaluate_model(model_path, test_dataset)，指标低于阈值则任务失败，不往下走 - 用
TriggerDagRunOperator启动部署 DAG，但部署 DAG 开头加ShortCircuitOperator+ 人工审批 Task（例如 SlackOperator 发消息并等待 webhook 回调） - 保留最近 3 个模型快照，用软链接
models/current -> models/v20240512控制线上加载路径，回滚只需改链接
真正难的从来不是“怎么让 Airflow 每天跑一次”，而是“怎么确认这次训练值得上线”。数据质量、特征一致性、评估偏差、服务兼容性——这些没法靠 schedule_interval 解决，得在 DAG 的每个环节埋钩子、设断点、留证据。
