SCD Type2 のテーブルを、dbt 標準の incremental model で管理する
dbt の Snapshot を使わずに、incremental model で SCD Type2 テーブルを管理する方法を紹介します。
2026.07.31
こんにちは、川田です。
今回は dbt にて SCD Type2 のテーブルを管理するに、dbt Snapshot を利用せず、標準の incremental model で実現する方法を紹介します。
環境
- dbt Core 1.12.0
- 利用 database: Snowflake
利用データ
以下の seeds/raw_products.csv を用意しています。
product_id,product_name,type,price,created,modified
111,キャベツ,野菜,200,2026-07-01 10:00:00,
222,たまねぎ,野菜,300,2026-07-01 10:00:00,
333,にんじん,野菜,250,2026-07-01 10:00:00,
111,キャベツ,野菜,230,2026-07-01 10:00:00,2026-07-02 10:00:00
222,たまねぎ,野菜,280,2026-07-01 10:00:00,2026-07-03 10:00:00
テーブルを作成しておきます。
$ dbt seed --select seeds/raw_products.csv
12:01:24 Running with dbt=1.12.0
12:01:24 Registered adapter: snowflake=1.12.0
12:01:25 Found 2 models, 1 seed, 8 data tests, 661 macros, 3 unit tests
12:01:25
12:01:25 Concurrency: 4 threads (target='dev')
12:01:25
12:01:27 1 of 1 START seed file DBT.raw_products ........................................ [RUN]
12:01:29 1 of 1 OK loaded seed file DBT.raw_products .................................... [INSERT 5 in 1.61s]
12:01:29
12:01:29 Finished running 1 seed in 0 hours 0 minutes and 4.29 seconds (4.29s).
12:01:29
12:01:29 Completed successfully
12:01:29
12:01:29 Done. PASS=1 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=1
最終的に、以下のような SCD Type2 が作成されることを期待しています。
| product_id | product_name | type | price | valid_from | valid_to | ingested_at |
|---|---|---|---|---|---|---|
| 111 | キャベツ | 野菜 | 200 | 2026-07-01 10:00:00 | 2026-07-02 10:00:00 | YYYY-MM-DD HH:MM:SS |
| 111 | キャベツ | 野菜 | 230 | 2026-07-02 10:00:00 | 2999-12-31 00:00:00 | YYYY-MM-DD HH:MM:SS |
| 222 | たまねぎ | 野菜 | 300 | 2026-07-01 10:00:00 | 2026-07-03 10:00:00 | YYYY-MM-DD HH:MM:SS |
| 222 | たまねぎ | 野菜 | 280 | 2026-07-03 10:00:00 | 2999-12-31 00:00:00 | YYYY-MM-DD HH:MM:SS |
| 333 | にんじん | 野菜 | 250 | 2026-07-02 10:00:00 | 2999-12-31 00:00:00 | YYYY-MM-DD HH:MM:SS |
作成 dbt model
staging
staging の model models/staging/stg_products.sql を用意しておきます。
{{
config(
materialized="view",
)
}}
select * from {{ ref('raw_products') }}
marts
実際に SCD Type2 となる model models/marts/scd_products.sql は下記となります。
{{
config(
materialized="incremental",
incremental_strategy="merge",
unique_key=[
"product_id",
"valid_from"
]
)
}}
-- dbt の実行開始日時（run_started_at）から 1 日前を YYYY-MM-DD 形式で取得
{% set yesterday = (run_started_at - modules.datetime.timedelta(days=1)).strftime('%Y-%m-%d') %}
with source_rows as (
-- 上流のテーブルから新規追記されたレコードを取得
select
product_id,
product_name,
type,
price,
coalesce(modified, created)::timestamp_ntz as valid_from,
from {{ ref('stg_products') }}
{%- if is_incremental() %}
-- incremental 時は、標準で前日以降の作成/更新レコードを読む
where
coalesce(modified, created) >= '{{ var("lookback_date", yesterday) }}'::timestamp_ntz
{%- endif %}
),
delta_start as (
-- 既存テーブルで影響受けるレコードを抽出するため、追記された product_id と最古「適用開始日時 (valid_from)」を取得
select
product_id,
min(valid_from) as min_valid_from
from source_rows
group by product_id
),
affected_target as (
{%- if is_incremental() %}
-- delta_start を利用して、既存テーブルから影響を受けるレコードを抽出
select
t.product_id,
t.product_name,
t.type,
t.price,
t.valid_from,
from {{ this }} as t
inner join delta_start as d
on t.product_id = d.product_id
-- 「適用終了日時 (valid_to)」が、追記レコードの「適用開始日時 (valid_from)」より新しいレコードを抽出
and t.valid_to > d.min_valid_from
{%- else %}
-- 非 incremental 時向けの dummy クエリ
select
null::varchar as product_id,
null::varchar as product_name,
null::varchar as type,
null::number as price,
null::timestamp_ntz as valid_from,
where false
{%- endif %}
),
recalculation_base as (
-- 「既存テーブル内の影響を受けるレコード」と「新規追記されたレコード」を集合
select
product_id,
product_name,
type,
price,
valid_from,
1 as priority
from affected_target
union all
select
product_id,
product_name,
type,
price,
valid_from,
2 as priority
from source_rows
),
deduplicated as (
-- 履歴を管理するキー（product_id, valid_from）が重複した場合、source 側を優先する（既存テーブルを上書きする）
select *
from recalculation_base
qualify row_number() over (
partition by product_id, valid_from
order by priority desc
) = 1
),
final as (
-- 「適用終了日時 (valid_to) 」の値を導出
select
product_id,
product_name,
type,
price,
valid_from,
coalesce(
lead(valid_from) over (
partition by product_id
order by valid_from
),
'2999-12-31 00:00:00'::timestamp_ntz
) as valid_to,
sysdate() as ingested_at,
from deduplicated
)
select * from final
model を実行
model を実行します。
$ dbt run --select +scd_products
14:37:49 Running with dbt=1.12.0
14:37:49 Registered adapter: snowflake=1.12.0
14:37:50 Found 2 models, 1 seed, 8 data tests, 661 macros, 3 unit tests
14:37:50
14:37:50 Concurrency: 4 threads (target='dev')
14:37:50
14:37:51 1 of 2 START sql view model DBT.stg_products ................................... [RUN]
14:37:52 1 of 2 OK created sql view model DBT.stg_products .............................. [SUCCESS 1 in 0.36s]
14:37:52 2 of 2 START sql incremental model DBT.scd_products ............................ [RUN]
14:37:53 2 of 2 OK created sql incremental model DBT.scd_products ....................... [SUCCESS 5 in 1.05s]
14:37:53
14:37:53 Finished running 1 incremental model, 1 view model in 0 hours 0 minutes and 2.97 seconds (2.97s).
14:37:53
14:37:53 Completed successfully
14:37:53
14:37:53 Done. PASS=2 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=2
期待した結果となっています。
dbt unit test を作成
dbt unit test を、以下のテストケースにて作成します。
- full refresh 時の処理
- 通常の、新規レコード追加時の increment 処理
- イレギュラーとなる、遅延データが発生した場合の処理
models/marts/scd_products_tests.yml。
version: 2
unit_tests:
- name: scd_products_full_refresh
description: "フルリフレッシュ動作時のテスト"
model: scd_products
overrides:
macros:
# フルリフレッシュ処理とする
is_incremental: false
given:
- input: ref('stg_products')
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
created: "2026-07-01 10:00:00"
modified: null
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
created: "2026-07-01 10:00:00"
modified: "2026-07-02 10:00:00"
- product_id: "002"
product_name: たまねぎ
type: 野菜
price: 300
created: "2026-07-01 10:00:00"
modified: null
expect:
rows:
- product_id: "001"
price: 200
valid_from: "2026-07-01 10:00:00"
valid_to: "2026-07-02 10:00:00"
- product_id: "001"
price: 230
valid_from: "2026-07-02 10:00:00"
valid_to: "2999-12-31 00:00:00"
- product_id: "002"
price: 300
valid_from: "2026-07-01 10:00:00"
valid_to: "2999-12-31 00:00:00"
- name: scd_products_append_latest_record
description: "通常のレコード更新時"
model: scd_products
overrides:
macros:
is_incremental: true
vars:
# 2026-07-05 以降の新規レコードを反映させる
lookback_date: "2026-07-05 00:00:00"
given:
- input: ref('stg_products')
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
created: "2026-07-01 10:00:00"
modified: null
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
created: "2026-07-01 10:00:00"
modified: "2026-07-02 10:00:00"
# 下記レコードが新規到着したとする
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 270
created: "2026-07-01 10:00:00"
modified: "2026-07-05 10:00:00"
- input: this
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
valid_from: "2026-07-01 10:00:00"
valid_to: "2026-07-02 10:00:00"
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
valid_from: "2026-07-02 10:00:00"
valid_to: "2999-12-31 00:00:00"
# 更新 1 件 / 追記 1 件の merge 処理となる
expect:
rows:
- product_id: "001"
price: 230
valid_from: "2026-07-02 10:00:00"
valid_to: "2026-07-05 10:00:00" # この値が更新される
# 下記レコードが追記される
- product_id: "001"
price: 270
valid_from: "2026-07-05 10:00:00"
valid_to: "2999-12-31 00:00:00"
- name: scd_products_insert_late_arriving_record
description: "遅延到着によって既存期間を分割して更新される"
model: scd_products
overrides:
macros:
is_incremental: true
vars:
# 2026-07-01 以降の新規レコードを反映させる
lookback_date: "2026-07-01 00:00:00"
given:
- input: ref('stg_products')
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
created: "2026-07-01 10:00:00"
modified: null
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
created: "2026-07-01 10:00:00"
modified: "2026-07-02 10:00:00"
# 新たに「2026-07-02 6時更新のデータ」が届いたとする
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 220
created: "2026-07-01 10:00:00"
modified: "2026-07-02 06:00:00"
- input: this
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
valid_from: "2026-07-01 10:00:00"
valid_to: "2026-07-02 10:00:00"
# SCD テーブル側には「2026-07-02 10:00」のデータが存在するとする
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
valid_from: "2026-07-02 10:00:00"
valid_to: "2999-12-31 00:00:00"
# 更新 2 件 / 追記 1 件の merge 処理となる
expect:
rows:
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 200
valid_from: "2026-07-01 10:00:00"
valid_to: "2026-07-02 06:00:00" # この値が更新される
# 下記レコードが追記される
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 220
valid_from: "2026-07-02 06:00:00"
valid_to: "2026-07-02 10:00:00"
# 下記レコードが追記される（lookback_date 範囲内のため再出力対象となる）
- product_id: "001"
product_name: キャベツ
type: 野菜
price: 230
valid_from: "2026-07-02 10:00:00"
valid_to: "2999-12-31 00:00:00"
確認します。
$ dbt test --select models/marts/scd_products_tests.yml
14:46:11 Running with dbt=1.12.0
14:46:11 Registered adapter: snowflake=1.12.0
14:46:12 Found 2 models, 1 seed, 8 data tests, 661 macros, 3 unit tests
14:46:12
14:46:12 Concurrency: 4 threads (target='dev')
14:46:12
14:46:13 1 of 3 START unit_test scd_products::scd_products_append_latest_record ......... [RUN]
14:46:13 3 of 3 START unit_test scd_products::scd_products_insert_late_arriving_record .. [RUN]
14:46:13 2 of 3 START unit_test scd_products::scd_products_full_refresh ................. [RUN]
14:46:15 1 of 3 PASS scd_products::scd_products_append_latest_record .................... [PASS in 2.21s]
14:46:15 2 of 3 PASS scd_products::scd_products_full_refresh ............................ [PASS in 2.18s]
14:46:15 3 of 3 PASS scd_products::scd_products_insert_late_arriving_record ............. [PASS in 2.70s]
14:46:16
14:46:16 Finished running 3 unit tests in 0 hours 0 minutes and 4.40 seconds (4.40s).
14:46:16
14:46:16 Completed successfully
14:46:16
14:46:16 Done. PASS=3 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=3
dbt data test を作成
Generic data test と Singular data test を用意します。
Generic data test
純粋な SQL のロジックから導かれるデータ状況をテストで確認します。
models/marts/scd_products.yml。
version: 2
models:
- name: scd_products
data_tests:
- dbt_utils.unique_combination_of_columns:
arguments:
description: "product_id と valid_from の複合ユニークキーとなる"
combination_of_columns:
- product_id
- valid_from
config:
store_failures: true
- dbt_utils.mutually_exclusive_ranges:
arguments:
description: "同一 product_id 内で valid_from 〜 valid_to の期間が重複しないこと"
lower_bound_column: valid_from
upper_bound_column: valid_to
partition_by: product_id
gaps: not_allowed
config:
store_failures: true
columns:
- name: product_id
data_type: number
data_tests:
- not_null:
description: "null 禁止"
config:
store_failures: true
- name: product_name
data_type: varchar
- name: type
data_type: varchar
- name: price
data_type: number
- name: valid_from
data_type: timestamp_ntz
data_tests:
- not_null:
description: "null 禁止"
config:
store_failures: true
- dbt_utils.expression_is_true:
description: "valid_from は valid_to より早い日時となる"
arguments:
expression: "< valid_to"
config:
store_failures: true
- name: valid_to
data_type: timestamp_ntz
data_tests:
- not_null:
description: "null 禁止"
config:
store_failures: true
- name: ingested_at
data_type: timestamp_ntz
data_tests:
- not_null:
description: "null 禁止"
config:
store_failures: true
動作確認します。
$ dbt test --select test_type:generic
14:59:20 Running with dbt=1.12.0
14:59:20 Registered adapter: snowflake=1.12.0
14:59:21 Found 2 models, 1 seed, 8 data tests, 661 macros, 3 unit tests
14:59:21
14:59:21 Concurrency: 4 threads (target='dev')
14:59:21
14:59:23 1 of 7 START test dbt_utils_expression_is_true_scd_products_valid_from___valid_to [RUN]
14:59:23 3 of 7 START test dbt_utils_unique_combination_of_columns_scd_products_product_id__valid_from [RUN]
14:59:23 4 of 7 START test not_null_scd_products_ingested_at ............................ [RUN]
14:59:23 2 of 7 START test dbt_utils_mutually_exclusive_ranges_scd_products_not_allowed__valid_from__product_id__valid_to [RUN]
14:59:24 2 of 7 PASS dbt_utils_mutually_exclusive_ranges_scd_products_not_allowed__valid_from__product_id__valid_to [PASS in 0.91s]
14:59:24 5 of 7 START test not_null_scd_products_product_id ............................. [RUN]
14:59:24 1 of 7 PASS dbt_utils_expression_is_true_scd_products_valid_from___valid_to .... [PASS in 1.12s]
14:59:24 6 of 7 START test not_null_scd_products_valid_from ............................. [RUN]
14:59:24 4 of 7 PASS not_null_scd_products_ingested_at .................................. [PASS in 1.32s]
14:59:24 7 of 7 START test not_null_scd_products_valid_to ............................... [RUN]
14:59:24 3 of 7 PASS dbt_utils_unique_combination_of_columns_scd_products_product_id__valid_from [PASS in 1.36s]
14:59:24 5 of 7 PASS not_null_scd_products_product_id ................................... [PASS in 0.87s]
14:59:25 6 of 7 PASS not_null_scd_products_valid_from ................................... [PASS in 1.10s]
14:59:25 7 of 7 PASS not_null_scd_products_valid_to ..................................... [PASS in 0.93s]
14:59:25
14:59:25 Finished running 7 data tests in 0 hours 0 minutes and 4.16 seconds (4.16s).
14:59:25
14:59:25 Completed successfully
14:59:25
14:59:25 Done. PASS=7 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=7
Singular data test
staging 側のデータも踏まえて、純粋な SQL のロジックから導かれるデータ状況をテストで確認します。
tests/assert_scd_active_records_match_stg.yml 。
data_tests:
- name: assert_scd_active_records_match_stg
description: |-
- SCD テーブル向けテスト
- ステージング側のユニークな商品数と、SCDの最新レコード数は常に一致する
tests/assert_scd_active_records_match_stg 。
with stg_summary as (
select count(distinct product_id) as unique_products_count
from {{ ref('stg_products') }}
),
scd_summary as (
select count(*) as active_products_count
from {{ ref('scd_products') }}
where
valid_to = '2999-12-31'::timestamp_ntz
)
select
stg_summary.unique_products_count,
scd_summary.active_products_count
from stg_summary
cross join scd_summary
where
stg_summary.unique_products_count != scd_summary.active_products_count
動作確認します。
$ dbt test --select test_type:singular
15:03:06 Running with dbt=1.12.0
15:03:07 Registered adapter: snowflake=1.12.0
15:03:07 Found 2 models, 1 seed, 8 data tests, 661 macros, 3 unit tests
15:03:07
15:03:07 Concurrency: 4 threads (target='dev')
15:03:07
15:03:08 1 of 1 START test assert_scd_active_records_match_stg .......................... [RUN]
15:03:09 1 of 1 PASS assert_scd_active_records_match_stg ................................ [PASS in 0.96s]
15:03:09
15:03:09 Finished running 1 test in 0 hours 0 minutes and 2.39 seconds (2.39s).
15:03:09
15:03:09 Completed successfully
15:03:09
15:03:09 Done. PASS=1 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=1
