MLOps工具链实战：基于MLFlow和Airflow构建可复现的机器学习CI/CD流水线

1次阅读

没有评论

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

在机器学习项目的全生命周期中，工程师们常常面临几个关键痛点：

实验记录混乱：手动记录超参数、代码版本和评估指标容易出错，导致实验结果无法复现。例如，团队成员可能忘记记录某次实验的随机种子，导致后续无法复现 ” 神奇 ” 的模型效果。
环境差异问题：开发环境和生产环境的软件版本、硬件配置差异，经常导致 ” 在我机器上能跑 ” 的情况。这种差异可能源自 Python 包版本、CUDA 驱动等细节。
部署流程断裂：从实验到生产的过渡往往依赖手工操作，包括模型导出、API 封装和服务器部署。这种人工干预不仅效率低下，还容易引入错误。

MLFlow：提供完整的实验生命周期管理，包括参数记录、指标追踪、artifact 存储和模型注册。其优势在于轻量级设计（只需pip install mlflow）和与任意 ML 框架的兼容性。
TensorBoard：虽然可视化效果优秀，但主要针对 TensorFlow 生态，且缺乏原生模型版本控制功能。

Airflow：基于 Python 的 DAG 定义方式，社区插件丰富（如 KubernetesOperator），适合需要灵活定制的场景。其 Web UI 和任务重试机制特别适合运维人员。
Kubeflow Pipelines：原生支持 Kubernetes，但架构较重，学习曲线陡峭，更适合已有 K8s 基础设施的团队。

组合优势：MLFlow+Airflow 形成了从实验到生产的完整闭环，两者都采用模块化设计，可以逐步引入现有工作流。例如，可以先单独使用 MLFlow Tracking，再逐步接入 Airflow 编排。

import mlflow
from sklearn.ensemble import RandomForestClassifier

try:
    mlflow.set_experiment("Fraud_Detection")
    with mlflow.start_run():
        # 记录参数和指标
        mlflow.log_param("n_estimators", 100)
        mlflow.log_param("max_depth", 8)

        model = RandomForestClassifier()
        model.fit(X_train, y_train)
        accuracy = model.score(X_test, y_test)

        mlflow.log_metric("accuracy", accuracy)
        # 保存模型（自动记录框架版本）mlflow.sklearn.log_model(model, "model")
except Exception as e:
    print(f"Tracking failed: {str(e)}")
    # 这里可以添加告警逻辑

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def train_with_mlflow():
    # 这里调用上述 MLFlow 代码
    pass

# 定义 DAG（每天凌晨运行）dag = DAG(
    'model_retraining',
    schedule_interval='0 0 * * *',
    default_args={
        'retries': 3,  # 失败自动重试
        'retry_delay': timedelta(minutes=5)
    }
)

train_task = PythonOperator(
    task_id='train_model',
    python_callable=train_with_mlflow,
    dag=dag
)

FROM python:3.8-slim

# 安装依赖（固定版本确保一致性）RUN pip install mlflow==1.20.2 \
    && pip install scikit-learn==0.24.2

# 暴露 MLFlow 模型服务端口
EXPOSE 5000

# 启动命令（加载特定版本的模型）CMD ["mlflow", "models", "serve", 
     "-m", "runs:/<RUN_ID>/model",
     "-p", "5000", "--no-conda"]

每次模型更新生成新的 MLFlow Run ID
通过模型注册表（Model Registry）管理 Staging/Production 阶段
采用蓝绿部署切换流量

配置 MLFlow 使用 S3/MinIO 存储 artifacts：

mlflow.set_tracking_uri("http://mlflow-server:5000")
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "minio123"
os.environ["MLFLOW_S3_ENDPOINT_URL"] = "http://minio:9000"