从零构建MLOps工具链：基于MLflow和Airflow的CI/CD实战指南

共计 2198 个字符，预计需要花费 6 分钟才能阅读完成。

背景痛点：为什么我们需要 MLOps

最近在团队里推进一个图像分类项目时，遇到了这些典型问题：

• 同事 A 修改了超参数但忘记记录，三天后无法复现当时的模型效果
• 测试环境用 TensorFlow 2.4 训练，生产环境却是 2.6 版本导致预测结果不一致
• 运维同学手动部署模型时，误将开发中的 v0.2 版本当成线上 v0.1 版本发布

这些场景让我意识到，机器学习项目需要像传统软件开发那样建立完整的 CI/CD 流程。经过两个月的实践，我们最终用 MLflow+Airflow 构建了解决方案。

工具选型：MLflow 和 Airflow 为什么胜出

实验跟踪工具对比

• MLflow vs TensorBoard
• TensorBoard：擅长训练过程可视化，但缺少参数记录和模型打包功能
• MLflow：提供完整的实验元数据管理（参数、指标、样本），还能打包 conda 环境
• 关键差异：MLflow 的模型注册表 (Registry) 支持阶段变更(Staging/Production)

工作流工具对比

• Airflow vs Luigi
• Luigi：配置简单但调度能力弱，缺少任务优先级机制
• Airflow：支持定时调度、任务依赖图可视化，社区插件丰富
• 决胜点：Airflow 的 PythonOperator 能直接调用 MLflow 的 API

架构设计：组件如何协同工作

工具链数据流向示意图

核心路径设计

1. 训练阶段
2. 实验代码通过 mlflow.log_param() 记录超参数

3. 模型保存路径遵循：s3://models/<project>/<experiment_id>/<run_id>/artifacts

4. 调度阶段

   # Airflow DAG 示例
   from airflow.operators.python import PythonOperator
   
   def train_model():
       try:
           import mlflow
           mlflow.set_tracking_uri("http://mlflow-server:5000")
           with mlflow.start_run():
               # ... 训练逻辑...
               mlflow.log_metric("accuracy", 0.92)
               mlflow.sklearn.log_model(model, "model")
       except Exception as e:
           # 邮件告警配置
           send_alert_email(str(e))
           raise
   
   train_task = PythonOperator(
       task_id='train',
       python_callable=train_model,
       dag=dag
   )

5. 部署触发

6. 当模型在 MLflow 注册表标记为 Production 时
7. Airflow 通过 REST API 检测到状态变更
8. 自动触发 Kubernetes Deployment 更新

生产环境配置要点

MLflow 持久化部署

# docker-compose.yml 片段
version: '3'
services:
  mlflow:
    image: mlflow/mlflow:1.23.1
    environment:
      - MLFLOW_S3_ENDPOINT_URL=http://minio:9000
      - AWS_ACCESS_KEY_ID=${MINIO_ACCESS_KEY}  # 从.env 文件注入
    volumes:
      - ./mlflow.db:/mlflow.db  # SQLite 持久化(生产建议用 PostgreSQL)
    ports:
      - "5000:5000"

Airflow 性能调优

# airflow.cfg 关键参数
[celery]
worker_concurrency = 8  # 根据 CPU 核心数调整
worker_prefetch_multiplier = 2  # 避免任务堆积

[scheduler]
max_threads = 4  # 控制调度器资源占用

模型监控埋点

# FastAPI 模型服务示例
from prometheus_client import Counter

PREDICTION_COUNTER = Counter(
    'model_predictions_total', 
    'Total prediction requests',
    ['model_version']
)

@app.post("/predict")
async def predict(data: InputSchema):
    PREDICTION_COUNTER.labels(model_version=MODEL_VERSION).inc()
    # ... 预测逻辑...

避坑经验：我们踩过的三个大坑

1. CUDA 版本冲突
2. 现象：训练用 CUDA 11.0 但生产环境是 10.2

3. 解决：在 MLflow 的 conda.yaml 中显式指定cudatoolkit=11.0

4. Airflow 任务堆积

5. 现象：高优先级任务被阻塞

6. 方案：设置任务优先级权重

   train_task = PythonOperator(
       task_id='train',
       priority_weight=3,  # 普通任务默认为 1
       ...
   )

7. 分布式日志丢失

8. 现象：多 GPU 训练时部分 worker 日志未记录
9. 解决：使用 mlflow.start_run(nested=True) 创建子运行

挑战与进阶

当前架构在模型 A / B 测试时还需要人工干预。下一步计划集成 Kubeflow Pipeline 来实现：

• 自动流量分配
• 多版本模型性能对比
• 基于指标的自动回滚

建议读者尝试用 KFP 的 Condition 组件实现自动化决策流程。完整的示例代码已放在GitHub 仓库。