データエンジニアリング — dbt incremental × COUNTIF 最適化

概要

問題

以下のBigQueryテーブルがあり、日次でキャンペーン効果分析レポートをdbtモデルとして作成しています。

テーブル仕様: raw.campaign_events

制約

• テーブルサイズ: 90日で約 3TB（1日 33GB）
• dbt Core 1.8+ を使用
• 本番環境では毎日 06:00 JST に実行される
• base CTEは全ての event_type を含む（現在は3回テーブルスキャンが発生）

現在の dbt モデル (Bad)

📋 コピー-- mart/campaign_daily_metrics.sql（問題あり）
{{ config(
    materialized='table'
) }}

WITH base AS (
    SELECT *
    FROM {{ source('raw', 'campaign_events') }}
    WHERE event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
),

impressions AS (
    SELECT
        event_date,
        campaign_id,
        COUNT(*) AS impression_count
    FROM base
    WHERE event_type = 'impression'
    GROUP BY 1, 2
),

clicks AS (
    SELECT
        event_date,
        campaign_id,
        COUNT(*) AS click_count
    FROM base
    WHERE event_type = 'click'
    GROUP BY 1, 2
),

conversions AS (
    SELECT
        event_date,
        campaign_id,
        COUNT(*) AS conversion_count,
        SUM(revenue) AS total_revenue
    FROM base
    WHERE event_type = 'conversion'
    GROUP BY 1, 2
)

SELECT
    i.event_date,
    i.campaign_id,
    i.impression_count,
    c.click_count,
    cv.conversion_count,
    cv.total_revenue,
    SAFE_DIVIDE(c.click_count, i.impression_count) AS ctr,
    SAFE_DIVIDE(cv.conversion_count, c.click_count) AS cvr
FROM impressions i
LEFT JOIN clicks c USING (event_date, campaign_id)
LEFT JOIN conversions cv USING (event_date, campaign_id)

ヒント（段階的開示）

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

{% if is_incremental() %}
  AND event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)
{% else %}
  AND event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
{% endif %}

-- 1スキャンで3種類の集計
SELECT
    event_date,
    campaign_id,
    COUNTIF(event_type = 'impression')   AS impression_count,
    COUNTIF(event_type = 'click')        AS click_count,
    ...

問題点分析

コスト削減の可視化

模範解答

📋 コピー{{ config(
    materialized='incremental',
    partition_by={
        "field": "event_date",
        "data_type": "date",
        "granularity": "day"
    },
    cluster_by=["campaign_id"],
    incremental_strategy='insert_overwrite',
    on_schema_change='sync_all_columns'
) }}

/*
  改善点:
  1. [BQコスト] COUNTIF による1スキャン集計（3スキャン → 1スキャン）
  2. [dbt]      incremental + insert_overwrite で日次増分処理（90日フルリビルド廃止）
  3. [品質]     COALESCE で click/conversion が 0 の場合の NULL を 0 に明示
*/

SELECT
    event_date,
    campaign_id,

    -- 1回のスキャンで3種類のイベントを集計
    COUNTIF(event_type = 'impression')                                 AS impression_count,
    COUNTIF(event_type = 'click')                                      AS click_count,
    COUNTIF(event_type = 'conversion')                                 AS conversion_count,
    SUM(IF(event_type = 'conversion', COALESCE(revenue, 0.0), 0.0))   AS total_revenue,

    -- 分母ゼロは NULL（SAFE_DIVIDE）で返す — ダッシュボード側でゼロ除算を避ける
    SAFE_DIVIDE(
        COUNTIF(event_type = 'click'),
        COUNTIF(event_type = 'impression')
    ) AS ctr,

    SAFE_DIVIDE(
        COUNTIF(event_type = 'conversion'),
        COUNTIF(event_type = 'click')
    ) AS cvr

FROM {{ source('raw', 'campaign_events') }}

WHERE 1=1

{% if is_incremental() %}
  -- 増分実行時: 直近3日間を上書き（遅延データ・修正データを考慮してバッファ3日）
  AND event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)
{% else %}
  -- 初回フルビルド時: 90日分をロード
  AND event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
{% endif %}

GROUP BY
    event_date,
    campaign_id

ポイント解説

コスト削減の実務判断フロー

実務への応用

📋 コピー# Argo Workflows からの dbt 実行コマンド例
dbt run --select mart.campaign_daily_metrics --target prod

# フルリフレッシュが必要な場合（スキーマ変更時）
dbt run --select mart.campaign_daily_metrics --full-refresh --target prod

• MOps の日次キャンペーンレポート: 同様のパターン（複数CTE参照）が発生しやすい。COUNTIF最適化はすぐに適用できる。
• スキーマ変更時: on_schema_change='sync_all_columns' を設定することで、 新カラム追加時も自動的にテーブルスキーマを同期できる。
• コスト削減をPMに伝える: 年間$19,800の節約は具体的な数値でPMへの説明材料になる。

次のステップ

• 参考: dbt Incremental Models ドキュメント
• 参考: BigQuery COUNTIF vs COUNT(IF(...)) のパフォーマンス比較
• 参考: BigQuery パーティション設計ガイドライン

今日のまとめ

自己評価

自分の回答

気づき・メモ