dbtモデルを増分化してコスト削減しよう

新田（エンジニア）

2025.12.19

IT技術

BigQuery dbt

はじめに

dbtはデータパイプラインを構築するためのとても便利なツールです。
単にSQLを実行してくれるだけでなく、関連する機能がたくさんあります。今回はincrementalモデルとして実行してコストを削減する方法をまとめました。
DWHはBigQueryを利用しています。

実装方法

早速実装方法です。
以下のようなSQLモデルがあったとしたら、

1{{
2    config(
3        materialized="table",
4        alias="daily_sales",
5        partition_by={
6            "field": "date",
7            "data_type": "date",
8            "granularity": "day",
9        }
10    )
11}}
12with
13
14    cleansed_orders as (
15        select *
16        from {{ ref("int__cleansed_orders") }}
17    )
18select
19    date(order_time_jst) as date,
20    sum(sales_jpy) as sales,
21    count(distinct user_id) as payment_uu,
22    safe_divide(sum(sales_jpy), count(distinct user_id)) as arppu
23from cleansed_orders
24group by 1

以下のように書き換えます。

1{{
2    config(
3        materialized="incremental",
4        incremental_strategy="insert_overwrite",
5        alias="daily_sales",
6        partition_by={
7            "field": "date",
8            "data_type": "date",
9            "granularity": "day",
10        }
11    )
12}}
13with
14
15    cleansed_orders as (
16        select *
17        from {{ ref("int__cleansed_orders") }}
18        {% if is_incremental() %}
19            where date(order_time_jst) >= current_date("+9") - 7
20        {% endif %}
21    )
22select
23    date(order_time_jst) as date,
24    sum(sales_jpy) as sales,
25    count(distinct user_id) as payment_uu,
26    safe_divide(sum(sales_jpy), count(distinct user_id)) as arppu
27from cleansed_orders
28group by 1

configのmaterializedにincremental、incremental_strategyにinsert_overwriteを指定し、ifブロックでincremental実行時の期間指定を行います。
出力先のテーブルが存在しないときやfull-refreshしたときはifブロック内は実行されず、全データが再構築されます。
テーブルが存在してfull-refreshしていないときはifブロック内が実行され、指定した期間のデータのみが追加されます。
insert_overwrite戦略は、select結果に含まれるパーティション（例ではdate列）を全て上書きします。

例えば5年分処理していたものを7日分だけ処理するようにした変更してクエリコストが1/100以下になった、というようなケースはよくあります。
ちなみにBigQueryのオンデマンド課金ではクエリ実行時にスキャンしたデータ量に応じて課金されるため、where句でフィルタするデータソース側の列がパーティション指定されている必要があります。

incremental_strategyについて

個人的な所感ですが、基本的にはすべてinsert_overwrite戦略を使っていればいいのかなと思っています。一番シンプルで使いやすいです。
merge戦略はBigQueryでは更新するパーティションが全件となるためコスト削減効果が少なくなってしまいます。
比較的新しいmicrobatchは1テーブルをバッチに分けて更新してくれる機能です。
月ごとや年ごとに分けて更新できるため一度に処理できないようなケースで有効です。全件更新する際も、順にバッチを処理していくので巨大なクエリを発行する必要がありません。

参考: 公式のincremental strategyについての記事
https://docs.getdbt.com/docs/build/incremental-strategy

過去の情報を使う場合の注意

例えば、以下のようなケースです。

1select
2    user_id,
3    date,
4    lag(date, 1) over (
5        partition by user_id order by date
6    ) as previous_access_date,
7    lead(date, 1) over (partition by user_id order by date) as next_access_date
8from daily_users

previous_access_dateは前回アクセスしたユーザーの日付を取得しています。
このクエリを先ほどの方法でincrementalモデルに変更した場合、過去のデータが取得されないため、previous_access_dateが正しくとることができません。
7日以前のデータを出力先のテーブル{{ this }}を参照して取得することで解決することはできますが、複雑度は上がります。

データソースと日付の基準が異なる場合の注意

access_20251210, access_20251211, ... のような日付別シャーディングテーブルをデータソースとしていると、以下のようなクエリで増分化することになります。（_suffix_dateは_table_suffixを元にしたカラムです）

1accesses as (
2    select user_id, datetime(created_at, "+9") as access_time_jst
3    from {{ ref("stg__accesses") }}
4        {% if is_incremental() %}
5    where
6            _suffix_date >= format_date("%y%m%d", current_date("+9") - 7) 
7        {% endif %}
8)

この方法でコストを削減することができますが、日付別シャーディングテーブルの日付基準と集計側の日付の基準が揃っていないケースでは注意が必要です。
例えば、日付別シャーディングテーブルがUTC基準、集計側の日付がJST基準の場合を考えてみましょう。
2025-12-17に処理を実行したとします。増分実行時、取得対象は2025-12-10以降の日付となります。
20251210のテーブルはUTC基準で2025-12-10 00:00:00から2025-12-10 23:59:59までのデータを含みます。JST基準に直すと+9時間されるため、2025-12-10 09:00:00から2025-12-11 08:59:59までのデータを含むことになります。つまり増分実行時は2025-12-10 09:00:00以降のデータが取得されることになります。
2025-12-10のパーティションが上書きされるため、2025-12-10 00:00:00から2025-12-10 08:59:59までのデータが欠損してしまいます。
このような場合は、以下のように条件を追加すれば綺麗に7日分のデータを取得できます。

1{% if is_incremental() %}
2where
3    _suffix_date >= format_date("%y%m%d", current_date("+9") - 8) 
4    and date(created_at, "+9") >= current_date("+9") - 7
5{% endif %}