データエンジニアリング — Argo Workflows 並列化 × dbt 品質ゲート × リトライ戦略

概要

問題

毎朝5時に実行される日次販促データパイプラインで以下の問題が発生しています。

要件

1. 並列化: staging内の3モデルを並列実行し、全て完了してからmartsを実行
2. データ品質ゲート: staging実行後にdbt testを実行し、テスト失敗時はmartsへ進まない
3. リトライ設定: 各ステップに最大3回のリトライ（指数バックオフ: 2分〜）
4. タイムゾーン修正: JST 05:00に正しく実行されるよう修正

現状の YAML（問題あり）

📋 コピーapiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: promo-pipeline-daily
spec:
  schedule: "0 5 * * *"
  timezone: "UTC"          # ← 問題: UTCのためJST 14:00に実行される
  workflowSpec:
    entrypoint: pipeline
    templates:
      - name: pipeline
        steps:
          - - name: extract
              template: bq-extract
          - - name: transform-a
              template: dbt-run-staging  # ← 問題: staging全体を1ステップで直列
          - - name: transform-b
              template: dbt-run-marts    # ← 問題: testなしにmartsへ進む
          - - name: load-report
              template: export-to-gcs

      - name: bq-extract
        container:              # ← 問題: retryStrategyなし
          image: gcr.io/project/pipeline:latest
          command: [python, extract.py]

      - name: dbt-run-staging
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, staging]

      - name: dbt-run-marts
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, marts]

      - name: export-to-gcs
        container:
          image: gcr.io/project/pipeline:latest
          command: [python, export.py]

ヒント（段階的開示）

steps:
  - - name: step-a
  - - name: step-b  # 直列
  - - name: step-c  # 直列

steps:
  - - name: step-a  # 同じリスト内
    - name: step-b  # = 並列実行
    - name: step-c  # = 並列実行

retryStrategy:
  limit: "3"
  retryPolicy: "Always"
  backoff:
    duration: "2m"
    factor: "2"
    maxDuration: "10m"

# タイムゾーン修正
schedule: "0 5 * * *"
timezone: "Asia/Tokyo"   # JSTのまま書ける

パイプライン構造図 — 改善後のDAGフロー

模範解答（改善後 YAML 全文）

📋 コピーapiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: promo-pipeline-daily
spec:
  schedule: "0 5 * * *"
  timezone: "Asia/Tokyo"   # 修正①: UTC→Asia/Tokyo でJST 05:00実行
  workflowSpec:
    entrypoint: pipeline
    templates:
      # ── メインフロー ──────────────────────────────
      - name: pipeline
        steps:
          # Step1: 抽出（直列）
          - - name: extract
              template: bq-extract

          # Step2: staging 3モデルを並列実行（修正②）
          - - name: stg-orders
              template: dbt-stg-orders
            - name: stg-customers
              template: dbt-stg-customers
            - name: stg-promotions
              template: dbt-stg-promotions

          # Step3: データ品質ゲート（修正③）- 失敗したらここで止まる
          - - name: quality-gate
              template: dbt-test-staging

          # Step4: martsはstagingテスト通過後のみ実行
          - - name: transform-marts
              template: dbt-run-marts

          # Step5: レポート出力
          - - name: load-report
              template: export-to-gcs

      # ── 抽出 ─────────────────────────────────────
      - name: bq-extract
        retryStrategy:   # 修正④: 全ステップにリトライ設定
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/pipeline:latest
          command: [python, extract.py]

      # ── staging 並列ステップ ───────────────────────
      - name: dbt-stg-orders
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, stg_orders]

      - name: dbt-stg-customers
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, stg_customers]

      - name: dbt-stg-promotions
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, stg_promotions]

      # ── データ品質ゲート ──────────────────────────
      - name: dbt-test-staging
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, test, --select, staging]

      # ── marts ─────────────────────────────────────
      - name: dbt-run-marts
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/dbt:latest
          command: [dbt, run, --select, marts]

      # ── レポート出力 ──────────────────────────────
      - name: export-to-gcs
        retryStrategy:
          limit: "3"
          retryPolicy: "Always"
          backoff:
            duration: "2m"
            factor: "2"
            maxDuration: "10m"
        container:
          image: gcr.io/project/pipeline:latest
          command: [python, export.py]

各変更点の説明

ポイント解説

実務への応用

• 毎朝5時の販促効果集計バッチはまさにこの構成で動いている
• staging並列化により、従来18分かかっていた処理が最長ステップ（例: stg_orders 8分）まで短縮 → レポート完成がJST 5:13頃（従来5:23頃）に
• データ品質ゲートに dbt test --select staging --store-failures を追加すると、失敗レコードがBigQueryテーブルに保存されデバッグが容易になる
• Argo Workflows の onExit テンプレートにSlack/PagerDuty通知を追加することで、品質ゲート失敗時に即時アラートが飛ぶ

今日のまとめ

自己評価

自分の回答

気づき・メモ