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. アトミックな操作：各タスクが1つの処理に専念するようにします
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