データエンジニアリング

データパイプラインで2年間ハマり続けた話｜Airflow・dbt・Kafka本番運用の現実

「昨日のデータが空ですよ」——その絶望、経験ありませんか？2年間の本番運用で踏んだ失敗とAirflow・dbt・Kafka設計の改善パターンを包み隠さず公開します。

データエンジニアリングデータパイプラインAirflowdbtKafka

データパイプライン2026年版｜本番運用2年で痛感したアーキテクチャ設計の現実

正直に言うと、2年前にデータパイプラインを「ちゃんと設計しよう」と意気込んで始めた当初、こんなにハマるとは思ってなかった。最初のバッチ処理が本番でサイレントフェイルして、翌朝に分析チームから「昨日のデータが空ですよ」って報告が来たときの絶望感はいまでも覚えてる。

その後2年間、Airflow・dbt・Kafka・Sparkといったエコシステムを組み合わせながらパイプラインを少しずつ改善してきた。2026年現在、ツールの成熟度が急速に上がっていて、かつてハマった問題の多くは「最初からこう設計すればよかった」と後から分かるパターンに収斂してきた。その知見を包み隠さずシェアしたい。

皆さんのチームはデータパイプラインの障害対応、どれくらいの頻度で発生してますか？

パイプラインアーキテクチャの全体像：2026年時点のスタンダード構成

2年前は「とりあえずAirflowで全部書く」スタイルだったが、今はもう少し責務を分離するのが当たり前になってきた。バッチとストリームを明示的に分け、変換ロジックはdbtに委ね、データ品質チェックをパイプラインの中に組み込む。構成図にすると、こんな感じだ。

flowchart TB
    subgraph Sources["データソース"]
        DB[(OLTP DB)]
        API["外部API"]
        EVT["アプリイベント"]
    end

    subgraph Ingestion["取り込み層"]
        CDC["CDC / Debezium"]
        KFK["Kafka Streams"]
        AIRBYTE["Airbyte 1.x"]
    end

    subgraph Storage["ストレージ層"]
        RAW[("Raw Layer\nS3 / GCS")]
        ICE[("Iceberg Table")]
    end

    subgraph Transform["変換層"]
        DBT["dbt 2.0"]
        SPARK["Spark 4.0"]
    end

    subgraph Orchestration["オーケストレーション"]
        AF["Airflow 3.0"]
        DLT["dlt (data load tool)"]
    end

    subgraph Serving["サービング層"]
        DWH[("Redshift / BigQuery")]
        BI["BI / Metabase"]
    end

    DB --> CDC
    API --> AIRBYTE
    EVT --> KFK

    CDC --> RAW
    AIRBYTE --> RAW
    KFK --> ICE

    RAW --> DBT
    ICE --> SPARK
    DBT --> DWH
    SPARK --> DWH

    AF --> DBT
    AF --> SPARK
    DLT --> RAW

    DWH --> BI

この構成で特に変化を感じるのが Airflow 3.0 と dbt 2.0 の登場だ。Airflow 3.0はDAGのシリアライゼーションが大幅に改善され、Executorの選択肢も増えた。dbt 2.0はモデルのバージョニングとユニットテストが正式サポートされ、「dbtで品質保証まで完結できる」という感覚が実感として得られるようになった（dbt 1.9→2.0移行の苦労話はこちらでも書いた）。

一方、ストリーム側はKafka Streamsを中心に据えつつ、Apache Icebergをストレージフォーマットとして採用することでバッチ・ストリームの境界を薄くするアプローチが有効だと分かってきた。これはいわゆる 「Lambda アーキテクチャの終焉」 とも言える流れで、個人的にはかなり正解に近いと思っている。

実際に踏んだ地雷：サイレントフェイルとデータスキュー

パイプライン設計で一番怖いのはエラーで止まることじゃなくて、「何も言わず間違ったデータが流れ続けること」 なんですよね。うちのチームで実際に経験した事例を2つ紹介する。

ケース1：NULL伝播によるサイレントフェイル

dbtモデルの結合ロジックで外部キーがNULLのレコードを意図せずフィルタしてしまい、売上集計が毎日5〜8%ほど低めに出ていた。3週間気づかなかった。発覚したのは分析チームが「なんか先月と数字の雰囲気が違う」と言い出したのがきっかけで、原因特定に丸1日かかった。

-- bad: NULL行が静かに消える
SELECT
    o.order_id,
    c.customer_name,
    o.amount
FROM orders o
JOIN customers c ON o.customer_id = c.id  -- NULLはここで消える

-- good: LEFT JOINで漏れを可視化 + dbtテストでNULL率を監視
SELECT
    o.order_id,
    c.customer_name,
    o.amount,
    CASE WHEN c.id IS NULL THEN TRUE ELSE FALSE END AS is_orphan
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.id

dbt 2.0のunit testで事前にこういう挙動を検証できるようになったのは地味に大きい。「テストを書く」ことへの心理的ハードルが下がっただけで、品質が目に見えて変わった。

# dbt 2.0 unit test
unit_tests:
  - name: test_null_customer_preserved
    model: orders_enriched
    given:
      - input: ref('orders')
        rows:
          - {order_id: 1, customer_id: null, amount: 1000}
    expect:
      rows:
        - {order_id: 1, is_orphan: true}

ケース2：Sparkのデータスキューで特定パーティションだけ3倍の処理時間

Sparkジョブでuser_idでパーティショニングしていたんだけど、一部の大口ユーザーがデータ量の30%以上を占めていて、そのパーティションだけ処理が異常に遅かった。AQEが有効でもスキューが極端すぎると自動では解消されなかった。これは正直、「AQEがあれば大丈夫」と油断していたのが原因だった。

# Spark 4.0 でのスキュー対策
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, rand

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB") \
    .getOrCreate()

# スキューが酷い場合はsalt keyで手動分散
def add_salt(df, salt_num=10):
    return df.withColumn("salt", (rand() * salt_num).cast("int"))

# 大口ユーザーだけ別処理
heavy_users = transactions.filter(col("user_id").isin(known_heavy_users))
normal_users = transactions.filter(~col("user_id").isin(known_heavy_users))

heavy_result = add_salt(heavy_users).repartition(50, "salt", "user_id")
normal_result = normal_users.repartition("user_id")

result = heavy_result.unionByName(normal_result, allowMissingColumns=True)

SparkのAQEやチューニング手法についてはApache Spark 2026年最新動向に詳しくまとまってるので参考にしてほしい。

2026年のツール選定：バッチ・ストリーム・オーケストレーター比較

正直まだ検証中のものも含まれるけど、現時点での選定基準を整理する。「どれが一番いいの？」という質問をよく受けるが、ぶっちゃけチームの規模とPythonへの習熟度で答えが変わる。

オーケストレーターの比較

ツールバージョン主なユースケース学習コストスケーラビリティ2026年の所感
Apache Airflow3.0複雑なバッチDAG高高Executor改善で運用負荷↓
Prefect3.xPython中心のチーム中中UIが使いやすくなった
Dagster1.9アセット指向パイプライン中高dbtとの統合が秀逸
dlt (data load tool)1.x軽量インジェスト低低〜中小規模では最速
Temporal2.x耐障害性重視高高ワークフローエンジン寄り

うちのチームはAirflow 3.0をメインに使っているが、Dagster + dbt の組み合わせは「アセット指向」という概念がデータエンジニアリングと相性が良く、新規プロジェクトではDagsterを採用している。Airflowはタスクの依存グラフを書くが、Dagsterはデータアセットの依存グラフを書く——この発想の違いが、運用中の見通しの良さにつながっている。

ストリーム処理の比較

ツール強み弱み月間コスト目安
Kafka StreamsKafkaとの親和性独立スケーリング難インフラ次第
Apache Flink 2.0低レイテンシ・ステートフル運用複雑高め
Spark Structured Streamingバッチとの統一レイテンシ数秒〜中
Redpanda + KSQL軽量・高速成熟度やや低い低め

ストリーム処理の選択は本当にユースケース次第で、「ほぼリアルタイム（数秒以内）」が要件なら Flink か Kafka Streams 一択。数分単位で良ければSpark Structured Streamingで済むことが多い。イベント駆動アーキテクチャの実装パターンも参考になるかも。

データ品質チェックをパイプラインに組み込む実装パターン

「データ品質は別チームが見る」時代は終わったと感じている。パイプラインの各ステージにバリデーションを組み込み、問題が起きたら自動で止める・アラートを出す仕組みが2026年のスタンダードだ。フローで表すとこうなる。

flowchart LR
    RAW["Raw Data"] --> V1{"Schema Check"}
    V1 -->|Pass| CLEAN["Staging"]
    V1 -->|Fail| DQ_ALERT1["Alert & DLQ"]
    CLEAN --> V2{"Volume Check\n±20% from yesterday"}
    V2 -->|Pass| TRANSFORM["Transform (dbt)"]
    V2 -->|Fail| DQ_ALERT2["Quarantine"]
    TRANSFORM --> V3{"Metric Sanity\nCheck"}
    V3 -->|Pass| SERVE["Data Warehouse"]
    V3 -->|Fail| DQ_ALERT3["Rollback"]

    style DQ_ALERT1 fill:#f66,color:#fff
    style DQ_ALERT2 fill:#f66,color:#fff
    style DQ_ALERT3 fill:#f66,color:#fff

実装は Great Expectations 1.x か dbt tests のどちらかをメインにする場合が多いが、個人的には最近 dbt契約テスト + Soda Core の組み合わせが気に入っている。Soda CoreのチェックはYAMLで書けるので、SQLが苦手なアナリストでも読めるし、レビューもしやすい。

# Soda Core によるカスタム品質チェック
# checks.yml
checks for orders_daily:
  - row_count > 1000:
      name: "最低件数チェック"
      fail: when < 500
      warn: when < 1000
  - missing_count(customer_id) = 0:
      name: "customer_id NULLチェック"
  - duplicate_count(order_id) = 0:
      name: "order_id重複チェック"
  - avg(amount) between 500 and 50000:
      name: "金額平均異常検知"
# Airflow DAGにSodaチェックを組み込む
from airflow.operators.python import PythonOperator
from soda.scan import Scan

def run_soda_check(data_source: str, checks_file: str):
    scan = Scan()
    scan.set_verbose()
    scan.add_configuration_yaml_file("soda_config.yml")
    scan.set_data_source_name(data_source)
    scan.add_sodacl_yaml_file(checks_file)
    
    exit_code = scan.execute()
    
    if exit_code != 0:
        raise ValueError(f"データ品質チェック失敗: {scan.get_logs_text()}")
    
    return scan.get_scan_results()

with dag:
    quality_check = PythonOperator(
        task_id="quality_check_orders",
        python_callable=run_soda_check,
        op_kwargs={
            "data_source": "redshift_prod",
            "checks_file": "checks/orders_daily.yml"
        }
    )
    
    transform_task >> quality_check >> serving_task

データ品質管理の体系的な話はデータ品質管理2026年版でもまとめているのでぜひ。

パフォーマンスの実測値と改善インパクト

「設計変更でどれくらい改善したか」の数字を持っておくと、社内の説明が楽になる。うちのチームで実際に計測した結果をまとめる。

xychart-beta
    title "パイプライン処理時間の改善推移（分）"
    x-axis ["2024Q1", "2024Q3", "2025Q1", "2025Q3", "2026Q1"]
    y-axis "処理時間（分）" 0 --> 180
    bar [165, 142, 98, 67, 41]
    line [165, 142, 98, 67, 41]

各フェーズで何をやったかも書いておく。

  • 2024Q3：Spark AQE有効化 + パーティション最適化 → 約14%削減
  • 2025Q1：バッチ→ストリーム処理への一部移行 + Iceberg採用 → 約31%削減
  • 2025Q3：dbt incremental model最適化 + データスキュー対応 → 約32%削減
  • 2026Q1：Airflow 3.0移行 + EMR Serverless本格活用 → 約39%削減

最終的に165分かかっていたパイプラインが41分になったのは素直に嬉しかったが、正直「もっと早く設計を見直せばよかった」という気持ちも強い。特にIncremental Modelへの移行は初期コストが高いと思っていたが、やってみたら想定の半分の時間でできた。先送りしてた時間が一番もったいなかった。

障害件数の改善も数字にしておく。

xychart-beta
    title "月間パイプライン障害件数の推移"
    x-axis ["2024Q1", "2024Q3", "2025Q1", "2025Q3", "2026Q1"]
    y-axis "障害件数" 0 --> 25
    bar [22, 18, 14, 8, 3]

障害件数は22件/月から3件/月まで落ちた。データ品質チェックの組み込みと、DLQパターン（Dead Letter Queue）の整備が一番効いた。分散バッチアーキテクチャの設計パターンにも関連する内容が詳しく書いてある。

2026年の現実的なパイプライン設計チェックリスト

「どこから始めればいいか分からない」という相談を受けることが増えてきたので、チームで実際に使っているチェックリストを公開する。完全ではないし、好みが分かれる項目もあると思う。ただ、最低限これだけ抑えれば初期の地雷は踏みにくくなる、という経験則から作ったものだ。

取り込み層

  • べき等性（Idempotency）が担保されているか
  • スキーマ変更に対応できるか（Schema Registry or dlt）
  • データ量の急増に対してバックプレッシャーが機能するか
  • タイムゾーンの統一（UTCで管理しているか）

変換層

  • Incremental Modelになっているか（Full Refreshは最終手段）
  • NULLの扱いが明示的に定義されているか
  • dbt testがStagingとMartの両方に存在するか
  • コードレビューの仕組みがあるか（PRベース）

オーケストレーション

  • 再試行ロジックが実装されているか（exponential backoff）
  • タイムアウトが設定されているか
  • 依存関係が明確に定義されているか（upstream/downstream）
  • 障害通知先が設定されているか（Slack/PagerDuty）
# Airflow 3.0 でのリトライ設定例
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import timedelta

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,  # Airflow 3.0 から安定
    "max_retry_delay": timedelta(minutes=60),
    "execution_timeout": timedelta(hours=2),
    "on_failure_callback": send_slack_alert,
}

with DAG(
    dag_id="orders_pipeline",
    default_args=default_args,
    schedule="0 2 * * *",
    catchup=False,
    tags=["orders", "critical"],
) as dag:
    pass

タイムゾーン問題は本当に細かいようで本番で死ぬので念を入れておきたい。Lake Formation本番導入でのタイムゾーン落とし穴は他人事じゃない話だった。

まとめ

2年間でデータパイプラインについて学んだことの核心は、「ツールより設計原則」 だったと思う。AirflowだろうがDagsterだろうが、べき等性・障害検知・インクリメンタル処理という基本を疎かにすると必ず後で火を噴く。

要点まとめ：

  1. サイレントフェイル対策が最優先：エラーで止まるより、間違ったデータが静かに流れ続ける方が怖い。dbt testとSoda Coreをパイプラインに組み込む
  2. バッチとストリームの境界をIcebergで薄くする：Lambda Architectureの二重管理地獄から解放される
  3. Incremental Modelへの移行は早めに：後から変えるコストは初期導入コストの数倍になる
  4. データスキューは定量的に把握して対処：AQEだけでは解消できないケースが必ずある
  5. Airflow 3.0 or Dagsterへの移行は検討する価値あり：Airflow 2系の運用負荷は意外と高かった

次のアクション：

  • 既存パイプラインのIncremental Model適用率を計測する
  • dbt unit testを1モデルに試験導入する
  • Soda CoreかGreat Expectationsをどちらか選んで品質チェックを1箇所入れる

「うちはこう解決した」「この選択は間違いだった」など、コメントやフィードバックもらえると嬉しい。データエンジニアリングはまだ正解が固まりきっていない領域なので、実務の知見を共有し合うのが一番早いと思ってる。

U

Untanbaby

ソフトウェアエンジニア｜AWS / クラウドアーキテクチャ / DevOps

10年以上のIT実務経験をもとに、現場で使える技術情報を発信しています。 記事の誤りや改善点があればお問い合わせからお気軽にご連絡ください。

関連記事

データエンジニアリング

dbt Core 2.0移行後1年、本番運用で積み上がった設計の知見

データエンジニアリング

Kafkaで本番止めた話｜ストリーム処理2年運用の失敗記録

データエンジニアリング

A/Bテスト基盤を自前構築して1年半、失敗から学んだデータエンジニアリングの現実

データエンジニアリング

dbt 1.9→2.0移行で半年ハマった話｜本番で気づいたモデル設計の現実

データエンジニアリング

本番で売上データが150%跳ね上がった日、データ品質管理と向き合った話

データエンジニアリング

データカタログ3回失敗して分かった、ようやく「使われる」ようになった話

← 記事一覧に戻る