Apache Airflow 2.10データパイプライン実践：5つのDAGオーケストレーションパターンでプロダクションETL構築

ETLスクリプト失制、タスク依存混乱、データパイプラインのサイレント障害

午前2時、データウェアハウスの日次増分が50万から3万に急減、Airflow DAGは全て緑で成功を示している——誰も上流APIがページネーションロジックを変更したことに気付いていない；ETLスクリプトが20個のCron Jobに散乱し、上流の失敗が下流で気付されず、ダーティデータがそのままレポートに流れ込む；新入社員が書いたDAGの循環依存でAirflow Schedulerがフリーズ。2026年、Apache Airflow 2.10はDynamic Task Mappingの強化、改良されたTaskFlow API、より強力なデータ認識スケジューリングを提供——DAG設計からプロダクション監視まで、1つのシステムで完結。

本記事は5つのDAGオーケストレーションパターンを通じて、TaskFlow API設計→ダイナミックタスクマッピング並列ETL→カスタムOperatorとSensor→XComデータ連携と分岐→エラーリトライとSLAモニタリングの全チェーン実践を、各ステップで完全に実行可能なPythonコードとともに解説。

Airflow中心コンセプト

コンセプト	説明
DAG	有向非巡環グラフ、タスク間の依存関係と実行順序を定義するAirflowの中心抽象
Task	DAG内の単一実行ユニット、Operatorインスタンスに対応
Operator	Taskの実行ロジックカプセル化（PythonOperator、BashOperator、カスタムOperator）
TaskFlow API	Airflow 2.0+で導入されたPython関数型DAG定義方式、`@task`デコレータでOperatorを置き換え
XCom	タスク間データ連携メカニズム、デフォルト48KB制限、2.10でカスタムXCom Backendをサポート
Dynamic Task Mapping	Airflow 2.3+で導入されたダイナミックタスク展開、ランタイムデータから並列タスクを動的生成
Sensor	特殊なOperator、条件が満たされるまで維続的にチェック
Connection	Airflow管理の接続情報（データベース、API等）、DAG内のハードコードを回避
Pool	リソースプール、同時実行タスク数を制限
Trigger	Airflow 2.2+で導入された非同期トリガー、SensorがWorker Slotを占有しないように
SLA	Service Level Agreement、指定時間内にタスク完了が必要、タイムアウトでアラートをトリガー
Dataset	Airflow 2.4+で導入されたデータ認識スケジューリング、プロデューサーがデータを生成するとコンシューマーが自動トリガー

問題分析：データパイプラインオーケストレーションの5つの課題

DAG設計複雑化失制：ETLタスクが3から30に增加し、DAGファイルがスパゲッティコード化
並列ETL手動オーケストレーション：100個のデータソースを毎日処理、手動で100個のTaskを書くのは非現実的
カスタムロジックの再利用が困難：社内システムのAPI呼び出しやデータ検証ロジックが各DAGで重複実装
タスク間データ連携の制限：XComのデフォルト48KB制限、大容量データ連携がエラーに
プロダクション環境の信頼性不足：タスク失敗時のリトライなし、SLAタイムアウト時のアラートなし

ステップバイストップ：5つのAirflow DAGオーケストレーションパターン

パターン1：TaskFlow APIでのDAG設計

# dags/taskflow_etl_dag.py
from datetime import datetime, timedelta
from airflow.decorators import dag, task

default_args = {
    "owner": "data-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

@dag(
    dag_id="taskflow_etl_pipeline",
    default_args=default_args,
    description="TaskFlow API ETL pipeline",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["etl", "taskflow"],
)
def taskflow_etl_pipeline():

    @task
    def extract_user_events(execution_date: str) -> dict:
        """从PostgreSQL抽取用户事件数据"""
        import psycopg2
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("postgres_source")
        pg_conn = psycopg2.connect(
            host=conn.host, port=conn.port, database=conn.schema,
            user=conn.login, password=conn.password,
        )
        cur = pg_conn.cursor()
        cur.execute(
            "SELECT user_id, event_type, amount, created_at "
            "FROM user_events "
            "WHERE created_at >= %s AND created_at < %s",
            (execution_date, execution_date + timedelta(days=1)),
        )
        rows = cur.fetchall()
        pg_conn.close()
        return {"row_count": len(rows), "data": [dict(zip(["user_id", "event_type", "amount", "created_at"], r)) for r in rows[:1000]]}

    @task
    def extract_api_data(execution_date: str) -> dict:
        """从REST API抽取数据"""
        import requests
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("api_source")
        resp = requests.get(
            f"{conn.host}/v2/data",
            headers={"Authorization": f"Bearer {conn.password}"},
            params={"date": execution_date},
            timeout=30,
        )
        resp.raise_for_status()
        data = resp.json()
        return {"row_count": len(data.get("records", [])), "data": data.get("records", [])[:1000]}

    @task
    def transform(user_events: dict, api_data: dict) -> dict:
        """合并、去重、清洗数据"""
        import pandas as pd
        df_events = pd.DataFrame(user_events["data"])
        df_api = pd.DataFrame(api_data["data"])
        if df_events.empty and df_api.empty:
            return {"row_count": 0, "data": []}
        df_all = pd.concat([df_events, df_api], ignore_index=True)
        df_all = df_all.drop_duplicates(subset=["user_id", "event_type"])
        df_all = df_all.dropna(subset=["user_id"])
        df_all["amount"] = df_all["amount"].fillna(0).astype(float)
        return {"row_count": len(df_all), "data": df_all.to_dict("records")[:500]}

    @task
    def load(transformed: dict) -> int:
        """加载数据到数据仓库"""
        import psycopg2
        from airflow.hooks.base import BaseHook
        if transformed["row_count"] == 0:
            return 0
        conn = BaseHook.get_connection("postgres_warehouse")
        pg_conn = psycopg2.connect(
            host=conn.host, port=conn.port, database=conn.schema,
            user=conn.login, password=conn.password,
        )
        cur = pg_conn.cursor()
        insert_sql = """
            INSERT INTO analytics.user_events_daily
            (user_id, event_type, amount, event_date)
            VALUES (%s, %s, %s, %s)
            ON CONFLICT (user_id, event_type, event_date) DO UPDATE
            SET amount = EXCLUDED.amount
        """
        batch = [(r.get("user_id"), r.get("event_type"), r.get("amount", 0), r.get("created_at")) for r in transformed["data"]]
        cur.executemany(insert_sql, batch)
        pg_conn.commit()
        pg_conn.close()
        return len(batch)

    @task
    def quality_check(loaded_count: int, transformed_count: int) -> str:
        if transformed_count > 0 and loaded_count / transformed_count < 0.95:
            raise ValueError(f"Quality gate failed: loaded {loaded_count}/{transformed_count}")
        return "PASSED"

    user_events = extract_user_events("{{ ds }}")
    api_data = extract_api_data("{{ ds }}")
    transformed = transform(user_events, api_data)
    loaded = load(transformed)
    quality_check(loaded, transformed["row_count"])

taskflow_etl_pipeline()

pip install "apache-airflow[postgres,amazon]==2.10.3"
airflow db init
airflow dags list | grep taskflow
airflow dags test taskflow_etl_pipeline 2026-06-19

パターン2：Dynamic Task Mappingによる並列ETL

# dags/dynamic_mapping_dag.py
from datetime import datetime
from airflow.decorators import dag, task

@dag(dag_id="dynamic_mapping_etl", schedule="0 3 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "dynamic-mapping"])
def dynamic_mapping_etl():

    @task
    def discover_data_sources() -> list[dict]:
        """动态发现需要处理的数据源列表"""
        import psycopg2
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("postgres_source")
        pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
        cur = pg_conn.cursor()
        cur.execute("SELECT source_id, source_type, connection_id FROM data_sources WHERE active = true")
        sources = [{"source_id": row[0], "source_type": row[1], "connection_id": row[2]} for row in cur.fetchall()]
        pg_conn.close()
        return sources

    @task
    def extract_from_source(source: dict) -> dict:
        """从单个数据源抽取数据"""
        import psycopg2, requests
        from airflow.hooks.base import BaseHook
        source_type, source_id = source["source_type"], source["source_id"]
        if source_type == "postgres":
            conn = BaseHook.get_connection(source["connection_id"])
            pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
            cur = pg_conn.cursor()
            cur.execute(f"SELECT * FROM source_{source_id}_data WHERE processed = false LIMIT 5000")
            rows = cur.fetchall()
            pg_conn.close()
            return {"source_id": source_id, "row_count": len(rows), "status": "success"}
        elif source_type == "api":
            conn = BaseHook.get_connection(source["connection_id"])
            resp = requests.get(f"{conn.host}/v1/data/{source_id}", headers={"Authorization": f"Bearer {conn.password}"}, timeout=30)
            resp.raise_for_status()
            return {"source_id": source_id, "row_count": len(resp.json()), "status": "success"}
        return {"source_id": source_id, "row_count": 0, "status": "skipped"}

    @task
    def validate_extracted(extract_result: dict) -> dict:
        if extract_result["status"] != "success" or extract_result["row_count"] < 1:
            return {**extract_result, "valid": False}
        return {**extract_result, "valid": True}

    @task
    def aggregate_results(validation_results: list[dict]) -> dict:
        total = len(validation_results)
        valid = sum(1 for r in validation_results if r.get("valid"))
        total_rows = sum(r.get("row_count", 0) for r in validation_results if r.get("valid"))
        return {"total_sources": total, "valid_sources": valid, "total_rows": total_rows}

    # Dynamic Task Mapping: expand()将列表展开为并行任务
    sources = discover_data_sources()
    extract_results = extract_from_source.expand(source=sources)
    validation_results = validate_extracted.expand(extract_result=extract_results)
    aggregate_results(validation_results)

dynamic_mapping_etl()

# plugins/operators/internal_api_operator.py
from airflow.models import BaseOperator
from airflow.exceptions import AirflowSkipException
from typing import Any
import requests

class InternalApiOperator(BaseOperator):
    """调用公司内部API的自定义Operator"""
    template_fields = ("endpoint", "payload")

    def __init__(self, endpoint: str, payload: dict | None = None, method: str = "POST",
                 connection_id: str = "internal_api", timeout: int = 60, skip_on_empty: bool = False, **kwargs):
        super().__init__(**kwargs)
        self.endpoint, self.payload, self.method = endpoint, payload or {}, method
        self.connection_id, self.timeout, self.skip_on_empty = connection_id, timeout, skip_on_empty

    def execute(self, context: Any) -> dict:
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection(self.connection_id)
        url = f"{conn.host}{self.endpoint}"
        headers = {"Authorization": f"Bearer {conn.password}", "Content-Type": "application/json"}
        self.log.info(f"Calling {self.method} {url}")
        try:
            if self.method.upper() == "GET":
                resp = requests.get(url, headers=headers, params=self.payload, timeout=self.timeout)
            else:
                resp = requests.post(url, headers=headers, json=self.payload, timeout=self.timeout)
            resp.raise_for_status()
            result = resp.json()
            if self.skip_on_empty and not result:
                raise AirflowSkipException("API returned empty result")
            return result
        except requests.exceptions.Timeout:
            raise TimeoutError(f"API call timed out after {self.timeout}s")
        except requests.exceptions.HTTPError as e:/n            if e.response.status_code == 404 and self.skip_on_empty:
                raise AirflowSkipException(f"Resource not found: {url}")
            raise

# plugins/sensors/data_ready_sensor.py
from airflow.sensors.base import BaseSensorOperator
from typing import Any

class DataReadySensor(BaseSensorOperator):
    """检测数据源是否已就绪的Sensor"""
    template_fields = ("source_table", "expected_date")

    def __init__(self, source_table: str, expected_date: str, connection_id: str = "postgres_source",
                 min_row_count: int = 1, poke_interval: int = 300, timeout: int = 7200, **kwargs):
        super().__init__(poke_interval=poke_interval, timeout=timeout, **kwargs)
        self.source_table, self.expected_date = source_table, expected_date
        self.connection_id, self.min_row_count = connection_id, min_row_count

    def poke(self, context: Any) -> bool:
        import psycopg2
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection(self.connection_id)
        pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
        cur = pg_conn.cursor()
        cur.execute(f"SELECT COUNT(*) FROM {self.source_table} WHERE partition_date = %s AND status = 'ready'", (self.expected_date,))
        count = cur.fetchone()[0]
        pg_conn.close()
        self.log.info(f"Data ready check: {count} rows (min: {self.min_row_count})")
        return count >= self.min_row_count

パターン3：カスタムOperatorとSensor

# dags/xcom_branching_dag.py
from datetime import datetime
from airflow.decorators import dag, task
from airflow.operators.python import BranchPythonOperator

@dag(dag_id="xcom_branching_pipeline", schedule="0 5 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "xcom", "branching"])
def xcom_branching_pipeline():

    @task
    def detect_data_type() -> dict:
        import psycopg2
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("postgres_source")
        pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
        cur = pg_conn.cursor()
        cur.execute("SELECT COUNT(*), SUM(amount) FROM user_events WHERE created_at >= CURRENT_DATE - INTERVAL '1 day'")
        row_count, total_amount = cur.fetchone()
        pg_conn.close()
        return {"row_count": row_count, "total_amount": float(total_amount or 0),
                "data_size": "large" if row_count > 100000 else "small"}

    @task
    def process_small_batch(metadata: dict) -> dict:
        return {"path": "small_batch", "processed": metadata["row_count"], "status": "completed"}

    @task
    def process_large_batch(metadata: dict) -> dict:
        chunk_size = 10000
        chunks = (metadata["row_count"] + chunk_size - 1) // chunk_size
        return {"path": "large_batch", "processed": metadata["row_count"], "chunks": chunks, "status": "completed"}

    def choose_processing_branch(**context):
        ti = context["ti"]
        metadata = ti.xcom_pull(task_ids="detect_data_type")
        return "process_large_batch" if metadata["data_size"] == "large" else "process_small_batch"

    metadata = detect_data_type()
    branch = BranchPythonOperator(task_id="choose_branch", python_callable=choose_processing_branch)
    small_result = process_small_batch(metadata)
    large_result = process_large_batch(metadata)
    metadata >> branch >> [small_result, large_result]

xcom_branching_pipeline()

# custom_xcom_backend.py - 自定义XCom Backend突破48KB限制
import json, hashlib
from airflow.models.xcom import BaseXCom
from typing import Any

class RedisXComBackend(BaseXCom):
    @staticmethod
    def serialize_value(value: Any) -> bytes:
        import redis
        r = redis.Redis(host="redis-host", port=6379, db=2)
        serialized = json.dumps(value, default=str).encode("utf-8")
        if len(serialized) <= 48000:
            return BaseXCom.serialize_value(value)
        key = f"xcom:{hashlib.md5(serialized).hexdigest()}"
        r.setex(key, 86400 * 7, serialized)
        return BaseXCom.serialize_value({"__xcom_redis_key": key, "size": len(serialized)})

    @staticmethod
    def deserialize_value(result: Any) -> Any:
        import redis
        value = BaseXCom.deserialize_value(result)
        if isinstance(value, dict) and "__xcom_redis_key" in value:
            r = redis.Redis(host="redis-host", port=6379, db=2)
            data = r.get(value["__xcom_redis_key"])
            if data: return json.loads(data)
            raise ValueError(f"XCom data expired in Redis")
        return value

# airflow.cfg: xcom_backend = custom_xcom_backend.RedisXComBackend

パターン4：XComデータ連携とパイプライン分岐

# dags/error_handling_sla_dag.py
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.exceptions import AirflowSkipException, AirflowFailOperator

default_args = {
    "owner": "data-team", "retries": 3, "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True, "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=60),
    "email_on_failure": True, "email": ["data-team@company.com"],
}

@dag(dag_id="error_handling_sla_pipeline", default_args=default_args,
     schedule="0 1 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "production", "sla"])
def error_handling_sla_pipeline():

    @task(retries=5, retry_delay=timedelta(minutes=2), retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=20))
    def extract_with_retry(**context) -> dict:
        import requests
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("api_source")
        try:
            resp = requests.get(f"{conn.host}/v2/data", headers={"Authorization": f"Bearer {conn.password}"}, params={"date": context["ds"]}, timeout=30)
            resp.raise_for_status()
            return {"row_count": len(resp.json().get("records", [])), "status": "success"}
        except requests.exceptions.ConnectionError as e:/n            raise ConnectionError(f"Connection failed: {e}")
        except requests.exceptions.HTTPError as e:/n            if e.response.status_code == 429: raise ConnectionError(f"Rate limited: {e}")
            elif e.response.status_code >= 500: raise ConnectionError(f"Server error: {e}")
            elif e.response.status_code == 404: raise AirflowSkipException(f"Data not found")
            else: raise AirflowFailOperator(f"Client error: {e}")

    @task
    def transform_with_validation(raw_data: dict) -> dict:
        import pandas as pd
        if raw_data.get("status") != "success": raise ValueError("Upstream failed")
        df = pd.DataFrame(raw_data.get("records", []))
        if df.empty: raise AirflowSkipException("No data")
        null_rate = df.isnull().sum().sum() / (df.shape[0] * df.shape[1])
        if null_rate > 0.3: raise ValueError(f"Data quality too low: {null_rate:.1%}")
        return {"row_count": len(df), "null_rate": null_rate, "status": "transformed"}

    @task
    def load_with_idempotency(transformed: dict) -> dict:
        import psycopg2
        from airflow.hooks.base import BaseHook
        conn = BaseHook.get_connection("postgres_warehouse")
        pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
        cur = pg_conn.cursor()
        cur.execute("DELETE FROM analytics.daily_metrics WHERE metric_date = CURRENT_DATE - INTERVAL '1 day'")
        cur.execute("INSERT INTO analytics.daily_metrics (metric_date, row_count, null_rate, status) VALUES (CURRENT_DATE - INTERVAL '1 day', %s, %s, %s)", (transformed["row_count"], transformed["null_rate"], transformed["status"]))
        pg_conn.commit(); pg_conn.close()
        return {"loaded": True, "inserted": 1}

    def sla_check(**context):
        import pendulum
        execution_date = context["execution_date"]
        if pendulum.now("UTC") > execution_date.add(hours=2):
            raise ValueError("SLA violated")
        return "SLA_OK"

    sla_check_task = PythonOperator(task_id="sla_check", python_callable=sla_check)
    sla_alert_task = PythonOperator(task_id="sla_alert", python_callable=lambda **c: print(f"SLA Alert: {c['dag'].dag_id}"), trigger_rule="one_failed")

    raw = extract_with_retry()
    transformed = transform_with_validation(raw)
    loaded = load_with_idempotency(transformed)
    loaded >> sla_check_task >> sla_alert_task

error_handling_sla_pipeline()

# Airflow 2.10数据感知调度
from airflow.datasets import Dataset

USER_EVENTS = Dataset("postgres://warehouse/analytics/user_events_daily")
ORDER_METRICS = Dataset("postgres://warehouse/analytics/order_metrics_daily")

@dag(schedule="0 2 * * *")
def producer():
    @task(outlets=[USER_EVENTS])
    def produce_user_events(): return "produced"
    @task(outlets=[ORDER_METRICS])
    def produce_order_metrics(): return "produced"
    produce_user_events(); produce_order_metrics()

@dag(schedule=[USER_EVENTS, ORDER_METRICS])  # 两个Dataset都更新后触发
def consumer():
    @task
    def generate_daily_report(): return "report_generated"
    generate_daily_report()

パターン5：エラーハンドリング、リトライ、SLAモニタリング

# ❌ 每次Scheduler解析都会执行查询
sources = requests.get("https://api.example.com/sources").json()
@dag(...)
def my_dag():
    @task
    def process():
        for source in sources: ...

# ✅ 动态逻辑放在Task内部
@dag(...)
def my_dag():
    @task
    def get_sources() -> list:
        return requests.get("https://api.example.com/sources").json()
    @task
    def process(sources: list):
        for source in sources: ...

5つの落とし穴ガイド

1. DAGファイルトップレベルコードでのデータベースクエリ実行

❌ 誤ったやり方：DAGファイルのトップレベルでデータベースクエリやAPI呼び出しを実行、SchedulerがDAGを解析するたびに実行される

✅ 正しいやり方：動的ロジックをTask内に配置し、DAGファイルのトップレベルは構造定義のみ

# ❌ DataFrame通过XCom传递
@task
def extract():
    df = pd.read_csv("huge_file.csv")
    return df.to_dict()  # XCom序列化失败

2. XComでの大容量データオブジェクト連携

❌ 誤ったやり方：DataFrame全体をXComで連携、48KBを超えるとエラー

✅ 正しいやり方：ファイルパス参照を連携、またはカスタムXCom Backendを使用

# ✅ 传递文件路径引用
@task
def extract():
    df = pd.read_csv("huge_file.csv")
    path = f"/tmp/extract_{datetime.now().strftime('%Y%m%d')}.parquet"
    df.to_parquet(path)
    return {"file_path": path, "row_count": len(df)}

@task
def transform(metadata: dict):
    df = pd.read_parquet(metadata["file_path"])
    # 处理逻辑...

3. catchup=Trueによる歴史バックフィルの雪崩

❌ 誤ったやり方：catchup=Trueでstart_dateが1年前のDAGが起動時に365個のDAG Runをバックフィル

✅ 正しいやり方：プロダクションDAGは常にcatchup=False、バックフィル時はairflow dags backfillで手動制御

4. SensorのWorker Slot占有によるデッドロック

❌ 誤ったやり方：複数のSensorが同時実行され、各「がWorker Slotを占有

✅ 正しいやり方：mode="reschedule"またはdeferrableモードを使用してWorker Slotを解放

# ✅ reschedule模式
FileSensor(task_id="wait", filepath="/data/{{ ds }}.csv", mode="reschedule")
# ✅ deferrable模式（Airflow 2.2+推荐）
FileSensor(task_id="wait", filepath="/data/{{ ds }}.csv", deferrable=True)

5. データ品質ゲートの無視

❌ 誤ったやり方：DAGがタスクの実行成功のみをチェック、空のデータでも緑表示

✅ 正しいやり方：DAGの末尾に品質ゲートを追加、データ量や品質が基準を満たさない場合は下流をブロック

@task
def quality_gate(loaded_count: int, expected_min: int = 1000) -> str:
    if loaded_count < expected_min:
        raise ValueError(f"Quality gate failed: {loaded_count} < {expected_min}")
    return "PASSED"

10のエラートラブルシューティング

#	エラーメッセージ	原因	解決策
1	`DAG import error`	Worker環境にPython依存が不足	`requirements.txt`に依存を追加、Workerを再起動
2	`XCom value size exceeded 48KB`	XComデータがデフォルト48KB制限を超過	ファイルパス参照を連携、またはカスタムXCom Backend（Redis）を設定
3	`Pool has no free slots`	リソースプールが満杯、タスクが待機	Poolのslot数を増加：`airflow pool set default_pool 128`
4	`DAG seems missing from DagBag`	DAGファイルに構文エラーや循環インポート	`airflow dags list-import-errors`で確認
5	`Sensor timeout`	Sensorの待機条件が常に満たされない	データソースを確認、timeoutを増加、deferrableモードを使用
6	`AirflowTaskTimeout`	タスク実行時間が`execution_timeout`を超過	timeoutを増加、サブタスクに分割、コードを最適化
7	`Detected deadlock`	循環依存やリソース競合	DAG依存グラフを確認、Pool設定を確認
8	`Connection refused`	データベースやAPI接続の失敗	Connection設定を確認：`airflow connections get-uri`
9	`Dynamic task mapping too many tasks`	展開されたタスク数が`max_map_length`制限を超過	`airflow.cfg`の`max_map_length`を調整（デフォルト1024）
10	`SLA missed`	タスクがSLA期限内に完了しなかった	上流の遅延を確認、パフォーマンスを最適化、SLA時間窓を調整

高度な最適化テクニック

1. DAG解析パフォーマンス最適化

@dag(
    dag_id="optimized_pipeline",
    dagrun_timeout=timedelta(hours=3),  # DAG Run最多3小时
    max_active_runs=1,  # 同时只允许1个DAG Run
    max_active_tasks=16,  # 最多16个并发Task
)
def optimized_pipeline(): ...

2. 接続プールとバッチ操作

from airflow.hooks.base import BaseHook
import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection(connection_id: str):
    conn_info = BaseHook.get_connection(connection_id)
    conn = psycopg2.connect(host=conn_info.host, port=conn_info.port, database=conn_info.schema, user=conn_info.login, password=conn_info.password)
    try: yield conn
    finally: conn.close()

def batch_insert(conn, table: str, columns: list[str], rows: list[tuple], batch_size: int = 5000):
    from psycopg2.extras import execute_values
    with conn.cursor() as cur:
        for i in range(0, len(rows), batch_size):
            execute_values(cur, f"INSERT INTO {table} ({', '.join(columns)}) VALUES %s", rows[i:i + batch_size])
    conn.commit()

3. Datasetデータ認識スケジューリングでCronを置き換え

from airflow.datasets import Dataset
RAW_DATA = Dataset("s3://data-lake/raw/events/")
CLEANED_DATA = Dataset("s3://data-lake/cleaned/events/")

@dag(schedule="0 2 * * *")
def producer():
    @task(outlets=[RAW_DATA])
    def extract(): ...
    @task(outlets=[CLEANED_DATA])
    def clean(): ...

@dag(schedule=[CLEANED_DATA])  # 数据就绪后自动触发
def consumer():
    @task
    def train_model(): ...

4. 優先度とリソース分離

# 关键DAG使用高优先级和专用Pool
@dag(dag_id="critical_pipeline", priority_weight=100, pool="revenue_pool")
def critical_pipeline(): ...

# 非关键DAG使用低优先级
@dag(dag_id="experimental_pipeline", priority_weight=1, pool="default_pool")
def experimental_pipeline(): ...

5. モニタリング指標とアラート統合

# airflow.cfg 开启StatsD指标导出
# [metrics]
# statsd_on = True
# statsd_host = localhost
# statsd_port = 8125

def sla_callback(dag, task_list, blocking_task_list, dagrun, *args, **kwargs):
    import requests
    from airflow.models import Variable
    webhook_url = Variable.get("slack_webhook_url")
    message = {"text": f":warning: SLA Violation\nDAG: {dag.dag_id}\nRun: {dagrun.execution_date}\nBlocking: {[t.task_id for t in blocking_task_list]}"}
    requests.post(webhook_url, json=message)

default_args = {"sla": timedelta(hours=2), "sla_miss_callback": sla_callback}

オーケストレーションツール比較

尺度	Apache Airflow	Prefect	Dagster	Luigi
オーケストレーション	DAG定義	関数型+デコレータ	アセット定義	タスク依存
データ認識スケジューリング	Dataset (2.4+)	ネイティブ	ネイティブAsset	非サポート
ダイナミックタスク	Dynamic Task Mapping	ネイティブ	ネイティブ	非サポート
カスタムOperator	強力なプラグインエコシステム	デコレータベース	Software-defined Asset	シンプルなTaskクラス
XCom/データ連携	48KB制限、カスタムBackend可能	ネイティブ無制限	ネイティブTypeシステム	シンプルなパラメータ連携
エラー処理	リトライ+SLA+コールバック	ネイティブリトライ+状態管理	ネイティブリトライ+Asset状態	シンプルなリトライ
モニタリング	Web UI + StatsD	Cloud UI	Dagit UI	基本Web
学習曲線	中	低	中高	低
コミュニティ	最も成熟、プラグイン最多	急速成長中	成長中	衰退
Kubernetes	KubernetesPodOperator	ネイティブK8s Agent	ネイティブK8s	非サポート
2026年推奨度	★★★★★	★★★★	★★★★	★★
ユースケース	エンタープライズETL	中小プロジェクト	データアセット密集型	シンプルバッチ処理