C: データエンジニアリング — dbt Incremental モデル × BigQuery コスト最適化

2026-04-15 (Day 3) C: データエンジニアリング ★★★☆☆ dbt Core 1.8+ / BigQuery 受注データ集計パイプライン

概要

⚡

Incrementalモデル

毎回フルスキャンではなく差分データのみを処理。コストを最大96%削減できる実務必須パターン。

🔢

NULL安全な集計

COALESCE・SAFE_DIVIDE・NULLIFを使い、NULLが混入しても正確な集計値を返す。

🧪

schema.yml テスト

unique・not_nullテストで下流モデルへのバグ伝播を防ぐ品質ゲート。

🔗

source()マクロ

テーブル参照を{{ source('raw', 'orders') }}で管理し、dbtがリネージを自動追跡。

問題

BigQuery + dbt を使った受注データの集計パイプライン。現在の dbt モデルに 3つの問題点 があります。

下流モデルの依存関係

モデル用途
models/marts/orders_daily.sql日次集計（今日修正対象）
models/marts/orders_weekly.sql週次集計（orders_dailyを参照）
models/reports/kpi_dashboard.sqlKPIダッシュボード（orders_dailyを参照）

制約・前提条件

  • dbt Core 1.8+ を使用
  • BigQuery がデータウェアハウス
  • raw.orders は日次バッチで更新（1日約100万行追加）
  • amount は NULL になりうる
  • 同じ日のデータを再集計（再実行）することがある
期待する回答形式: dbt モデルのコード（SQL + YAML設定）+ 問題点の説明

悪いコード (Before)

このコードには 3つの問題 があります。NULL処理・コスト・品質保証の観点から探してください。 
-- models/marts/orders_daily.sql
-- 現状のコード（Bad）

SELECT
    DATE(created_at) AS order_date,
    COUNT(*) AS order_count,
    SUM(amount) AS revenue,
    SUM(amount) / COUNT(*) AS avg_order_value
FROM `project.raw.orders`
WHERE status != 'cancelled'
GROUP BY 1
ORDER BY 1

ヒント（段階的開示）

ヒント1 — 方向性
「NULL の扱い」「増分更新（Incremental）」「ドキュメント・テスト」の3つの観点から問題点を探す。 現状のコードがBigQueryの料金コストや再実行時の整合性にどう影響するかを考えよう。
ヒント2 — アプローチ
  • SUM(amount) に NULL が含まれると... SUMはNULL無視するがCOUNT(*)はNULL行もカウントするため平均が不正確になる
  • raw.orders が毎日100万行追加されるのに、毎回全件スキャンしたら... コストと時間の無駄
  • dbt でモデルの品質を保証するには schema.yml の tests: を使う
ヒント3 — Incrementalモデルの骨格 
{{
  config(
    materialized='incremental',
    unique_key='order_date',
    partition_by={
      "field": "order_date",
      "data_type": "date"
    }
  )
}}

SELECT ...
{% if is_incremental() %}
WHERE DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)
{% endif %}

COALESCE(SUM(amount), 0) で NULL を 0 に変換、SAFE_DIVIDE() でゼロ除算を防ぐことができます。

問題点分析

#問題点分類影響改善方法
1 SUM(amount) / COUNT(*) でNULLを考慮していない NULL非考慮 COUNT(*)はNULL行も含むため平均単価が不正確になる SAFE_DIVIDE(COALESCE(SUM(amount),0), NULLIF(COUNT(amount),0))
2 デフォルトmaterialized='table'でフルスキャン コスト増大 100万行/日 × 30日 = 3000万行を毎回スキャン incrementalモデル + ルックバック3日で差分のみ処理
3 schema.ymlがなくテストとドキュメントが欠如 品質保証なし order_dateのユニーク性等の保証がなく下流モデルが壊れたデータを使うリスク unique・not_nullテストをschema.ymlに定義

dbt リネージ構造図

データの依存関係と今回修正するモデルの位置づけ。

raw.orders 1日100万行追加 amount: NULL あり orders_daily materialized: incremental unique_key: order_date ← 今日修正するモデル orders_weekly 週次集計 kpi_dashboard KPIダッシュボード source() マクロで参照 ref() ref() コスト比較（BigQuery） Before フルスキャン: 約$15/月 After Incremental(3日): 約$0.45/月（96%削減） dbt build --select orders_daily+ 下流モデルまで一括再実行

raw.orders から orders_daily（修正対象）を経て、orders_weekly と kpi_dashboard が参照する。Incrementalモデルにすることで毎日のスキャン量を96%削減できる。

模範解答

-- models/marts/orders_daily.sql
-- 改善後コード（Good）

{{
  config(
    materialized='incremental',
    unique_key='order_date',
    partition_by={
      "field": "order_date",
      "data_type": "date",
      "granularity": "day"
    },
    cluster_by=["order_date"],
    on_schema_change='fail'
  )
}}

/*
  日次受注集計モデル。
  Incremental + Partition により、差分のみ処理してコストを最小化する。
  再実行時は直近3日分を再計算して、遅延データを取り込む。
*/

-- 定数定義（マジックナンバーを避ける）
{% set LOOKBACK_DAYS = 3 %}

SELECT
    DATE(created_at)            AS order_date,
    COUNT(*)                    AS order_count,
    -- NULL を 0 として扱い、SUM の精度を保証
    COALESCE(SUM(amount), 0)    AS revenue,
    -- 平均単価: amount が NULL の行は分母から除外
    SAFE_DIVIDE(
        COALESCE(SUM(amount), 0),
        NULLIF(COUNT(amount), 0)
    )                           AS avg_order_value
FROM {{ source('raw', 'orders') }}
WHERE
    status != 'cancelled'
    {% if is_incremental() %}
    -- 増分実行時: 直近N日のみ処理（遅延到着データを考慮して余裕を持つ）
    AND DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL {{ LOOKBACK_DAYS }} DAY)
    {% endif %}
GROUP BY
    order_date
ORDER BY
    order_date
# models/marts/schema.yml
version: 2

models:
  - name: orders_daily
    description: >
      日次受注集計。キャンセル以外のすべての受注を対象に、
      日単位で件数・売上・平均単価を集計する。
      Incremental モデルのため直近3日分を差分更新する。

    config:
      tags: ["mart", "orders", "daily"]

    columns:
      - name: order_date
        description: "受注日（JST）"
        data_tests:
          - unique
          - not_null

      - name: order_count
        description: "受注件数（キャンセル除く）"
        data_tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

      - name: revenue
        description: "売上合計（円）。amount が NULL の場合は 0 として扱う"
        data_tests:
          - not_null

      - name: avg_order_value
        description: "平均注文単価（円）。注文件数が0の日は NULL"

ポイント解説

1 SAFE_DIVIDE vs 除算演算子
BigQueryでは a / b は b=0 のとき DIVISION_BY_ZERO エラー。SAFE_DIVIDE(a, b) は 0 除算時に NULL を返す。実務では SAFE_DIVIDE を使う方が安全。
2 Incrementalモデルのルックバック期間
本番では「バッチ遅延」や「更新データの遅着」が頻繁に起きる。INTERVAL 1 DAY だと当日分しか処理されず、昨日の更新データを取りこぼす。3日程度のルックバックが一般的。
3 unique_key と Partition の組み合わせ
unique_key='order_date' により、同じ order_date のレコードは MERGE か DELETE+INSERT で更新される。Partitionと組み合わせることで、スキャン対象パーティションを絞り込んでコストを削減できる。
4 source() マクロ
raw.orders を直接書くのではなく {{ source('raw', 'orders') }} と書くことで、dbt がリネージを追跡でき、dbt docs generate で自動的に上流・下流の依存関係グラフが生成される。

実務への応用

ECサイトの販促システム（MOps）文脈での活用
  • dbt build --select orders_daily+ で下流モデルまで一括再実行できる
  • dbt test --select orders_daily で order_date のユニーク性チェックを自動化できる
  • BigQuery コスト観点: フルスキャン（約$15/月）→ Incremental（約$0.45/月）= 96%削減

次のステップ

発展問題: orders_daily に対して「前日比成長率（LAG ウィンドウ関数）」と「7日移動平均」を追加するには？
  • 参考: dbt Core 1.8 ドキュメント「Incremental models」
  • 参考: BigQuery「Partitioned tables best practices」

今日のまとめ

dbt の Incremental モデル + BigQuery Partition は「コスト削減 × データ品質保証」を両立する実務必須パターン。NULL 処理（SAFE_DIVIDE / COALESCE）と schema.yml のテスト定義は下流モデルへのバグ伝播を防ぐ防壁として機能する。

自己評価

自分の回答

気づき・メモ