2026-06-10 C: データエンジニアリング | dbt incremental × BigQuery partition/cluster × SAFE_DIVIDE × COALESCE × dbt 1.8 Unit Tests（mart_campaign

概要

🔗

ref() で依存関係グラフに乗せる

直接テーブル名（mops_dataset.stg_orders）を SQL に書くと dbt の DAG に乗らない。{{ ref('stg_orders') }} に変えることで依存順序が自動管理され、dbt run --select mart_campaign_revenue+ で上流モデルが先に実行される。

⚡

incremental + insert_overwrite で BQ コスト削減

materialized='table' は毎回全件 INSERT でクエリコストが線形増加する。incremental_strategy='insert_overwrite' + 7日ルックバックで差分パーティションのみ更新し、遅延データも確実に取り込める。

🔪

SAFE_DIVIDE × COALESCE で NULL/ゼロ除算を防ぐ

A / B は B=0 で BigQuery エラー。SAFE_DIVIDE(A, B) はゼロを NULL として扱い安全。COALESCE(category, '未分類') で NULL グループを GROUP BY から消さない。

🧪

dbt 1.8 Unit Tests でロジックを自動検証

モックデータで SQL ロジックを単体テスト。BigQuery への実クエリ不要で CI が高速になる。正常系・ゼロ除算・NULL カテゴリを網羅したテストケースで回帰バグを検出する。

問題

ECサイトの MOps チームでは、dbt を使って BigQuery 上でキャンペーン別・商品カテゴリ別の日次売上指標を算出している。現在の dbt モデル mart_campaign_revenue.sql には 6つの設計上の問題 が潜んでいる。問題点を全て洗い出し、dbt Core 1.8+ × BigQuery ベストプラクティス に従って修正せよ。

制約・前提条件

dbt Core 1.8+（Unit Tests 機能・dbt Mesh 対応）
BigQuery（パーティション: order_date、クラスタリング: campaign_id, category）
incremental モデルで insert_overwrite 戦略を使い、遅延データ（最大7日）に対応すること
SAFE_DIVIDE でゼロ除算を防ぐこと
dbt 1.8 Unit Tests で3件以上のテストケースを実装すること
COALESCE / NULLIF で NULL を適切にハンドリングすること

期待する回答形式: 問題点の列挙（番号付き）+ 改善後 dbt モデル（SQL + schema.yml）+ Unit Tests（YAML）+ 設計意図の説明

悪い dbt モデル (Before)

この dbt モデルには 6つの設計上の問題 が隠れています。

bad_mart_campaign_revenue.sql — 問題だらけの mart モデル

-- 問題①: config がなく materialized のデフォルト(view)または別ファイルで table 指定
-- 問題③: partition_by / cluster_by がない（フルスキャン確定）
{{
  config(
    materialized='table'
  )
}}

SELECT
  order_date,
  campaign_id,
  -- 問題⑤: COALESCE なし（NULL カテゴリが GROUP BY から消える）
  category,
  COUNT(DISTINCT order_id)           AS order_count,
  SUM(subtotal)                      AS gross_revenue,
  SUM(discount_amount)               AS total_discount,
  SUM(subtotal - discount_amount)    AS net_revenue,
  -- 問題④: SAFE_DIVIDE なし（注文ゼロ日に DIVISION_BY_ZERO エラー）
  SUM(subtotal - discount_amount)
    / COUNT(DISTINCT order_id)       AS avg_order_value,
  SUM(discount_amount)
    / SUM(subtotal)                  AS discount_rate

-- 問題①: ref() なし（直接テーブル参照 → dbt DAG に乗らない）
FROM mops_dataset.stg_orders

-- 問題②: incremental 未使用（毎回フルスキャン・全件 INSERT）
-- WHERE 句なし: is_incremental() マクロ未使用

GROUP BY
  order_date,
  campaign_id,
  category

問題点サマリー（6点）

1直接テーブル参照（FROM mops_dataset.stg_orders）— dbt の DAG に乗らず、CI での依存順序制御が不可。{{ ref('stg_orders') }} に変更する

2materialized='table' でフルリビルド — 毎回全件 INSERT でコスト線形増加。materialized='incremental' + insert_overwrite + 7日ルックバックに変更

3partition_by / cluster_by 未設定 — フルテーブルスキャン確定。order_date パーティション + campaign_id, category クラスタリングを設定

4ゼロ除算リスク（/ COUNT(DISTINCT order_id)）— 注文ゼロ日に DIVISION_BY_ZERO エラー。SAFE_DIVIDE() に変更

5NULL カテゴリの消失 — COALESCE(category, '未分類') がないと NULL カテゴリの注文が集計から漏れる

6dbt Unit Tests 未実装 — dbt 1.8+ の unit_tests ブロックがなく、SQL ロジックの自動検証ができない

ヒント（段階的開示）

ヒント1 — 方向性

直接テーブル名を SQL 内に書くと dbt の依存関係グラフに乗らず、CI で dbt run --select の順序制御ができない。incremental モデルは is_incremental() マクロと this を使ってパーティション絞り込みを行うことでフルスキャンを防ぐ。SAFE_DIVIDE(A, B) は B=0 のとき NULL を返し、ゼロ除算例外を発生させない。

ヒント2 — アプローチ

問題①: FROM mops_dataset.stg_orders → FROM {{ ref('stg_orders') }}
問題②: materialized='incremental' + incremental_strategy='insert_overwrite' + is_incremental() マクロで7日ルックバック
問題③: config ブロックに partition_by（order_date, granularity='day'）と cluster_by=['campaign_id', 'category'] を追加
問題④: SUM(net) / COUNT(order_id) → SAFE_DIVIDE(SUM(net), COUNT(order_id))
問題⑤: category → COALESCE(category, '未分類') AS category
問題⑥: schema.yml に unit_tests: ブロックを追加。正常系・ゼロ除算・NULL カテゴリの3ケース以上を定義

ヒント3 — コードの骨格

-- 改善後の config ブロック
{{
  config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
      'field': 'order_date',
      'data_type': 'date',
      'granularity': 'day'
    },
    cluster_by=['campaign_id', 'category'],
    on_schema_change='fail'
  )
}}

WITH orders AS (
  SELECT
    order_date,
    campaign_id,
    COALESCE(category, '未分類') AS category,  -- NULL → '未分類'
    subtotal,
    discount_amount,
    order_id
  FROM {{ ref('stg_orders') }}   -- ref() に変更

  {% if is_incremental() %}
  WHERE order_date >= DATE_SUB(
    (SELECT MAX(order_date) FROM {{ this }}),
    INTERVAL 7 DAY
  )
  {% endif %}
)

SELECT
  ...,
  SAFE_DIVIDE(
    SUM(subtotal - discount_amount),
    COUNT(DISTINCT order_id)
  ) AS avg_order_value   -- ゼロ除算安全
FROM orders
GROUP BY order_date, campaign_id, category

問題点分析（6点）

#	問題点	分類	改善方法
1	直接テーブル参照（`FROM mops_dataset.stg_orders`）	dbt DAG	`{{ ref('stg_orders') }}` に変更
2	`materialized='table'` でフルリビルド（毎回全件 INSERT）	コスト	`incremental + insert_overwrite + 7日ルックバック`
3	`partition_by / cluster_by` 未設定（フルスキャン）	コスト/性能	`order_date` パーティション + `campaign_id, category` クラスタ
4	ゼロ除算リスク（`/ COUNT(order_id)`）	信頼性	`SAFE_DIVIDE()` で NULL 返却に変更
5	NULL カテゴリの消失（GROUP BY から除外）	データ品質	`COALESCE(category, '未分類')`
6	dbt Unit Tests 未実装（ロジック自動検証なし）	テスト品質	`unit_tests:` ブロックで正常系・異常系を定義

アーキテクチャ図 — Bad vs Good の変換フロー

模範解答

-- models/mart/mart_campaign_revenue.sql
-- キャンペーン別・カテゴリ別 日次売上指標（incremental + BQパーティション最適化）
--
-- 設計意図:
--   - incremental + insert_overwrite: フルスキャンを避けて BQ コスト削減
--   - 7日ルックバック: 遅延データ（最大T+7着荷）を確実に取り込む
--   - SAFE_DIVIDE: ゼロ除算を NULL として扱い、ダウンストリームで COALESCE
--   - COALESCE(category, '未分類'): NULL グループを捨てない
--   - cluster_by: campaign_id + category フィルタクエリを高速化

{{
  config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
      'field': 'order_date',
      'data_type': 'date',
      'granularity': 'day'
    },
    cluster_by=['campaign_id', 'category'],
    on_schema_change='fail'       -- スキーマ変更を CI で即座に検出
  )
}}

WITH
-- 修正①: ref() で依存関係グラフに乗せる
orders AS (
  SELECT
    order_date,
    campaign_id,
    -- 修正⑤: NULL カテゴリを '未分類' に統一（GROUP BY で行が消えるのを防ぐ）
    COALESCE(category, '未分類')     AS category,
    subtotal,
    discount_amount,
    COALESCE(shipping_fee, 0)        AS shipping_fee,   -- NULL → 0 円
    order_id
  FROM {{ ref('stg_orders') }}   -- 直接テーブル名 → ref()

  -- 修正②: incremental 時は7日ルックバックで遅延データに対応
  -- is_incremental() = false（--full-refresh / 初回）の場合は WHERE 句が除去される
  {% if is_incremental() %}
  WHERE order_date >= DATE_SUB(
    (SELECT MAX(order_date) FROM {{ this }}),
    INTERVAL 7 DAY
  )
  {% endif %}
),

aggregated AS (
  SELECT
    order_date,
    campaign_id,
    category,

    -- 売上指標
    COUNT(DISTINCT order_id)                                      AS order_count,
    SUM(subtotal)                                                 AS gross_revenue,
    SUM(discount_amount)                                          AS total_discount,
    SUM(subtotal - discount_amount)                               AS net_revenue,
    SUM(shipping_fee)                                             AS total_shipping,

    -- 修正④: SAFE_DIVIDE でゼロ除算を防ぐ（注文ゼロ日は NULL）
    SAFE_DIVIDE(
      SUM(subtotal - discount_amount),
      COUNT(DISTINCT order_id)
    )                                                             AS avg_order_value,

    -- 割引率（クーポン効果測定用）
    SAFE_DIVIDE(
      SUM(discount_amount),
      NULLIF(SUM(subtotal), 0)   -- NULLIF: subtotal 合計がゼロの場合も NULL 扱い
    )                                                             AS discount_rate,

    CURRENT_TIMESTAMP()                                           AS _loaded_at

  FROM orders
  GROUP BY
    order_date,
    campaign_id,
    category
)

SELECT * FROM aggregated

# models/mart/schema.yml
version: 2

models:
  - name: mart_campaign_revenue
    description: |
      キャンペーン別・カテゴリ別の日次売上指標（mart 層）。
      incremental + insert_overwrite で BQ コスト削減。
      遅延データ最大T+7に対応。
    config:
      contract:
        enforced: true   # スキーマ変更を CI で即座に検出
    columns:
      - name: order_date
        data_type: date
        description: 注文日（パーティションキー）
        constraints:
          - type: not_null
      - name: campaign_id
        data_type: string
        description: キャンペーンID（クラスタキー①）
        constraints:
          - type: not_null
      - name: category
        data_type: string
        description: 商品カテゴリ（COALESCE済み、NULLなし）
        constraints:
          - type: not_null
      - name: order_count
        data_type: int64
        description: 注文件数
      - name: gross_revenue
        data_type: numeric
        description: クーポン適用前総売上
      - name: total_discount
        data_type: numeric
        description: クーポン割引総額
      - name: net_revenue
        data_type: numeric
        description: クーポン適用後純売上
      - name: avg_order_value
        data_type: numeric
        description: 注文単価（注文ゼロ日は NULL）
      - name: discount_rate
        data_type: numeric
        description: 割引率 0.0〜1.0（ゼロ除算時 NULL）

    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [order_date, campaign_id, category]
      - not_null:
          column_name: net_revenue

# ── dbt 1.8+ Unit Tests ─────────────────────────────────────────────────────
unit_tests:
  - name: test_mart_campaign_revenue_basic
    description: 正常系: 2件の注文を正しく集計できるか検証
    model: mart_campaign_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-10", campaign_id: "CMP-001", category: "apparel",
             subtotal: 5000, discount_amount: 500, shipping_fee: 300, order_id: "O-001"}
          - {order_date: "2026-06-10", campaign_id: "CMP-001", category: "apparel",
             subtotal: 3000, discount_amount: 0,   shipping_fee: 0,   order_id: "O-002"}
    expect:
      rows:
        - {order_date: "2026-06-10", campaign_id: "CMP-001", category: "apparel",
           order_count: 2, gross_revenue: 8000, total_discount: 500,
           net_revenue: 7500, total_shipping: 300}

  - name: test_mart_campaign_revenue_zero_subtotal
    description: ゼロ売上注文: gross_revenue=0 のとき discount_rate が NULL になるか検証（NULLIF）
    model: mart_campaign_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-09", campaign_id: "CMP-002", category: "electronics",
             subtotal: 0, discount_amount: 0, shipping_fee: 0, order_id: "O-003"}
    expect:
      rows:
        - {order_date: "2026-06-09", campaign_id: "CMP-002", category: "electronics",
           order_count: 1, gross_revenue: 0, total_discount: 0,
           net_revenue: 0, avg_order_value: 0, discount_rate: null}

  - name: test_mart_campaign_revenue_null_category
    description: NULL カテゴリ: COALESCE で '未分類' に変換されるか検証（行が消えないことの確認）
    model: mart_campaign_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-10", campaign_id: "CMP-003", category: null,
             subtotal: 10000, discount_amount: 1000, shipping_fee: 500, order_id: "O-004"}
    expect:
      rows:
        - {order_date: "2026-06-10", campaign_id: "CMP-003", category: "未分類",
           order_count: 1, gross_revenue: 10000, total_discount: 1000, net_revenue: 9000}

  - name: test_mart_campaign_revenue_discount_rate
    description: 割引率: discount_rate = discount / gross_revenue の計算を検証
    model: mart_campaign_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-10", campaign_id: "CMP-004", category: "apparel",
             subtotal: 10000, discount_amount: 2000, shipping_fee: 0, order_id: "O-005"}
    expect:
      rows:
        - {order_date: "2026-06-10", campaign_id: "CMP-004", category: "apparel",
           gross_revenue: 10000, total_discount: 2000, net_revenue: 8000,
           discount_rate: 0.2}

incremental_strategy 比較（BigQuery）

戦略	動作	推奨シーン
`merge`	primary key で UPDATE/INSERT。行レベル更新	注文ステータス更新などが頻繁な場合
`insert_overwrite`	パーティション単位で丸ごと置き換え	日次集計（今日の全件を再計算）
`append`	新規行のみ追加（重複チェックなし）	イベントログなど追記のみのデータ

7日ルックバックの理由

-- なぜ7日ルックバックか？
-- ECサイトの注文データは注文時点ではなく「配送完了」「返品処理」タイミングで
-- stg_orders に同期される場合がある（最大T+7日遅延）。
-- 7日ルックバックで遅延データを確実に取り込む。

{% if is_incremental() %}
WHERE order_date >= DATE_SUB(
  -- {{ this }}: 現在の mart テーブル自身を参照
  -- MAX(order_date): 最後に更新されたパーティションの日付
  (SELECT MAX(order_date) FROM {{ this }}),
  INTERVAL 7 DAY   -- 7日前から再計算（遅延データ対応）
)
{% endif %}

-- is_incremental() = false の場合（初回実行・--full-refresh）:
-- WHERE 句が除去され、stg_orders の全データをフルスキャン
-- その後は incremental 実行でコストを削減

on_schema_change='fail' の重要性

-- on_schema_change オプション比較
-- 'fail'        : スキーマ変更を検出したら dbt run を失敗させる（推奨）
--                 意図しないカラム追加・型変更を CI で即座に検出できる
-- 'ignore'      : スキーマ変更を無視（デフォルト）
--                 新カラムが追加されても既存データに適用されない
-- 'append_new_columns' : 新カラムを追加（既存データの新カラムは NULL）
-- 'sync_all_columns'   : 完全同期（危険: カラム削除で既存データ欠損）

{{
  config(
    on_schema_change='fail'   -- mart 層では 'fail' を推奨
  )
}}

BQ コスト試算（Bad vs Good）

条件	Bad（table）	Good（incremental）	削減率
データ量	365日分 × 1GB/日 = 365GB	7日分 × 1GB/日 = 7GB	—
dbt run のスキャン量	365 GB	7 GB	98%削減
dbt run コスト/回	365 × $5/TB = $1.83	7 × $5/TB = $0.04	$1.79削減
月間 dbt run（30回）	$54.8/月	$1.1/月	$53.7/月削減
Looker Studio クエリ（1,000回/月）	365GB × 1,000 = 365TB	30日分（30GB）× 1,000 = 30TB	92%削減

注: クラスタリング（cluster_by）の効果はワークロードに依存。特定のキャンペーン ID + カテゴリへの絞り込みが多い場合、さらに70〜90%のスキャン削減が期待できる。

INFORMATION_SCHEMA でパーティション効果を確認

-- BigQuery でパーティション剪定の効果を確認
SELECT
  table_name,
  partition_id,
  total_rows,
  total_logical_bytes / POW(1024, 3) AS size_gb,
  last_modified_time
FROM `mops_dataset.INFORMATION_SCHEMA.PARTITIONS`
WHERE table_name = 'mart_campaign_revenue'
ORDER BY partition_id DESC
LIMIT 14;  -- 直近14日分のパーティション状況を確認

問題	修正内容	効果
① 直接テーブル参照	`{{ ref('stg_orders') }}`	dbt DAG で依存順序が保証・CI で制御可能
② table フルリビルド	incremental + insert_overwrite + 7日ルックバック	BQ コスト98%削減・遅延データ対応
③ partition/cluster 未設定	order_date パーティション + 2カラムクラスタ	BI クエリコスト90%以上削減・高速化
④ ゼロ除算リスク	`SAFE_DIVIDE()`	ゼロ件日の dbt 実行エラーを排除
⑤ NULL カテゴリ消失	`COALESCE(category, '未分類')`	全注文を漏れなく集計・BI 表示が正確に
⑥ Unit Tests 未実装	dbt 1.8 unit_tests 4ケース	回帰バグを CI で自動検出・BQ 接続不要

ポイント解説

1 ref() による依存関係管理（問題①）
dbt は {{ ref('stg_orders') }} を解析して DAG（有向非巡回グラフ）を構築する。dbt run --select mart_campaign_revenue+ で上流の staging モデルも自動実行でき、CI での依存順序が保証される。直接テーブル名を書くと DAG に乗らず、stg が未更新の状態で mart が先に実行されるリスクがある。また dbt test での lineage 追跡や dbt docs generate でのドキュメント生成にも影響する。

2 incremental_strategy='insert_overwrite' + 7日ルックバック（問題②）
BigQuery の insert_overwrite はパーティション単位で丸ごと上書きする。7日ルックバックを設定することで、T+1〜T+7 に着荷する遅延注文データ（配送完了後に同期されるケース）を確実に取り込める。is_incremental() マクロが false（初回実行・--full-refresh 時）は WHERE 句が除去されて全件処理になる。merge 戦略は行レベルの MERGE 文を発行するため、日次集計のような「パーティション丸ごと再計算」には insert_overwrite が適している。

3 BigQuery partition_by + cluster_by（問題③）
order_date パーティションにより、日付範囲フィルタクエリのスキャン量を削減（例: 過去30日→1/12 のデータのみスキャン）。cluster_by=['campaign_id', 'category'] でさらにブロックプルーニングが効き、特定キャンペーン・カテゴリへの絞り込みクエリが高速化する。BigQuery はクラスタリングカラムの順序が重要で、最初のカラムへのフィルタが最も効果が高い。コスト削減効果は数十〜数百倍になることがある。INFORMATION_SCHEMA.PARTITIONS でパーティション剪定の実効果を確認できる。

4 SAFE_DIVIDE vs 通常の除算（問題④）
A / B は B=0 のとき BigQuery で DIVISION_BY_ZERO エラーになり、dbt ジョブ全体が失敗する。SAFE_DIVIDE(A, B) は B=0 または B=NULL のとき NULL を返す。ダウンストリームのダッシュボードでは COALESCE(avg_order_value, 0) で NULL を 0 に変換して表示する。NULLIF(SUM(subtotal), 0) は「合計がゼロのとき NULL を返す」で、SAFE_DIVIDE と組み合わせて二重の安全網を張ることができる。

5 COALESCE(category, '未分類') の重要性（問題⑤）
BigQuery の GROUP BY では NULL 同士はグルーピングされる（NULL グループが1行できる）が、Looker Studio などの BIツールでは NULL のグループが (blank) として表示され見づらい。また、アプリケーション側で WHERE category IS NOT NULL を入れているケースでは NULL グループが完全に除外され、売上が過少集計に見える問題が起きる。COALESCE で事前に代替値を設定しておくことで、下流での処理が統一される。

6 dbt 1.8 Unit Tests（問題⑥）
unit_tests ブロックは SQL ロジックをモックデータで単体テストする機能（dbt 1.8 以降）。BigQuery への実際のクエリ実行なしに SQL の集計ロジックを検証できるため、CI の実行時間を大幅に短縮できる。dbt test --select mart_campaign_revenue で既存の schema テスト（not_null、unique）と Unit Tests が同時実行される。ポイントは「正常系」「境界値（ゼロ）」「異常値（NULL）」の3軸をカバーすること。

実務への応用

Argo Workflows との連携: dbt run --select mart_campaign_revenue を Argo の dag ステップとして組み込み、stg_orders の更新後に mart_campaign_revenue が自動実行されるパイプラインを構築する。--target prod で本番 BQ プロジェクトに対して実行し、--vars '{"run_date": "{{ execution_date }}"}' で実行日を渡せる。
dbt Mesh（cross-project ref）: stg_orders が別の dbt プロジェクト（例: orders-platform）に移管された場合、{{ ref('orders-platform', 'stg_orders') }} の cross-project ref を使い、プロジェクト間の依存を明示的に管理できる（dbt 1.6+ / dbt Cloud 限定）。
Looker Studio × BQコスト管理: partition_by=order_date を設定しておくことで、Looker Studio のデータソースが自動的に日付範囲フィルタをパーティション剪定として解釈し、BI クエリコストが激減する。cluster_by の恩恵は複数フィルタ組み合わせ時に顕著（キャンペーン + カテゴリの複合フィルタなど）。
DataDog モニタリング: _loaded_at カラムを使い「最終更新から2時間以上経過したら Alert」の DataDog Monitor を設定することで、dbt パイプラインの遅延を検知できる。dbt source freshness コマンドを Argo のステップに組み込んで source データの鮮度も監視する。

今日のまとめ

dbt の mart モデルは「ref() による DAG 管理 + incremental+insert_overwrite + partition_by/cluster_by + SAFE_DIVIDE/COALESCE + dbt 1.8 Unit Tests」の5点セットが本番品質の最低ラインであり、直接テーブル参照・フルリビルド・ゼロ除算放置・NULL 放置の4つが最もよく見られる設計負債パターン。

insert_overwrite + 7日ルックバックは「遅延データ対応 × コスト削減 × シンプルな実装」のバランスが最も良い。merge は行レベルの更新コストが高く、日次集計には不向き。

自己評価

完全理解おおむね理解要復習

C データエンジニアリング — dbt incremental × BQ partition/cluster × SAFE_DIVIDE × COALESCE × dbt 1.8 Unit Tests（mart_campaign_revenue Bad→Good）

概要

ref() で依存関係グラフに乗せる

incremental + insert_overwrite で BQ コスト削減

SAFE_DIVIDE × COALESCE で NULL/ゼロ除算を防ぐ

dbt 1.8 Unit Tests でロジックを自動検証

問題

制約・前提条件

悪い dbt モデル (Before)

ヒント（段階的開示）

問題点分析（6点）

アーキテクチャ図 — Bad vs Good の変換フロー

模範解答

incremental_strategy 比較（BigQuery）

7日ルックバックの理由

on_schema_change='fail' の重要性

BQ コスト試算（Bad vs Good）

INFORMATION_SCHEMA でパーティション効果を確認

ポイント解説

実務への応用

今日のまとめ

自己評価

自分の回答

気づき・メモ