ml-pipeline-automation

ML 流水线自动化

使用经过生产环境验证的 Airflow、Kubeflow 和 MLflow 模式，编排从数据摄取到生产部署的端到端机器学习工作流。

何时使用此技能

加载此技能的场景：

• 构建 ML 流水线：编排数据 → 训练 → 部署工作流
• 安排重新训练：设置自动模型重新训练计划
• 实验跟踪：跟踪不同运行中的实验、参数和指标
• MLOps 实施：构建可复现、可监控的 ML 基础设施
• 工作流编排：管理复杂的多步骤 ML 工作流
• 模型注册表：管理模型版本和部署生命周期

快速入门：5 步构建 ML 流水线

# 1. Install Airflow and MLflow (check for latest versions at time of use)
pip install apache-airflow==3.1.5 mlflow==3.7.0

# Note: These versions are current as of December 2025
# Check PyPI for latest stable releases: https://pypi.org/project/apache-airflow/

# 2. Initialize Airflow database
airflow db init

# 3. Create DAG file: dags/ml_training_pipeline.py
cat > dags/ml_training_pipeline.py << 'EOF'
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'ml_training_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1)
)

def train_model(**context):
    import mlflow
    import mlflow.sklearn
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.datasets import load_iris
    from sklearn.model_selection import train_test_split

    X, y = load_iris(return_X_y=True)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

    mlflow.set_tracking_uri('http://localhost:5000')
    mlflow.set_experiment('iris-training')

    with mlflow.start_run():
        model = RandomForestClassifier(n_estimators=100)
        model.fit(X_train, y_train)

        accuracy = model.score(X_test, y_test)
        mlflow.log_metric('accuracy', accuracy)
        mlflow.sklearn.log_model(model, 'model')

train = PythonOperator(
    task_id='train_model',
    python_callable=train_model,
    dag=dag
)
EOF

# 4. Start Airflow scheduler and webserver
airflow scheduler &
airflow webserver --port 8080 &

# 5. Trigger pipeline
airflow dags trigger ml_training_pipeline

# Access UI: http://localhost:8080

结果：不到 5 分钟即可获得可运行的 ML 流水线和实验跟踪功能。

核心概念

流水线阶段

1. 数据收集 → 从来源获取原始数据
2. 数据验证 → 检查架构、质量和分布
3. 特征工程 → 将原始数据转换为特征
4. 模型训练 → 使用超参数调优进行训练
5. 模型评估 → 在测试集上验证性能
6. 模型部署 → 指标达标后推送至生产环境
7. 监控 → 跟踪生产环境中的漂移和性能

编排工具对比

工具	最适用于	优势
Airflow	通用 ML 工作流	成熟、灵活、原生支持 Python
Kubeflow	Kubernetes 原生 ML	基于容器、可扩展
MLflow	实验跟踪	模型注册表、版本管理
Prefect	现代 Python 工作流	动态 DAG、原生缓存
Dagster	面向资产的流水线	数据感知、可测试

基本 Airflow DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

default_args = {
    'owner': 'ml-team',
    'depends_on_past': False,
    'email': ['[email protected]'],
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'ml_training_pipeline',
    default_args=default_args,
    description='End-to-end ML training pipeline',
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False
)

def validate_data(**context):
    """Validate input data quality."""
    import pandas as pd

    data_path = "/data/raw/latest.csv"
    df = pd.read_csv(data_path)

    # Validation checks
    assert len(df) > 1000, f"Insufficient data: {len(df)} rows"
    assert df.isnull().sum().sum() < len(df) * 0.1, "Too many nulls"

    context['ti'].xcom_push(key='data_path', value=data_path)
    logger.info(f"Data validation passed: {len(df)} rows")

def train_model(**context):
    """Train ML model with MLflow tracking."""
    import mlflow
    import mlflow.sklearn
    from sklearn.ensemble import RandomForestClassifier

    data_path = context['ti'].xcom_pull(key='data_path', task_ids='validate_data')

    mlflow.set_tracking_uri('http://mlflow:5000')
    mlflow.set_experiment('production-training')

    with mlflow.start_run():
        # Training logic here
        model = RandomForestClassifier(n_estimators=100)
        # model.fit(X, y) ...

        mlflow.log_param('n_estimators', 100)
        mlflow.sklearn.log_model(model, 'model')

validate = PythonOperator(
    task_id='validate_data',
    python_callable=validate_data,
    dag=dag
)

train = PythonOperator(
    task_id='train_model',
    python_callable=train_model,
    dag=dag
)

validate >> train

预防已知问题

1. 任务失败但没有告警

问题：流水线静默失败，直到用户投诉才有人发现。

解决方案：配置失败时的邮件/Slack 告警：

default_args = {
    'email': ['[email protected]'],
    'email_on_failure': True,
    'email_on_retry': False
}

def on_failure_callback(context):
    """Send Slack alert on failure."""
    from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

    slack_msg = f"""
    :red_circle: Task Failed: {context['task_instance'].task_id}
    DAG: {context['task_instance'].dag_id}
    Execution Date: {context['ds']}
    Error: {context.get('exception')}
    """

    SlackWebhookOperator(
        task_id='slack_alert',
        slack_webhook_conn_id='slack_webhook',
        message=slack_msg
    ).execute(context)

task = PythonOperator(
    task_id='critical_task',
    python_callable=my_function,
    on_failure_callback=on_failure_callback,
    dag=dag
)

2. 任务之间缺少 XCom 数据

问题：任务需要上一个任务的 XCom 值，却收到 None，导致崩溃。

解决方案：始终验证 XCom 拉取结果：

def process_data(**context):
    data_path = context['ti'].xcom_pull(
        key='data_path',
        task_ids='upstream_task'
    )

    if data_path is None:
        raise ValueError("No data_path from upstream_task - check XCom push")

    # Process data...

3. DAG 未显示在 UI 中

问题：DAG 文件存在于 dags/ 中，但未显示在 Airflow UI 中。

解决方案：检查 DAG 解析错误：

# Check for syntax errors
python dags/my_dag.py

# View DAG import errors in UI
# Navigate to: Browse → DAG Import Errors

# Common fixes:
# 1. Ensure DAG object is defined in file
# 2. Check for circular imports
# 3. Verify all dependencies installed
# 4. Fix syntax errors

4. 硬编码路径在生产环境中失效

问题：/Users/myname/data/ 这样的路径在本地可用，在生产环境中却会失效。

解决方案：使用 Airflow Variables 或环境变量：

from airflow.models import Variable

def load_data(**context):
    # ❌ Bad: Hardcoded path
    # data_path = "/Users/myname/data/train.csv"

    # ✅ Good: Use Airflow Variable
    data_dir = Variable.get("data_directory", "/data")
    data_path = f"{data_dir}/train.csv"

    # Or use environment variable
    import os
    data_path = os.getenv("DATA_PATH", "/data/train.csv")

5. 卡住的任务占用资源

问题：任务无限期挂起，占用工作进程槽位并浪费资源。

解决方案：为任务设置 execution_timeout：

from datetime import timedelta

task = PythonOperator(
    task_id='long_running_task',
    python_callable=my_function,
    execution_timeout=timedelta(hours=2),  # Kill after 2 hours
    dag=dag
)

6. 未验证数据导致模型训练效果不佳

问题：使用损坏或不完整的数据训练，导致模型在生产环境中表现不佳。

解决方案：添加数据质量验证任务：

def validate_data_quality(**context):
    """Comprehensive data validation."""
    import pandas as pd

    df = pd.read_csv(data_path)

    # Schema validation
    required_cols = ['user_id', 'timestamp', 'feature_a', 'target']
    missing_cols = set(required_cols) - set(df.columns)
    if missing_cols:
        raise ValueError(f"Missing columns: {missing_cols}")

    # Statistical validation
    if df['target'].isnull().sum() > 0:
        raise ValueError("Target column contains nulls")

    if len(df) < 1000:
        raise ValueError(f"Insufficient data: {len(df)} rows")

    logger.info("✅ Data quality validation passed")

7. 未跟踪实验导致知识丢失

问题：无法复现实验结果，也不知道哪些超参数有效。

解决方案：使用 MLflow 跟踪所有实验：

import mlflow

mlflow.set_tracking_uri('http://mlflow:5000')
mlflow.set_experiment('model-experiments')

with mlflow.start_run(run_name='rf_v1'):
    # Log ALL hyperparameters
    mlflow.log_params({
        'model_type': 'random_forest',
        'n_estimators': 100,
        'max_depth': 10,
        'random_state': 42
    })

    # Log ALL metrics
    mlflow.log_metrics({
        'train_accuracy': 0.95,
        'test_accuracy': 0.87,
        'f1_score': 0.89
    })

    # Log model
    mlflow.sklearn.log_model(model, 'model')

何时加载参考资料

加载参考文件，了解详细的生产环境实现：

• Airflow DAG 模式：构建包含错误处理、动态生成、传感器、任务组或重试逻辑的复杂 DAG 时，加载 references/airflow-patterns.md。其中包含完整的生产环境 DAG 示例。

• Kubeflow 与 MLflow 集成：使用 Kubeflow Pipelines 进行容器原生编排、集成 MLflow 跟踪、构建 KFP 组件或管理模型注册表时，加载 references/kubeflow-mlflow.md。

• 流水线监控：实施数据质量检查、漂移检测、告警配置或使用 Prometheus 监控流水线运行状况时，加载 references/pipeline-monitoring.md。

最佳实践

1. 幂等任务：任务重复运行时应产生相同的结果
2. 原子操作：每个任务都做好一件事
3. 为所有内容添加版本控制：数据、代码、模型、依赖项
4. 全面记录日志：记录所有重要事件及其上下文
5. 错误处理：快速失败并提供清晰的错误消息
6. 监控：跟踪管道健康状况、数据质量、模型漂移
7. 测试：集成前独立测试任务
8. 文档：记录 DAG 的用途和任务依赖关系

常见模式

条件执行

from airflow.operators.python import BranchPythonOperator

def choose_branch(**context):
    accuracy = context['ti'].xcom_pull(key='accuracy', task_ids='evaluate')

    if accuracy > 0.9:
        return 'deploy_to_production'
    else:
        return 'retrain_with_more_data'

branch = BranchPythonOperator(
    task_id='check_accuracy',
    python_callable=choose_branch,
    dag=dag
)

train >> evaluate >> branch >> [deploy, retrain]

并行训练

from airflow.utils.task_group import TaskGroup

with TaskGroup('train_models', dag=dag) as train_group:
    train_rf = PythonOperator(task_id='train_rf', ...)
    train_lr = PythonOperator(task_id='train_lr', ...)
    train_xgb = PythonOperator(task_id='train_xgb', ...)

# All models train in parallel
preprocess >> train_group >> select_best

等待数据

from airflow.sensors.filesystem import FileSensor

wait_for_data = FileSensor(
    task_id='wait_for_data',
    filepath='/data/input/{{ ds }}.csv',
    poke_interval=60,  # Check every 60 seconds
    timeout=3600,  # Timeout after 1 hour
    mode='reschedule',  # Don't block worker
    dag=dag
)

wait_for_data >> process_data