C データエンジニアリング — BigQuery CVR 集計 99%コスト削減 × dbt 1.8+ Unit Tests × incremental insert_overwrite

概要

問題

ECサイトの Argo Workflows で夜間に実行されるデータパイプラインが、dbt のモデル更新ステップで毎回フルリフレッシュしており、BigQuery の処理コストが月額 $4,500 を超えている。以下の「悪いコード」を見て問題点を全て洗い出し、dbt Core 1.8+ の Unit Tests と incremental insert_overwrite を活用した形にリファクタリングせよ。

対象モデルは「商品カテゴリ別・日次セッション → 購入転換率（CVR）集計テーブル」。販促チームが Looker Studio でリアルタイムに参照しており、遅延は1時間以内が要件。

制約・前提条件

• BigQuery（asia-northeast1）、dbt Core 1.8+（Unit Tests 使用可）
• raw.sessions は session_date STRING('YYYY-MM-DD') で保存されており変更不可
• stg_orders は 2026-05-20 設計済みで存在すること前提
• 販促チームが参照する mart は直近 7日の CVR サマリー（Looker Studio 直参照）
• 目標: スキャン量 1TB 以下/月、dbt Unit Tests で CVR ロジックの正しさを保証

悪いコード (Before)

📋 コピー-- ❌ materialized='table' で毎回全件フルリフレッシュ
{{ config(materialized='table') }}

SELECT
    -- ❌ CAST(STRING → DATE) で partition pruning が無効化
    CAST(s.session_date AS DATE)    AS order_date,
    s.product_category              AS category,
    COUNT(DISTINCT s.session_id)    AS sessions,
    COUNT(DISTINCT o.order_id)      AS orders,
    -- ❌ sessions=0 の場合に NULL が返る（0.0 にならない）
    COUNT(DISTINCT o.order_id) /
        NULLIF(COUNT(DISTINCT s.session_id), 0) AS cvr
FROM `prj-prod.raw.sessions` s     -- 600GB・pruning 不可
LEFT JOIN `prj-prod.raw.orders` o
    -- ❌ CAST での JOIN → pruning 無効
    ON CAST(s.session_date AS DATE) = CAST(o.order_date AS DATE)
    AND s.product_category = o.product_category
-- ❌ 直近30日を毎回フルスキャン
WHERE CAST(s.session_date AS DATE)
    >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 30 DAY)
GROUP BY 1, 2
-- ❌ Unit Tests 一切なし → CVR バグがサイレントに本番流入

ヒント（段階的開示）

{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
        'field': 'session_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['category']
) }}

SELECT
    session_date,
    category,
    sessions,
    orders,
    COALESCE(SAFE_DIVIDE(orders, sessions), 0.0) AS cvr
FROM {{ ref('int_session_order_joined') }}
{% if is_incremental() %}
WHERE session_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 2 DAY)
{% endif %}

# dbt 1.8+ Unit Test の骨格
unit_tests:
  - name: test_cvr_zero_when_no_sessions
    model: mart_category_cvr_daily
    given:
      - input: ref('int_session_order_joined')
        rows:
          - {session_date: '2026-05-27', category: 'apparel', sessions: 0, orders: 0}
    expect:
      rows:
        - {session_date: '2026-05-27', category: 'apparel', cvr: 0.0}

問題点分析（5点）

アーキテクチャ図 — Before（1段 table）vs After（3層 incremental + Unit Tests）

模範解答 — dbt モデル群 + Unit Tests

📋 コピー{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
        'field': 'session_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['product_category', 'user_id'],
    on_schema_change='append_new_columns'
) }}

-- 目的: raw.sessions（STRING型 session_date）を DATE 型に正規化し
--       partition pruning が効く形に再パーティション化する
SELECT
    PARSE_DATE('%Y-%m-%d', session_date) AS session_date,  -- ✅ STRING → DATE
    session_id,
    user_id,
    product_category,
    page_views,
    duration_seconds,
    created_at
FROM {{ source('raw', 'sessions') }}
{% if is_incremental() %}
-- 遅延セッション対応: 2日分を insert_overwrite で再計算（重複は自動排除）
WHERE PARSE_DATE('%Y-%m-%d', session_date)
    >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 2 DAY)
{% endif %}

📋 コピー{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'session_date', 'data_type': 'date', 'granularity': 'day'},
    cluster_by=['category']
) }}

-- 目的: セッション × 注文を product_category × 日次で事前 JOIN
WITH sessions AS (
    SELECT
        session_date,
        product_category AS category,
        COUNT(DISTINCT session_id) AS sessions
    FROM {{ ref('stg_sessions') }}
    {% if is_incremental() %}
    WHERE session_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 2 DAY)
    {% endif %}
    GROUP BY session_date, product_category
),
orders AS (
    SELECT
        order_date,
        product_category AS category,
        COUNT(DISTINCT order_id) AS orders
    FROM {{ ref('stg_orders') }}  -- 2026-05-20 設計済み
    {% if is_incremental() %}
    WHERE order_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 2 DAY)
    {% endif %}
    GROUP BY order_date, product_category
)
SELECT
    s.session_date,
    s.category,
    s.sessions,
    -- ✅ LEFT JOIN で orders が NULL の場合は 0 に正規化
    COALESCE(o.orders, 0) AS orders
FROM sessions s
LEFT JOIN orders o
    -- ✅ DATE 型同士の比較 → partition pruning 有効
    ON s.session_date = o.order_date
    AND s.category    = o.category

📋 コピー{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
        'field': 'session_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['category']
    -- ✅ ORDER BY なし（ビュー内ソートは再実行されるため削除）
) }}

-- 目的: カテゴリ × 日次の CVR を事前集計して保持する累積テーブル
SELECT
    session_date,
    category,
    sessions,
    orders,
    -- ✅ SAFE_DIVIDE: sessions=0 の場合 NULL ではなく 0.0 を返す
    -- ✅ COALESCE: SAFE_DIVIDE が NULL を返すエッジケース（sessions=NULL）も 0.0 に
    COALESCE(SAFE_DIVIDE(orders, sessions), 0.0) AS cvr
FROM {{ ref('int_session_order_joined') }}
{% if is_incremental() %}
WHERE session_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 2 DAY)
{% endif %}

📋 コピー{{ config(
    materialized='table',  -- ✅ 7日窓の小テーブル（数十MB）。dbt run 毎に再生成
    cluster_by=['category']
) }}

-- 目的: Looker Studio が直接参照する固定窓テーブル。
--       mart_category_cvr_daily から直近7日のみを切り出す。
SELECT
    session_date,
    category,
    sessions,
    orders,
    cvr,
    ROUND(cvr * 100, 2) AS cvr_pct  -- 表示用: パーセント表記（例: 5.00%）
FROM {{ ref('mart_category_cvr_daily') }}
WHERE session_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 7 DAY)
  AND session_date <  CURRENT_DATE('Asia/Tokyo')

📋 コピーversion: 2

models:
  - name: mart_category_cvr_daily
    description: "カテゴリ別日次 CVR 累積テーブル（セッション → 購入転換率）"

unit_tests:
  # ✅ テスト1: sessions=0 のカテゴリは cvr=0.0 であること
  - name: test_cvr_zero_when_no_sessions
    model: mart_category_cvr_daily
    given:
      - input: ref('int_session_order_joined')
        rows:
          - {session_date: '2026-05-27', category: 'new_arrivals', sessions: 0, orders: 0}
    expect:
      rows:
        - {session_date: '2026-05-27', category: 'new_arrivals', cvr: 0.0}

  # ✅ テスト2: 通常ケースの CVR 計算精度（10/200=0.05）
  - name: test_cvr_normal_case
    model: mart_category_cvr_daily
    given:
      - input: ref('int_session_order_joined')
        rows:
          - {session_date: '2026-05-27', category: 'apparel', sessions: 200, orders: 10}
    expect:
      rows:
        - {session_date: '2026-05-27', category: 'apparel', cvr: 0.05}

  # ✅ テスト3: orders=0（LEFT JOIN でマッチなし）の場合 cvr=0.0
  - name: test_cvr_when_no_orders
    model: mart_category_cvr_daily
    given:
      - input: ref('int_session_order_joined')
        rows:
          - {session_date: '2026-05-27', category: 'electronics', sessions: 500, orders: 0}
    expect:
      rows:
        - {session_date: '2026-05-27', category: 'electronics', cvr: 0.0}

月額コスト試算（Before / After）

※ BigQuery オンデマンド料金 $5/TB として試算。リージョン: asia-northeast1。

ポイント解説

実務への応用

• MOps 販促チームの CVR モニタリング: クーポン施策ごとの CVR 変化を追うユースケースはまさにこのパターン。category を campaign_name に置き換えるだけで転用できる。
• Argo Workflows との統合: dbt run --select mart_cvr_last7d+ を Argo の DAG に組み込み、毎時バッチの最後のステップとして実行。前段の incremental stg/int が失敗した場合は Argo のリトライで再計算される。
• DataDog でバッチコストを可視化: BigQuery の INFORMATION_SCHEMA.JOBS_BY_PROJECT から total_bytes_billed を取得し、dbt ジョブ別のスキャン量を DataDog カスタムメトリクスとして送信。突然のコスト増（--full-refresh 誤実行など）をアラートで検知できる。
• dbt --full-refresh は本番禁止: スキーマ変更時のみ手動実行。Argo の dbt run ステップにはフラグを付けず、dbt run --full-refresh は別の手動ワークフローに分離する。

今日のまとめ

自己評価

自分の回答

気づき・メモ