1 億行を超えるイベントログを毎回 `table` materialization で全件書き直すと、 なら 1 回 $50、ビルド 30 分が当たり前になります。incremental は 新規/更新分だけ書く仕組みで、これを 1/100 程度に抑えます。
incremental の最小例
{{ config( materialized='incremental', unique_key='event_id', incremental_strategy='merge', on_schema_change='sync_all_columns') }}
SELECT event_id, user_id, event_type, occurred_at, payloadFROM {{ source('events', 'raw') }}
{% if is_incremental() %} -- 過去の最新時刻 + 1 時間バッファでオーバーラップ取り込み WHERE occurred_at >= ( SELECT DATE_SUB(MAX(occurred_at), INTERVAL 1 HOUR) FROM {{ this }} ){% endif %}`is_incremental()` は「初回でない & --full-refresh でない」とき true。ここで上流から 増分だけ を切り出す WHERE 条件を書く。
4 つの incremental_strategy
| strategy | 挙動 | 向く場面 | |
|---|---|---|---|
| append | INSERT only、重複可能 | イミュータブルログ | 全 DWH |
| merge | unique_key で UPSERT | 再処理ありのイベント | ///Postgres |
| delete+insert | 対象行を消して入れ直す | merge 不可な DWH | Postgres/Redshift |
| insert_overwrite | パーティション単位で置換 | BQ パーティションテーブル | BigQuery のみ |
戦略 1: append (最速、重複に注意)
{{ config( materialized='incremental', incremental_strategy='append') }}
SELECT * FROM {{ source('events', 'raw') }}
{% if is_incremental() %} WHERE occurred_at > (SELECT MAX(occurred_at) FROM {{ this }}){% endif %}WHERE 条件のオーバーラップで重複が出る。`MAX(occurred_at)` で切ると、同じ秒に 2 イベントあると後続が漏れる/重複する。INSERT前の重複排除を model 側で書く or merge 戦略を選ぶ。
戦略 2: merge (一番安全、推奨)
{{ config( materialized='incremental', unique_key='event_id', incremental_strategy='merge', merge_update_columns=['payload', 'updated_at'], -- 更新するカラムを限定 on_schema_change='sync_all_columns') }}
SELECT event_id, user_id, event_type, occurred_at, payload, CURRENT_TIMESTAMP() AS updated_atFROM {{ source('events', 'raw') }}
{% if is_incremental() %} -- 過去 24h を再取り込み (遅延到着・修正データに対応) WHERE occurred_at >= DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY){% endif %}戦略 4: insert_overwrite (BigQuery 専用、最も効率的)
{{ config( materialized='incremental', incremental_strategy='insert_overwrite', partition_by={ 'field': 'event_date', 'data_type': 'date', 'granularity': 'day' }, partitions=['DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)', 'CURRENT_DATE()']) }}
SELECT DATE(occurred_at) AS event_date, event_id, user_id, event_typeFROM {{ source('events', 'raw') }}
{% if is_incremental() %} WHERE DATE(occurred_at) IN ( DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY), CURRENT_DATE() ){% endif %}指定パーティションだけを丸ごと置換するため、冪等で速くて安い。`partitions` で「再処理対象」を明示するのがポイント。
on_schema_change の挙動
| 設定 | 新カラム追加時の挙動 |
|---|---|
| `ignore` | 新カラムは無視、既存スキーマで継続 |
| `fail` | 失敗してエラー |
| `append_new_columns` | 新カラムを追加、既存行は NULL |
| `sync_all_columns` | 新カラム追加 + 削除カラムも反映 (推奨) |
本番運用パターン
- 毎日 0:00 に full-refresh、毎時 incremental ── 1 日 1 回はクリーンビルドで安心
- 遅延到着対策: WHERE 条件で「現在 - 24h」分を毎回再取り込み (overlap)
- partition by date + cluster by user_id 等の でクエリ高速化
- 監視: ビルド時間・スキャンサイズの推移を Looker でダッシュ化
- 急増アラート: 通常の 3 倍以上スキャンしたら Slack 通知
incremental の落とし穴チェックリスト
イベントが「過去日付」で後から到着するケース。WHERE で `MAX(occurred_at) - X 日` を引くことで取りこぼしを防ぐ。
`event_id` のつもりが上流で重複している場合、merge 戦略では 「どっちを残すか」が不定。 な ID 設計と、staging で `dbt_utils.deduplicate` を入れる。
「過去 6 ヶ月分の集計を計算しなおして」と言われたとき、incremental だと 手動で `--full-refresh` が必要。スケジューラに 月 1 回 full-refresh を組み込んでおく。
次の話
EP.12 では Slim CI を使った CI/CD パイプラインを扱います。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。