C データエンジニアリング — dbt microbatch incremental（event_time + lookback=3）× BigQuery MERGE 冪等 UPSERT × 遅延到着データ対応 × dbt source freshness × Argo Workflows DAG（dependencies + retryStrategy + onExit）（MOps 注文集計マート Bad→Good）

📋 コピー# models/staging/_sources.yml
# 修正④: source freshness で上流の取り込み遅延を検知
version: 2

sources:
  - name: events
    database: my-gcp-project
    schema: events
    # ソース全体のデフォルト鮮度基準
    freshness:
      warn_after:  {count: 6,  period: hour}   # 6時間遅延で警告
      error_after: {count: 25, period: hour}   # 25時間でエラー（日次跨ぎ許容）
    loaded_at_field: ordered_at                # 鮮度判定に使う列

    tables:
      - name: order_events
        description: "注文確定イベント（最大3日 late-arriving あり）"
        # 遅延到着があるため error_after を緩めに上書き
        freshness:
          warn_after:  {count: 12, period: hour}
          error_after: {count: 72, period: hour}  # 3日 late-arriving を許容
        columns:
          - name: order_id
            tests: [not_null, unique]
          - name: channel
            tests:
              - accepted_values:
                  values: ["web", "app", "store"]

📋 コピー# argo-cron-daily-orders.yaml
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: daily-orders-mart
  namespace: mops
spec:
  schedule: "0 20 * * *"          # UTC 20:00 = JST 05:00
  timezone: "Asia/Tokyo"
  concurrencyPolicy: Forbid        # 前回未完了なら新規起動しない
  startingDeadlineSeconds: 300
  workflowSpec:
    entrypoint: main
    serviceAccountName: dbt-runner  # WIF で BigQuery 認証
    onExit: notify-slack            # 修正⑦: 終了時に必ず実行
    arguments:
      parameters:
        - name: dbt-target
          value: "prod"

    templates:
      # ── DAG 定義（修正⑤: 依存を明示）──
      - name: main
        dag:
          tasks:
            - name: stg-order-events
              template: dbt-run
              arguments:
                parameters: [{name: select, value: "stg_order_events"}]

            - name: mart-daily-orders
              template: dbt-run
              dependencies: [stg-order-events]   # staging 完了後
              arguments:
                parameters: [{name: select, value: "mart_daily_orders"}]

            - name: freshness-check
              template: dbt-source-freshness
              dependencies: [mart-daily-orders]  # マート完了後

      # ── dbt run（修正⑥: retryStrategy）──
      - name: dbt-run
        inputs:
          parameters: [{name: select}]
        retryStrategy:
          limit: 3
          retryPolicy: "OnError"      # システムエラーのみリトライ
          backoff:
            duration: "30s"
            factor: 2                 # 30s → 60s → 120s
            maxDuration: "5m"
        container:
          image: .../mops/dbt-runner@sha256:abc123
          command: ["dbt"]
          args: ["run", "--select", "{{inputs.parameters.select}}",
                 "--target", "{{workflow.parameters.dbt-target}}"]

      - name: dbt-source-freshness
        retryStrategy:
          limit: 2
          backoff: {duration: "20s", factor: 2, maxDuration: "2m"}
        container:
          image: .../mops/dbt-runner@sha256:abc123
          command: ["dbt"]
          args: ["source", "freshness",
                 "--target", "{{workflow.parameters.dbt-target}}"]

      # ── 修正⑦: Exit handler（Slack 通知）──
      - name: notify-slack
        container:
          image: .../mops/slack-notifier@sha256:def456
          command: ["python", "/app/notify.py"]
          env:
            - name: WORKFLOW_STATUS
              value: "{{workflow.status}}"     # Succeeded / Failed / Error
            - name: WORKFLOW_NAME
              value: "{{workflow.name}}"
            - name: FAILURES
              value: "{{workflow.failures}}"