【データ基盤構築/BigQuery】transaction処理を使用して、テーブルの差分更新を実装する。

今回の課題

BigQueryのトランザクション機能を使って、
データマートを差分更新する際に、冪等性の担保&データ欠損防止ができるように実装したい。

参考：冪等性とは「同じ操作を何度繰り返しても、同じ結果が得られる性質」のこと

トランザクション処理とは

簡潔に言うと、
begin transactionからcommit transactionの中（トランザクション内）の複数のクエリの処理を、
１つの処理として実行して管理する仕組み。

トランザクション内で１でも処理が失敗した場合は、
すべての処理がトランザクション開始前に戻る。
といったイメージ。

参考：BigQueryでトランザクション処理がサポート開始！！

差分更新の実装方法

処理の流れ

1. トランザクション開始
2. TEMPテーブルを生成
3. 差分データ分を更新したいテーブルからDELETE（更新データを削除）
4. TEMPテーブルの内容をINSERT
5. TEMPテーブルの削除
6. トランザクション終了

使用クエリ

※使用しない部分はコメントアウト、使用する部分はコメントアウト解除。
　というかたちで記述しておくと、全量洗い替えしたくなった時でも便利。

/* トランザクションでロールバックできるように調整 */
begin transaction; -- トランザクションはプレビューなので注意

/* 日付関連UDF */
create temp function yesterday() as (
    date('$yesterday$') -- 前日のこと
);

/* この部分は、初回・全量洗い替えの時用 */
-- create or replace table
--     `差分データを追加したいデータマートテーブル`
-- as

/* 一時テーブルを生成 */
create temp table 一時テーブル名 as
select
    *
from
    `元データ`
where 
    date = yesterday() --データを取得したい時点を指定する
;

delete
from
    `差分データを追加したいデータマートテーブル`
where
    date = yesterday() -- 上記で一時テーブルを生成したのと同じ時間のデータをデータマートから削除する。
;

/* 一時テーブルの内容をINSERT */
insert `差分データを追加したいデータマートテーブル`
select
    *
from
    `一時テーブル名`
;

/* 一時テーブルを削除しておく。*/
drop table `一時テーブル名`;

commit transaction;

使用例

実際に業務内でどのように使用したのかメモ

1)1回目の処理でデータマートを生成する

まずは、差分更新が必要なテーブルを用意する。（既にデータマートが存在する場合は、このクエリは不要）

create temp function yesterday() as (
    date('$yesterday$')
);

create or replace table
    `result_table`
as

select
    date
    , fruit
    , revenue
from
    `fruit_revenue_table`
where
    1=1
;

実行結果

date	fruit	revenue
2022-11-01	りんご	200
2022-11-05	りんご	200
2022-11-05	みかん	300

2)2回目以降の処理はこちらのクエリで処理を走らせる

• 前日分のデータが格納された一時テーブルを作る。
• 冪等性担保のために、前日分データをデータマートから除去する。
• データマートに一時テーブルのデータを挿入する。

これで差分更新が実現できる。

create temp function yesterday() as (
    date('$yesterday$') -- 前日のこと
);

create temp table result_table_temp as
select
    *
from
    `fruit_revenue_table`
where 
    date = yesterday() -- 前日の売り上げを取得する
;

delete
from
    `result_table`
where
    date = yesterday() -- 冪等性担保のために、前日の売り上げをresult_tableから除去する。
;

/* 一時テーブルの内容をresult_tableにINSERT */
insert `result_table`
select
    *
from
    result_table_temp
;

/* 一時テーブルを削除しておく。*/
drop table result_table_temp;

commit transaction;

実行結果

date	fruit	revenue
2022-11-01	りんご	200
2022-11-05	りんご	200
2022-11-05	みかん	300
2022-11-06	いちご	300
2022-11-06	りんご	200

以上です。