データパイプラインOSS比較：Apache Airflow vs Prefect vs Dagster でETLをセルフホスト

データエンジニアリングチームが直面する課題は「定期的なデータ収集・変換・ロードのジョブが複雑化し、依存関係管理・エラーリトライ・モニタリングが追いつかない」ことです。Apache Airflow（業界標準）・Prefect（Pythonファーストのモダン設計）・Dagster（アセット指向の型安全ETL）は、OSSのデータパイプラインオーケストレーターで、スケジュール実行・依存管理・ログ・アラートを統合します。

データパイプラインツールの選定理由

• 依存関係管理: タスクAが成功した後にタスクBを実行する依存グラフを定義したい
• エラーリトライ: 外部APIの一時的な障害で失敗したタスクを自動再試行したい
• スケジューリング: 毎日AM3時にDBから全データをS3にエクスポートしたい
• 可観測性: どのタスクが成功・失敗・遅延しているかをダッシュボードで確認したい
• データ品質: パイプラインの中間成果物（データセット）のスキーマと品質を検証したい

主要ツールの概要

Apache Airflow

業界で最も広く使われるデータパイプラインオーケストレーターです。GitHubスター36k+。PythonでDAG（有向非巡回グラフ）としてパイプラインを定義し、WebUIでスケジュール管理・実行ログ・リトライを管理します。AWSのManaged Airflow（MWAA）・Google Cloud Composerなどマネージドサービスも多いです。

# AirflowをDockerで起動（Celery Executor）
# docker-compose.yamlをAirflow公式から取得
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'

# 環境変数ファイルを作成
echo "AIRFLOW_UID=$(id -u)" > .env
echo "AIRFLOW_GID=0" >> .env

# 初期化
docker compose up airflow-init

# 起動（デフォルトでポート8080）
docker compose up -d

# Airflow DAGでSupabaseからデータを抽出してS3に保存
# dags/export_articles_to_s3.py

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.amazon.aws.transfers.sql_to_s3 import SqlToS3Operator

default_args = {
    "owner": "data-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-alerts@yoursite.com"],
}

with DAG(
    "export_articles_to_s3",
    default_args=default_args,
    description="毎日の記事データをS3にエクスポート",
    schedule_interval="0 3 * * *",  # 毎日AM3時
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["export", "articles"],
) as dag:

    # タスク1: DBから記事データを取得してS3に保存
    export_articles = SqlToS3Operator(
        task_id="export_articles",
        sql="SELECT * FROM articles WHERE status = 'published'",
        s3_bucket="your-data-bucket",
        s3_key="exports/articles/{{ ds }}/articles.csv",
        postgres_conn_id="supabase_postgres",
        aws_conn_id="aws_s3",
        replace=True,
    )

    # タスク2: エクスポート完了をSlackに通知
    def notify_slack(**context):
        import requests
        webhook_url = context["var"]["value"].get("slack_webhook_url")
        requests.post(webhook_url, json={
            "text": f"記事データのS3エクスポート完了: {context['ds']}"
        })

    notify = PythonOperator(
        task_id="notify_completion",
        python_callable=notify_slack,
        provide_context=True,
    )

    export_articles >> notify

# AirflowのカスタムPythonタスクで記事分析パイプライン
# dags/analyze_article_performance.py

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@weekly",
    start_date=pendulum.datetime(2024, 1, 1, tz="Asia/Tokyo"),
    catchup=False,
)
def analyze_article_performance():

    @task
    def fetch_articles():
        import psycopg2
        import json
        conn = psycopg2.connect(
            host="db.ucceyqlkrzojulwyryia.supabase.co",
            database="postgres",
            user="postgres",
            password="{{ var.value.supabase_password }}",
        )
        cursor = conn.cursor()
        cursor.execute(
            "SELECT slug, title, created_at FROM articles WHERE status='published' ORDER BY created_at DESC LIMIT 100"
        )
        return [{"slug": r[0], "title": r[1], "created_at": str(r[2])} for r in cursor.fetchall()]

    @task
    def calculate_stats(articles: list):
        from collections import Counter
        categories = [a["slug"].split("-")[0] for a in articles]
        return {"total": len(articles), "by_prefix": dict(Counter(categories))}

    @task
    def save_report(stats: dict):
        print(f"週次レポート: {stats}")
        # BigQuery・Notionへの保存ロジックを追加

    articles = fetch_articles()
    stats = calculate_stats(articles)
    save_report(stats)

analyze_article_performance()

Prefect

Pythonファーストのモダンなデータパイプラインフレームワークです。GitHubスター16k+。Airflowと比べて学習コストが低く、既存のPythonスクリプトに@flow・@taskデコレータを付けるだけでオーケストレーション対応になります。クラウド版（Prefect Cloud）も無料プランあり。

# Prefect serverをDockerで起動
docker run -d   --name prefect-server   -p 4200:4200   -e PREFECT_API_DATABASE_CONNECTION_URL="postgresql+asyncpg://prefect:prefect_pass@postgres:5432/prefect"   prefecthq/prefect:3-latest   prefect server start --host 0.0.0.0

# PrefectでNext.js記事バックログ処理パイプライン
# flows/process_article_backlog.py
# pip install prefect

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
import json
import httpx

@task(retries=3, retry_delay_seconds=30, log_prints=True)
async def fetch_backlog():
    with open("data/article-backlog.json", "r", encoding="utf-8") as f:
        return json.load(f)

@task(retries=2, log_prints=True)
async def check_published_slugs():
    async with httpx.AsyncClient() as client:
        res = await client.get(
            "https://ucceyqlkrzojulwyryia.supabase.co/rest/v1/articles?select=slug&status=eq.published",
            headers={"apikey": "your-anon-key"},
        )
        return {a["slug"] for a in res.json()}

@task(log_prints=True)
async def filter_unpublished(backlog: list, published_slugs: set):
    return [item for item in backlog if item["slug"] not in published_slugs]

@task(log_prints=True)
def generate_article(theme: dict) -> dict:
    print(f"記事生成中: {theme['title']}")
    # Claude APIを呼び出して記事を生成するロジック
    return {"slug": theme["slug"], "content_md": "# Generated content..."}

@flow(
    name="article-production-pipeline",
    task_runner=ConcurrentTaskRunner(),
    log_prints=True,
)
async def produce_articles():
    backlog = await fetch_backlog()
    published = await check_published_slugs()
    unpublished = await filter_unpublished(backlog, published)

    # 最大5本を並行生成
    targets = unpublished[:5]
    articles = generate_article.map(targets)

    print(f"生成完了: {len(articles)}本")
    return articles

if __name__ == "__main__":
    import asyncio
    asyncio.run(produce_articles())

Dagster

アセット（データセット）指向のデータパイプラインフレームワークです。GitHubスター12k+。「パイプラインを実行する」ではなく「データアセットを定義して最新に保つ」という思想で設計されており、dbt・Spark・Great Expectations との統合が充実しています。

# Dagsterのアセット定義（データアセット指向）
# assets/article_pipeline.py
# pip install dagster dagster-webserver

from dagster import asset, AssetExecutionContext, define_asset_job, ScheduleDefinition
import json
import httpx

@asset(description="バックログから未公開テーマを取得")
def unpublished_themes(context: AssetExecutionContext) -> list:
    with open("data/article-backlog.json", "r", encoding="utf-8") as f:
        backlog = json.load(f)

    res = httpx.get(
        "https://ucceyqlkrzojulwyryia.supabase.co/rest/v1/articles?select=slug",
        headers={"apikey": "your-anon-key"},
    )
    published = {a["slug"] for a in res.json()}
    themes = [t for t in backlog if t["slug"] not in published]

    context.log.info(f"未公開テーマ数: {len(themes)}")
    return themes[:5]

@asset(
    description="未公開テーマから記事コンテンツを生成",
    deps=[unpublished_themes],
)
def generated_articles(unpublished_themes: list) -> list:
    articles = []
    for theme in unpublished_themes:
        # 実際にはClaude APIを呼び出す
        articles.append({
            "slug": theme["slug"],
            "title": theme["title"],
            "content_md": f"# {theme['title']}

コンテンツ...",
        })
    return articles

@asset(
    description="生成した記事をSupabaseに投入",
    deps=[generated_articles],
)
def published_articles(generated_articles: list) -> int:
    for article in generated_articles:
        httpx.post(
            "https://ucceyqlkrzojulwyryia.supabase.co/rest/v1/articles",
            headers={"apikey": "your-anon-key", "Content-Type": "application/json"},
            json=article,
        )
    return len(generated_articles)

# 毎日AM2時にパイプラインを実行
daily_schedule = ScheduleDefinition(
    job=define_asset_job("article_pipeline"),
    cron_schedule="0 2 * * *",
)

機能比較表

比較項目	Apache Airflow	Prefect	Dagster
ライセンス	Apache-2.0	Apache-2.0	Apache-2.0
設計思想	タスク依存グラフ	Pythonデコレータ	データアセット指向
学習コスト	高い	低い	中程度
UIダッシュボード	✅ 成熟	✅	✅
リトライ	✅	✅	✅
dbt連携	✅	✅	✅ 優秀
データ品質チェック	❌	❌	✅
クラウド版	AWS MWAA等	Prefect Cloud	Dagster Cloud
最小RAM	4GB	512MB	1GB
GitHub Stars	36k+	16k+	12k+

データパイプラインの結果を格納するログ管理はopen-source-log-management（/categories/devops）を参照。パイプラインのエラーを検知するアラートはdevopsカテゴリ（/categories/devops）でまとめています。

FAQ

Q. Apache AirflowとPrefectのどちらから始めるべきか、チームの状況で判断する方法は？

A. 以下の基準で判断します。Prefectを選ぶケース: ①チームが小規模（5人以下）でデータエンジニアリング専任がいない②既存のPythonスクリプトをできるだけ変更せずにスケジュール実行したい③AirflowのDAG定義の学習コストを避けたい④Prefect Cloud（無料プラン: 月3ワークスペース）を使うと管理コスト最小化。Airflowを選ぶケース: ①データエンジニアがいてAirflowの経験がある②AWS MWAA・GCP Composerとの統合が前提③コミュニティが最大でProvider（プラグイン）数が最多④既存のAirflowワークフローを引き継ぐ必要がある。Dagsterを選ぶケース: dbtを使ったデータトランスフォームがメインで、データアセットの系譜（リネージュ）を追いたい。

Q. PrefectのデプロイはDockerを使う場合とVercel Cronと比べてどちらがよいですか？

A. 用途によって適切な選択が変わります。Vercel Cron（vercel.jsonのcrons）: シンプルなHTTPエンドポイント呼び出しには最適。無料・設定ゼロ・Next.jsと一体化。ただしタイムアウトが10秒〜5分に制限されており、複雑な処理には不向き。Prefect（セルフホスト）: 数分〜数時間かかる重いデータ処理に最適。リトライ・依存管理・ログ・UIダッシュボードが必要なケース。ハイブリッド推奨: 軽いタスク（Slack通知・メール送信・Webhook呼び出し）→Vercel Cron。重いバッチ処理（データエクスポート・AI記事生成・DB集計）→Prefect/Airflow。

Q. DagsterのアセットグラフとAirflowのDAGの概念的な違いは何ですか？

A. 根本的な設計思想が異なります。Airflow DAG: 「何をする（タスク）」を中心に設計。extract → transform → loadという処理の流れを定義。Dagster Asset: 「何を作る（データ）」を中心に設計。raw_articles（生データ）→ cleaned_articles（クリーンデータ）→ article_stats（集計）という成果物の依存関係を定義。実用上の違い: ①Dagsterは「記事アセットが古くなったら再生成する」という宣言的な管理ができる②データカタログとしての機能があり、誰がどのデータを作ったかの追跡が容易③dbtのモデルと直接統合でき、SQLとPythonのパイプラインを混在できる。チーム規模が大きくデータの系譜管理が重要→Dagster。処理フローのオーケストレーションだけ必要→Airflow/Prefect。

Q. セルフホストのAirflowでメモリ消費を抑えるための設定は？

A. AirflowのメモリはExecutorの選択で大きく変わります。①SequentialExecutor（デフォルト）: 最小構成だが並列実行不可。開発・テスト用。②LocalExecutor（推奨: 小〜中規模）: 並列実行可能・Celeryより軽量・同一マシン上でsubprocessを起動。設定: AIRFLOW__CORE__EXECUTOR=LocalExecutor。最小RAM: 2GB。③CeleryExecutor（大規模）: 複数ワーカーノードに分散実行。Redis/RabbitMQ + 複数Workerで高スループット。最小RAM: 4GB+。メモリ節約設定: AIRFLOW__SCHEDULER__MAX_THREADS=2（スケジューラのスレッド数を制限）、AIRFLOW__WEBSERVER__WORKERS=2（WebサーバーのWorker数を制限）。Dockerのメモリ制限: mem_limit: 2gでコンテナのRAMを2GBに制限することで、他のサービスへの影響を防げます。

まとめ

ユースケース	推奨ツール
業界標準・豊富なProvider	Apache Airflow
学習コスト低・既存Pythonスクリプト流用	Prefect
dbt連携・データアセット管理	Dagster
小規模・すぐ始めたい	Prefect

関連外部リソース

• Apache Airflow GitHubリポジトリ↗
• Prefect GitHubリポジトリ↗
• Dagster GitHubリポジトリ↗
• Airflow公式ドキュメント↗