data-pipelines

データパイプラインスキル

バージョン： 1.0 スタック： Airflow、Step Functions、dbt、一般的な ETL

パイプラインの障害は、気付かれないまま発生します。冪等性のないタスクは、再試行のたびに重複した行を追加します。検証のないタスクは不正なデータを下流に渡し、ダッシュボードにあり得ない数値が表示されて、3 段階後にようやく問題に気付きます。日付をパラメーター化していないパイプラインではバックフィルできません。そのため、火曜日に問題が起きると、最後に正常だった状態以降のすべての日を手作業で再処理することになります。

冪等なタスク、ステージ間の品質チェック、パラメーター化された実行により、障害から復旧でき、問題が下流ではなく発生した場所で検出されます。

---

対象範囲と境界

このスキルの対象：

• ETL/ELT アーキテクチャの選択
• オーケストレーションパターン（DAG、依存関係）
• 冪等性と再試行戦略
• データ品質チェック
• 増分更新と全件更新
• バックフィルパターン

他のスキルに委ねる内容：

• data-python：タスク内の Python コード
• data-sql：変換処理でのクエリパターン
• data-aws：AWS 固有のサービスパターン

このスキルを使う場面： データパイプラインを設計または構築するとき。

---

基本原則

1. 冪等なタスク — 2 回実行しても同じ結果になります。
2. アトミックな操作 — 全件実行か未実行かのどちらかであり、部分的な書き込みはありません。
3. データコントラクト — ステージ間のスキーマ要件を定義します。
4. 可観測なパイプライン — 各ステージでログ、メトリクス、アラートを記録します。
5. 可能な限り増分処理 — 変更されたデータだけを処理します。

---

パターン

冪等な書き込み

# Bad - appends duplicate data on retry
def load_data(df, table):
    df.to_sql(table, engine, if_exists="append")

# Good - replace partition, safe to retry
def load_data(df, table, partition_date):
    with engine.begin() as conn:
        conn.execute(f"DELETE FROM {table} WHERE date = '{partition_date}'")
        df.to_sql(table, conn, if_exists="append", index=False)

ステージパターン

raw/           # Exact copy of source, immutable
├── source_a/
└── source_b/

staged/        # Cleaned, typed, validated
├── source_a/
└── source_b/

curated/       # Business logic applied, joined
├── dim_users/
├── dim_products/
└── fact_orders/

aggregated/    # Pre-computed metrics
├── daily_revenue/
└── user_cohorts/

データ品質チェック

def validate_output(df: pd.DataFrame) -> None:
    """Run after transform, before load."""
    assert not df.empty, "Output is empty"
    assert df["id"].is_unique, "Duplicate IDs found"
    assert df["amount"].notna().all(), "NULL amounts found"
    assert (df["amount"] >= 0).all(), "Negative amounts found"

---

アンチパターン

アンチパターン	問題	対策
冪等性のないタスク	再試行時に重複が発生する	DELETE してから INSERT するか、upsert を使う
モノリシックな DAG	デバッグが難しく、復旧に時間がかかる	小規模で目的を絞った DAG に分割する
データ検証がない	不正なデータが気付かれないまま伝播する	ステージ間にチェックを追加する
日付のハードコード	バックフィルできない	実行日をパラメーター化する
サイレントな障害	問題の発見が遅すぎる	エラーだけでなく、異常もアラートする

---

チェックリスト

• すべてのタスクが冪等である
• ステージ間にデータ品質チェックがある
• 実行日がパラメーター化されている
• バックフィルをテストし、文書化している
• 障害と異常の両方でアラートする
• 明確なデータリネージュを文書化している

---

参考資料

• references/idempotency.md — 冪等な書き込みパターンとテスト

アセット

• assets/pipeline-checklist.md — 包括的なパイプライン設計チェックリスト