【dbt × BigQuery】増分更新について挙動を詳しく確認してみる【insert_overwrite（Dynamic partitions）, append】株式会社ライトコード

こんにちは！

普段データ関連のお仕事をしています。

現在所属しているチームではデータマートの作成の際にdbtを使用しており、コスト削減のために差分更新の対応中です。

差分更新処理を書いていると「あれ？どんな挙動になるんだっけ？」ということが多いので、そんな自分に向けて増分更新の詳しい挙動を確認して、備忘録がてらブログを書いていきたいと思います。

環境・準備

BigQuery
dbt

差分更新の種類

僕が所属しているチームでは「insert_overwrite（Dynamic partitions）」方式と「append」方式を採用しているので、今回はその2つについて挙動を見ていきたいと思います。

insert_overwrite（Dynamic partitions）

下記のモデルの挙動を確認していきます。

{{

    config(

        partition_by={

            "field": "time_jst",

            "data_type": "datetime",

            "granularity": "day",

        },

        cluster_by=["user_id"],

        materialized="incremental",

        incremental_strategy="insert_overwrite",

    )

}}



select user_id, datetime(timestamp, "+9") as time_jst

from {{ ref("source_table_name") }}

{% if is_incremental() %} 

where create_date >= current_date("+9") - 7 

{% endif %}

初回実行時は、下記のようなクエリが生成されます。

create or replace table `create_table_name`

  

partition by datetime_trunc(time_jst, day)

cluster by user_id

as (

  select user_id, datetime(timestamp, "+9") as time_jst

  from `source_table_name`

);

これで「create_table_name」というテーブルが作成されます。

そして、差分更新をするために再度実行してみると下記のようなクエリが生成されました。

create or replace table `create_table_name__dbt_tmp`

  

partition by datetime_trunc(time_jst, day)

cluster by user_id

as (

  select user_id, datetime(timestamp, "+9") as time_jst

  from `source_table_name`

  where create_date >= current_date("+9") - 7 

);



-- generated script to merge partitions into `create_table_name`

declare dbt_partitions_for_replacement array;



-- 1. temp table already exists, we used it to check for schema changes

-- 2. define partitions to update

set (dbt_partitions_for_replacement) = (

    select as struct

        -- IGNORE NULLS: this needs to be aligned to _dbt_max_partition, which ignores null

        array_agg(distinct datetime_trunc(time_jst, day) IGNORE NULLS)

    from `create_table_name__dbt_tmp`

);



-- 3. run the merge statement

merge into `create_table_name` as DBT_INTERNAL_DEST

    using (

    select

    * from `create_table_name__dbt_tmp`

  ) as DBT_INTERNAL_SOURCE

    on FALSE

when not matched by source

      and datetime_trunc(DBT_INTERNAL_DEST.time_jst, day) in unnest(dbt_partitions_for_replacement) 

    then delete

when not matched then insert

    (`user_id`, `time_jst`)

values

    (`user_id`, `time_jst`)

;



-- 4. clean up the temp table

drop table if exists `create_table_name__dbt_tmp`

詳しく見ていきます。

まず1~9行目

create or replace table `create_table_name__dbt_tmp`

  

partition by datetime_trunc(time_jst, day)

cluster by user_id

as (

  select user_id, datetime(timestamp, "+9") as time_jst

  from `source_table_name`

  where create_date >= current_date("+9") - 7 

);