【2026年版】AIデータパイプラインの自動化：Apache Airflow・Prefect・Dagsterの比較と実践

# Airflow DAGの基本例
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.s3 import S3CopyObjectOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alert@example.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'ml_data_pipeline',
    default_args=default_args,
    description='ML用データパイプライン',
    schedule_interval='0 2 * * *',  # 毎日AM2時
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['ml', 'data-pipeline'],
)

def extract_data(**context):
    """データソースからの抽出"""
    import pandas as pd
    df = pd.read_sql("SELECT * FROM raw_data WHERE date = %(ds)s",
                     conn, params={'ds': context['ds']})
    df.to_parquet(f"/tmp/raw_{context['ds']}.parquet")
    return f"/tmp/raw_{context['ds']}.parquet"

def transform_data(**context):
    """特徴量エンジニアリング"""
    import pandas as pd
    ti = context['task_instance']
    file_path = ti.xcom_pull(task_ids='extract')
    df = pd.read_parquet(file_path)
    # 特徴量生成処理
    df['feature_1'] = df['col_a'] / df['col_b']
    df['feature_2'] = df['col_c'].rolling(7).mean()
    df.to_parquet(f"/tmp/features_{context['ds']}.parquet")

def validate_data(**context):
    """データ品質チェック"""
    import pandas as pd
    df = pd.read_parquet(f"/tmp/features_{context['ds']}.parquet")
    assert df.isnull().sum().sum() == 0, "NULL値が検出されました"
    assert len(df) > 100, "データ件数が不足しています"

extract = PythonOperator(task_id='extract', python_callable=extract_data, dag=dag)
transform = PythonOperator(task_id='transform', python_callable=transform_data, dag=dag)
validate = PythonOperator(task_id='validate', python_callable=validate_data, dag=dag)

extract >> transform >> validate

# Prefect フローの基本例
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
import pandas as pd

@task(retries=3, retry_delay_seconds=60,
      cache_key_fn=task_input_hash,
      cache_expiration=timedelta(hours=1))
def extract_data(date: str) -> pd.DataFrame:
    """データソースからの抽出"""
    df = pd.read_sql(
        f"SELECT * FROM raw_data WHERE date = '{date}'",
        connection_string
    )
    return df

@task(retries=2)
def transform_data(df: pd.DataFrame) -> pd.DataFrame:
    """特徴量エンジニアリング"""
    df['feature_1'] = df['col_a'] / df['col_b']
    df['feature_2'] = df['col_c'].rolling(7).mean()
    df = df.dropna()
    return df

@task
def validate_data(df: pd.DataFrame) -> bool:
    """データ品質チェック"""
    assert df.isnull().sum().sum() == 0, "NULL値が検出されました"
    assert len(df) > 100, "データ件数が不足しています"
    return True

@task
def save_features(df: pd.DataFrame, date: str):
    """特徴量の保存"""
    df.to_parquet(f"s3://ml-features/features_{date}.parquet")

@flow(name="ML Data Pipeline", log_prints=True)
def ml_data_pipeline(date: str):
    """MLデータパイプラインのメインフロー"""
    raw_df = extract_data(date)
    features_df = transform_data(raw_df)
    is_valid = validate_data(features_df)
    if is_valid:
        save_features(features_df, date)
        print(f"パイプライン完了: {date}")

if __name__ == "__main__":
    ml_data_pipeline(date="2026-02-12")

# Dagster アセットの基本例
from dagster import asset, AssetIn, DailyPartitionsDefinition, Definitions
from dagster import MaterializeResult, MetadataValue
import pandas as pd

daily_partitions = DailyPartitionsDefinition(start_date="2026-01-01")

@asset(
    partitions_def=daily_partitions,
    description="データソースからの生データ抽出",
    group_name="data_pipeline",
)
def raw_data(context) -> pd.DataFrame:
    """生データの抽出"""
    partition_date = context.partition_key
    df = pd.read_sql(
        f"SELECT * FROM raw_data WHERE date = '{partition_date}'",
        connection_string
    )
    context.log.info(f"抽出件数: {len(df)}")
    return df

@asset(
    ins={"raw_data": AssetIn()},
    partitions_def=daily_partitions,
    description="特徴量エンジニアリング済みデータ",
    group_name="data_pipeline",
)
def feature_data(context, raw_data: pd.DataFrame) -> pd.DataFrame:
    """特徴量の生成"""
    df = raw_data.copy()
    df['feature_1'] = df['col_a'] / df['col_b']
    df['feature_2'] = df['col_c'].rolling(7).mean()
    df = df.dropna()
    return MaterializeResult(
        metadata={
            "row_count": MetadataValue.int(len(df)),
            "columns": MetadataValue.text(str(df.columns.tolist())),
        }
    )

@asset(
    ins={"feature_data": AssetIn()},
    partitions_def=daily_partitions,
    description="品質検証済みの特徴量データ",
    group_name="data_pipeline",
)
def validated_features(context, feature_data: pd.DataFrame) -> pd.DataFrame:
    """データ品質の検証"""
    null_count = feature_data.isnull().sum().sum()
    row_count = len(feature_data)
    assert null_count == 0, f"NULL値が{null_count}件検出されました"
    assert row_count > 100, f"データ件数が不足: {row_count}件"
    context.log.info(f"検証完了: {row_count}件")
    return feature_data

defs = Definitions(assets=[raw_data, feature_data, validated_features])

# Airflow: 大規模バッチETLの例
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryInsertJobOperator,
)
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import (
    GCSToBigQueryOperator,
)

load_to_bq = GCSToBigQueryOperator(
    task_id='load_raw_data',
    bucket='data-lake',
    source_objects=['raw/{{ ds }}/*.parquet'],
    destination_project_dataset_table='project.dataset.raw_table',
    source_format='PARQUET',
    write_disposition='WRITE_TRUNCATE',
    dag=dag,
)

transform_bq = BigQueryInsertJobOperator(
    task_id='transform_features',
    configuration={
        "query": {
            "query": """
                CREATE OR REPLACE TABLE `project.dataset.features`
                PARTITION BY date AS
                SELECT
                    *,
                    col_a / NULLIF(col_b, 0) AS feature_1,
                    AVG(col_c) OVER (
                        ORDER BY date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
                    ) AS feature_2
                FROM `project.dataset.raw_table`
                WHERE date = '{{ ds }}'
            """,
            "useLegacySql": False,
        }
    },
    dag=dag,
)

load_to_bq >> transform_bq

# Prefect: ML学習パイプラインの例
from prefect import flow, task
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score, f1_score
import mlflow

@task
def prepare_dataset(feature_path: str):
    """データセットの準備"""
    import pandas as pd
    df = pd.read_parquet(feature_path)
    X = df.drop('target', axis=1)
    y = df['target']
    return train_test_split(X, y, test_size=0.2, random_state=42)

@task
def train_model(X_train, y_train, params: dict):
    """モデルの学習"""
    model = GradientBoostingClassifier(**params)
    model.fit(X_train, y_train)
    return model

@task
def evaluate_model(model, X_test, y_test):
    """モデルの評価"""
    predictions = model.predict(X_test)
    metrics = {
        'accuracy': accuracy_score(y_test, predictions),
        'f1_score': f1_score(y_test, predictions, average='weighted'),
    }
    return metrics

@flow(name="ML Training Pipeline")
def ml_training_pipeline(
    feature_path: str,
    model_params: dict = None,
):
    """ML学習パイプライン"""
    if model_params is None:
        model_params = {
            'n_estimators': 200,
            'max_depth': 5,
            'learning_rate': 0.1,
        }

    X_train, X_test, y_train, y_test = prepare_dataset(feature_path)
    model = train_model(X_train, y_train, model_params)
    metrics = evaluate_model(model, X_test, y_test)

    with mlflow.start_run():
        mlflow.log_params(model_params)
        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, "model")

    print(f"学習完了: {metrics}")
    return metrics

# Dagster: データ品質管理重視のパイプライン
from dagster import (
    asset, AssetIn, FreshnessPolicy,
    AutoMaterializePolicy, AssetCheckResult,
    asset_check, AssetCheckSpec,
)
import pandas as pd

@asset(
    freshness_policy=FreshnessPolicy(maximum_lag_minutes=60),
    auto_materialize_policy=AutoMaterializePolicy.eager(),
    description="品質チェック付き特徴量データ",
)
def quality_checked_features(context, raw_data: pd.DataFrame) -> pd.DataFrame:
    """品質チェック済みの特徴量を生成"""
    df = raw_data.copy()
    df['feature_1'] = df['col_a'] / df['col_b'].replace(0, float('nan'))
    df['feature_2'] = df['col_c'].rolling(7).mean()

    # 品質メトリクスをログ
    context.log.info(f"NULL率: {df.isnull().mean().to_dict()}")
    context.log.info(f"レコード数: {len(df)}")

    return df

@asset_check(asset=quality_checked_features)
def check_no_nulls(quality_checked_features: pd.DataFrame) -> AssetCheckResult:
    """NULL値がないことを確認"""
    null_count = quality_checked_features.isnull().sum().sum()
    return AssetCheckResult(
        passed=null_count == 0,
        metadata={"null_count": null_count},
    )

@asset_check(asset=quality_checked_features)
def check_row_count(quality_checked_features: pd.DataFrame) -> AssetCheckResult:
    """最低レコード数を確認"""
    row_count = len(quality_checked_features)
    return AssetCheckResult(
        passed=row_count >= 100,
        metadata={"row_count": row_count},
    )

# docker-compose.yml（監視スタック）
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:v2.50.0
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  grafana:
    image: grafana/grafana:10.3.0
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=secure_password
    volumes:
      - ./grafana/dashboards:/var/lib/grafana/dashboards
    ports:
      - "3000:3000"

  statsd-exporter:
    image: prom/statsd-exporter:v0.26.0
    ports:
      - "9125:9125/udp"
      - "9102:9102"

# .github/workflows/pipeline-ci.yml
name: Data Pipeline CI/CD
on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run pipeline tests
        run: pytest tests/ -v --cov=pipelines
      - name: Lint DAGs
        run: python -m py_compile pipelines/*.py

  deploy:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to production
        run: |
          # Airflowの場合
          aws s3 sync dags/ s3://airflow-dags-bucket/dags/
          # Prefectの場合
          prefect deploy --all
          # Dagsterの場合
          dagster-cloud ci deploy