【Google Cloud】BigqueryのDataformを試してみた

## はじめに
こんにちは。
前回 [【Google Cloud】Bigqueryに組み込まれたDataformとはどのようなサービスなのか？](https://qiita.com/sickboy/items/3f246f11723394717131) という記事を書かせて頂きましたが実際にDataformを軽く試してみました。

## Dataformを試してみる

### パターン①

パターン①としては下記内容を試してみました。

1. test_sales : GCSのファイルをテーブルへロードする
2. test_id_mst : マスタテーブルを登録する
3. test_dm : 項番1、2を結合してデータマート（テーブル）を更新する。更新はCTAS文を実行する。
4. test_dm_snapshot : 項番3のテーブルを利用してスナップショットマートを更新する。更新はDelete＆Insert文を実行する。

最終的には以下のDAGが作成されます。
![drawing0.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/54383d82-9529-c97d-c16d-bdf3b9934f1b.png)


#### 1. test_sales : GCSのファイルをテーブルへロードする

GCSのファイルをload data文で取り込む処理を実装します。
基本のテーブル作成処理ではないので`type: 'operations'`を利用します。
`self`関数は自分自身のテーブル名を参照できる関数です。下記の例ではconfigで明示的にテーブル名を指定していないのでファイル名の`test_sales`がテーブル名となります。プロジェクト名、データセット名もconfigで明示的に指定していない為、デフォルトで設定したものが利用されます。明示的に記載したい場合下記となります。
※ configに設定している`hasOutput`の動作は正確に理解はできていませんが、`type: 'operations'`で`self()`関数を利用する際には必須となるようです。
※ GCSへアクセスするので`Dataform ServiceAccount`には適切なGCSの権限設定が必要です。

```sql
config {
    database: 'your-project-name',
    schema: 'your-dataset-name',
    name: 'your-table-name',
}
```

configにtagを記載することでワークフローをグループ化できます。ワークフロー実行の際にtag単位で実行できるので管理が楽になります。
この例では`tags: ['create_dm']`を記載しています。

```sql:test_sales.sqlx
config {
    type: 'operations',
    hasOutput: true,
    tags: ['create_dm'],
}

create or replace table ${self()}
(
    id int64
    ,sales_amount int64
)
;

load data into ${self()}
from files(
    format = 'CSV',
    field_delimiter = '\t',
    skip_leading_rows = 1,
    uris = ['gs://your-bucket/*.tsv']
)
;
```

利用したデータファイルです。
```text:test_sales.tsv
id	sales_amount
1	100
2	200
3	300
4	400
5	500
1	1000
3	1500
4	1200
```

テーブルをクエリした結果です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/ef6d7005-cac7-591f-9d73-75da5b695141.png)


#### 2. test_id_mst : マスタテーブルを登録する

項番３で実行するSQLのソーステーブルを登録します。
Dataformのワークフロー処理で作成•更新されるテーブルではないテーブルを登録することが可能です。主にテーブル更新の際にソースとして利用されるテーブルになると思います。ソーステーブルを登録しておくことで、DAGに依存関係が表示される様になりテーブルの関係性が分かりやすくなるメリットがあります。

テーブルを作成するSQLです。Bigqueryで直接テーブルを作成しておきます。
```sql
create or replace table temp.test_id_mst
as
with mst as (
  select 1 as id, '壱' as name
  union all
  select 2 as id, '弐' as name
  union all
  select 3 as id, '参' as name
  union all
  select 4 as id, '肆' as name
  union all
  select 5 as id, '伍' as name
)

select * from st
;
```

`type: 'declaration'`を利用してDataformへテーブルを登録します。
明示的にテーブル名を指定していないのでファイル名がテーブル名として登録されます。

```sql:test_id_mst.sqlx
config {
    type: 'declaration',
    schema: 'temp'
}
```

テーブルをクエリした結果です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/d18e4c30-2aa1-4f2c-38b0-fc0a47ab3d28.png)

#### 3. test_dm : 項番1、2を結合してデータマート（テーブル）を更新する。更新はCTAS文を実行する

項番1、2で作成したテーブルをソースとしてテーブルを更新します。
基本的なテーブル作成になる為`type: 'table'`を記載します。`type: 'table'`は実際にはCTASが実行されます。Dataformのメイン処理の1つです。
登録されたソーステーブルやワークフローで作成されたテーブルは`ref`関数で参照できます。
`ref`関数はSQLXファイル（≒テーブル）を参照する関数です。`ref`関数でテーブルを参照することによりテーブル間の依存関係が自動的に登録されます。以下の例ですと、`test_dm`のソースとして`test_sales`と`test_id_mst`が`ref`関数で参照されているので下記のグラフが作成されます。

```mermaid
graph LR;
    test_sales-->test_dm;
    test_id_mst-->test_dm;
```

`ref`関数を利用せずにテーブル名をハードコードすることで同等の処理を実行することも可能ですが、基本的には`ref`関数あってのDataformの価値だと思います。
こちらも明示的にテーブル名を指定していないのでファイル名がテーブル名となります。
検証では利用していませんが、下記の記載も可能です。
- テーブル•カラムのコメント
- パーティショニング、クラスタリング
- ポリシータグ
- メインSQL実行の前後にセット実行できるSQLの記載。例えばメインSQLで作成したテーブルにGrantを実行する。

```sql:test_dm.sqlx
config {
    type: 'table',
    tags: ['create_dm'],
}

select
    fact.id
    ,mst.name
    ,sum(fact.sales_amount) as total_amount
from
    ${ref('test_sales')} as fact
left outer join
    ${ref('test_id_mst')} as mst
on
    fact.id = mst.id
group by
    id
    ,name
```

#### 4. test_dm_snapshot : 項番3のテーブルを利用してスナップショットマートを更新する。更新はDelete＆Insert文を実行する

項番3で作成したテーブルをソースとしてテーブルを更新します。
基本のテーブル作成処理ではないので`type: 'operations'`を利用します。
項番3と同じように`ref`関数でテーブルを参照しているので下記のグラフが作成されます。

```mermaid
graph LR;
    test_dm-->test_dm_snapshot;
```

明示的にテーブル名を指定していないのでファイル名がテーブル名となります。

```sql:test_dm_snapshot
   type: 'operations',
    hasOutput: true,
    tags: ['create_dm'],
}

create table if not exists ${self()}
(
    extract_date date
    ,id int64
    ,name string
    ,total_amount int64
)
;

delete from ${self()} where extract_date = current_date('Asia/Tokyo')
;

insert into ${self()}
select
    current_date('Asia/Tokyo') as extract_date
    ,id
    ,name
    ,total_amount
from
    ${ref('test_dm')}
;
```

#### ワークフローを実行する

最終的なグラフです。
```mermaid
graph LR;
    test_sales-->test_dm;
    test_id_mst-->test_dm;
    test_dm-->test_dm_snapshot;    
```

実行内容のグラフです。
```mermaid
graph LR;    
    Load_Data-->CTAS;
    Declare_Datasorce-->CTAS;
    CTAS-->Delete_Insert;
```

出来上がったワークフローを実行すると`test_dm`、`test_dm_snapshot`ともに正しい結果で実行できました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/1d943508-6deb-2162-fef2-0a2fb8270578.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/3db3d65f-2ae6-74d8-8c3e-2816ac328ad0.png)


### パターン②

パターン②としては下記内容を試してみました。

1. GCSファイルをソースとして外部テーブルを作成する
2. 項番1で作成したテーブルを利用してテーブルを更新する。更新はMerge文を実行する。

最終的には以下のDAGが作成されます。
![drawing1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/08bd436d-7e02-67e7-e69d-6c555a2f2e27.png)

#### 1. GCSファイルをソースとして外部テーブルを作成する

GCSのファイルをソースにcreate external table文で外部テーブルを作成します。
基本のテーブル作成処理ではないので`type: 'operations'`を利用します。
tagは`'merge_table'`を記載しています。
明示的にテーブル名を指定していないのでファイル名がテーブル名となります。
※ パターン１でも記載しましたが、GCSへアクセスするので`Dataform ServiceAccount`には適切なGCSの権限設定が必要です。

```sql:test_sales_external.sqlx
config {
    type: 'operations',
    hasOutput: true,
    tags: ['merge_table'],
}

create or replace external table ${self()}
(
    id int64
    ,sales_amount int64
)
options (
    format = 'CSV',
    field_delimiter = '\t',
    skip_leading_rows = 1,
    uris = ['gs://your-bucket/*.tsv']
)
;
```

利用したデータファイルです。
```text:test_sales_external_1.tsv
id	sales_amount
1	100
2	200
3	300
4	400
5	500
```

テーブルをクエリした結果です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/fde76254-7bf7-468d-b65a-5d11074962a3.png)


#### 2. 項番1で作成したテーブルを利用してテーブルを更新する。更新はMerge文を実行する

Mergeでのデータ更新になる為`type: 'incremental'`を記載します。`type: 'incremental'`は実際にはInsertかMergeが実行されます。configに`uniqueKey`を指定するとMerge、指定しないとInsert処理となります。Dataformのメイン処理の1つです。`uniqueKey`はMergeを実行する際のテーブル結合キーです。
初回実行時（正確にはテーブルが存在しない場合だと思われる）にはCTASが実行されテーブルが作成されます。２回目以降（正確にはテーブルが存在する場合と思われる）は、MergeまたはInsertが実行されます。Merge、Insert処理を成功させるにはターゲットテーブル（ここでの例だと`test_sales_merge`）とソーステーブル（ここでの例だと`test_sales_external`）が同じカラム構成を保っている必要があります。カラム変更が発生する場合はCTASでターゲットテーブルを再作成することも可能です。
この例でも`ref`関数を利用しているので以下のグラフが作成されます。

```mermaid
graph LR;
    test_sales_external-->test_sales_merge;
```

明示的にテーブル名を指定していないのでファイル名がテーブル名となります。

```sql:test_sales_merge.sqlx
config {
    type: 'incremental',
    uniqueKey: ['id'],
    tags: ['merge_table'],
}

select 
    id
    ,sales_amount
from
    ${ref('test_sales_external')}
```

#### ワークフローを実行する

最終的なグラフです。
```mermaid
graph LR;
    test_sales_external-->test_sales_merge;   
```

実行内容のグラフです。
```mermaid
graph LR;    
    Create_External_Table-->Merge;
```

出来上がったワークフローを実行すると`test_sales_merge`が正しい結果で実行できました（テーブルも作成されています）。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/e9145d58-164d-8e21-bf69-02cb51a2e87d.png)

２回目以降、Merge処理が正しく実行されているか確認してみます。
初回実行で利用した`test_sales_external_1.tsv`をGCSから削除の上データファイルを置き換えます。
`id 5`の`sales_amount`がUpdateされ`id 6`がInsertされていれば成功です。

```text:test_sales_external_2.tsv
id	sales_amount
6	6000
5	5000
```

正しい結果で実行できました。
`id 5のsales_amountが500から５０００`にupdateされ`id 6`がInsertされています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3300248/47c97541-52fa-af21-ffa0-f92497cf3701.png)

## 終わりに
簡単な検証ではありますが、実装もめちゃくちゃ簡単でした。SQLさえ書ければすぐに実装できるので抵抗なく利用できそうです。
後はBigquery上からのスケジュール実行や、エラーが発生した際にリトライ•通知（理想はSlack）が出来るようになると更に使いやすくなると思います。
ありがとうございました。

## 参照

Google CloudのDataform公式ドキュメント

https://cloud.google.com/dataform/docs

Dataform自体のドキュメント（旧？）

https://docs.dataform.co/