PostgreSQLのデータをCDC形式で転送し、BigQueryに自動同期する

## 概要
PostgreSQLからBigQueryにデータを転送する際、転送の設定が複雑で面倒に感じることがあるかもしれません。
そこで、今回は[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)というETL & ワークフローサービスを使い、CDC形式で転送する方法をご紹介します。
CDC形式を取り入れることで、効率的で低レイテンシなデータ転送を実現できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/8d4d7c1a-86ef-f146-a3b6-273644ea60eb.png)


### 1-0. CDCとは？
今回はPostgreSQLのWALをCDC(Change Data Capture = 変更データキャプチャ)形式で転送します。
これにより、初回のみ全件転送、それ以後は差分転送をして、転送量の削減・転送の高速化を図ることができます。

DBへの負荷を抑えつつバッチ処理でのデータ同期を行いたいときには、CDC形式での転送がうってつけです。

全件転送との比較は以下の通りです。


|                                  | 全件転送                 | CDC                                  | 
| -------------------------------- | ------------------------ | ------------------------------------ | 
| trocco転送量(初回実行時)         | ソーステーブルの全データ | ソーステーブルの全データ             | 
| trocco転送量(2回目以降)          | ソーステーブルの全データ | 更新ログのみ(※1)                    | 
| スキーマ変更発生時<br>(列追加等) | 自動追従                 | 自動追従                             | 
| データ削除(DELETE)の挙動         | 物理削除                 | 論理削除                             | 
| データ更新(UPDATE)の挙動         | 物理更新                 | 物理更新                             | 
| BigQueryスキャン量               | なし                     | 現在のBigQuery上のテーブル全量+更新ログ    | 
| データソースDBへの負荷           | 高                       | 低（初回転送時のみ全量のため高負荷） | 

※1 INSERT,DELETE,DDL文については1行文、UPDATE文については更新前後の2行分

[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)でCDC転送をする際には
(1)[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)がPostgreSQLのWALをBigQuery上に転送
(2)WALのデータから重複を排除し、それぞれのレコードの最新のレコードの集合を作成
(3)↑の集合を現在のBigQueryテーブルとマージし、マージ後のテーブルで置換する
という手順を踏んでいます。

### 2-1. MySQLのパラメータ設定
[troccoのドキュメント](https://docs.trocco.io/s/article/PostgreSQL-to-BigQuery%E8%BB%A2%E9%80%81-CDC)を参照しつつ、PostgreSQLのパラメータ設定をします。今回はRDSを使っていますので、AWSのコンソールから環境変数を確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/bd9c8aa5-49a0-1dc4-a10f-325542e29478.png)

rds.logical_replicationは1に設定されています。
max_replication_slotsとmax_wal_sendersも連携するテーブルの数と同じになっていることが確認できたので、このまま次のステップに進めます。

### 2-2. Slotの作成
[troccoのドキュメント](https://docs.trocco.io/s/article/PostgreSQL-to-BigQuery%E8%BB%A2%E9%80%81-CDC)を参照しつつ、Slotの作成をします。

SUPERUSERでログイン後、下記SQLでスロットを作成してください。

```SQL
SELECT * FROM pg_create_logical_replication_slot('trocco_<table_name>', 'wal2json');
```

作成後、下記のSQLでSlotからデータが取得できるかの確認ができます。

```SQL
SELECT COUNT(*) FROM pg_logical_slot_peek_changes('<replication_slot_name>', null, null);
```

### 2-3. PostgreSQLの権限設定
[troccoのドキュメント](https://docs.trocco.io/s/article/PostgreSQL-to-BigQuery%E8%BB%A2%E9%80%81-CDC)を参照しつつ、PostgreSQLの権限設定をします。

ここではCONNECT, USAGE, SELECT, rds_replication, ALTER DEFAULT PRIVILEGES の5つの権限を設定します。
（rds_replicationは、RDSの場合のみ必要となります）

```SQL
GRANT CONNECT ON DATABASE <database_name> TO <username>;
GRANT USAGE ON SCHEMA <schema_name> TO <username>;
GRANT SELECT ON <schema_name>.<table_name> TO <username>;
GRANT rds_replication TO <username>;
ALTER DEFAULT PRIVILEGES IN SCHEMA <schema_name> GRANT SELECT ON TABLES TO <username>;
```

### 2-4. Primary Keyの設定
転送元のテーブルにはPrimary Keyが設定されている必要があります。
ここでは既に設定されているものとして、割愛します。

その他、[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)で転送する際の必須条件などについては[こちらのドキュメント](https://docs.trocco.io/s/article/PostgreSQL-to-BigQuery%E8%BB%A2%E9%80%81-CDC)を参照してください。

これにてPostgreSQL側での設定は完了です。
次に、[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)の設定をします。

### 3-0. [trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)に登録
[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)のアカウントが必要です。
フリープランがありますので、前もって申し込み・登録をしておいてください！
[https://trocco.io/lp/inquiry_free.html](https://trocco.io/lp/inquiry_free.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)

### 3-1. 転送元・転送先を決定
[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)にアクセスし、ダッシュボードから「転送設定を作成」のボタンを押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/081f5586-55bc-f8ad-71e3-5ec41d44e515.png)

転送元にPostgreSQL WAL(CDC)、転送先にBigQueryを選択し、転送設定作成ボタンを押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/21c8b121-c5c9-b0f4-924b-7ece0a5bcab9.png)
すると、設定画面になるので、転送に必要な情報を入力していきます。

### 3-2. PostgreSQLとの連携設定
転送設定の名前とメモを入力します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/9ac337b9-4aca-15ee-57f4-3901106ba71f.png)

転送設定の名前を決めたら、「転送元の設定」内の「接続情報を追加」ボタンを押し、PostgreSQLの接続情報の設定をします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/146a3d72-5760-8efe-b2b3-8119bf4a0759.png)

接続設定の名前・ホスト・ポート・ユーザー名・パスワードを入力します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/8343f637-6fcd-8ec0-4e81-dc0c944b11c6.png)

入力をしたのち、設定の保存をします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/5a55be5d-9a5e-1f26-3991-15b16be0bac3.png)


再び転送設定画面に戻り、「接続情報を読み込む」ボタンを押すと、作成した接続情報が選択できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/2d3d4a8d-567a-8250-301a-6974f20d5754.png)


### 3-3. PostgreSQLからのデータ抽出設定
これでPostgreSQLとの連携は完了です。
次に、PostgreSQLの取得データを設定します。  
まずは必須項目の「データベース名」「スキーマ名」「テーブル」「スロット」をセレクトボックスの中から選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/c3d6fbec-2152-6c81-fd11-1e6fdf7648d6.png)

次に、スキーマ追従の有効/無効を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/49eefca8-527a-2df5-da8d-75a5ebb00c86.png)

スキーマ追従が有効になっている場合、

   - クエリにRENAME COLUMNが含まれている場合、変更後の名前でデータ連携される
   - クエリにADD COLUMNが含まれている場合、追加したテーブルが自動で転送されるようになる

といった挙動になります。

最後に接続確認が通るか確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/4b2c0105-7b3e-0132-4089-1209ae19065c.png)


大丈夫ですね。以上でPostgreSQL側の設定は完了です。
次は転送先のBigQueryの設定をしていきましょう。

### 3-4. 転送先BigQueryの設定
「接続情報を追加」ボタンからBigQueryの接続設定を行います。名前と、認証方式としてサービスアカウント(JSONキー)を入力します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/38a485ad-e9ce-2738-03cb-66cb6b65212c.png)

JSONキーの取得方法については[こちらのドキュメント](https://docs.trocco.io/s/article/BigQuery%E3%81%AE%E6%8E%A5%E7%B6%9A%E8%A8%AD%E5%AE%9A)を参照してください。

「接続情報を追加」ボタンからBigQueryの接続設定を行い、データベース・テーブル・データセットのロケーションを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/255acd12-1691-3582-71eb-ceffc1dc4afc.png)


最後に、接続確認が問題なく通るか確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/f27137e5-758e-1761-cfc0-0ce79fbeaf51.png)

これで入力は完了です。「次のステップへ」をクリックして次に進みましょう。

### 3-5. カラム名・データ型の確認
「データプレビュー」の画面でカラム名・データ型の確認をします。
PostgreSQLのデータは自動的にBigQueryのデータ型に変換されます。

データ型については[こちらのドキュメント](https://docs.trocco.io/s/article/PostgreSQL-to-BigQuery%E8%BB%A2%E9%80%81-CDC)をご参照ください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/ac33e623-d9a8-05e1-a36b-fcc64f963e7c.png)

問題なければ、確認画面に移り、保存して適用しましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/45cfa14d-47f5-0ad0-2337-34020c364ebd.png)



### 3-6. 初回転送
CDCは変更データのみを転送する方式ですが、初回だけは全件転送をする必要があります。
このステップでは、手動で全件転送をします。

右上にある「実行」ボタンを押します。

![saiki-fix.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/61228/db0a9327-61b6-23d0-bd2b-cf135930e46a.png)

「全件転送を行う」にチェックマークを入れて、「ジョブを実行」を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/318c35c8-0136-fd29-737d-46f0ab30e2bf.png)

転送完了まで待ちます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/cc5afa72-65ba-f646-7104-08c47021c9ec.png)

転送が成功しました。これで初回の全件転送は完了です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/18dce69c-8dfa-948f-b7fb-d50fc133f321.png)

次に、次回以降に行うCDC転送の設定をします。
転送設定の詳細画面に戻りましょう。

### 3-7. スケジュール設定
「スケジュール・トリガー設定」タブを開き、スケジュールを追加します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/cc8c18fd-0752-ccd9-d304-115926a0f222.png)


以下のように実行スケジュールを設定することで、転送を自動化することが出来ます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/755133a3-7f2e-83e1-eb7e-b61e43d4a0b3.png)


### 3-8. 通知設定
必須の設定ではないですが、ジョブの実行ステータスに応じてEmailやSlackに通知することが出来ます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/0f470c5c-5ce4-a663-86ff-da0f4a158ae8.png)


ここまでで全設定が完了しました。
これ以降は、設定したスケジュールに合わせてCDC転送が実行され、差分のデータがBigQueryへ統合されるようになります。

### 4-1. 転送結果の確認
最後に、PostgreSQLのデータがBigQueryに統合できているか確認してみます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/8c0435fc-e221-a9e7-26e1-13e3a0782dcb.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/3c5a2a30-7316-aa9a-bf2d-7166da3ff12a.png)


両方とも同じデータが入っていることが確認できますね。

ここで、元のテーブルにいくつか変更を加えて、BigQuery側でどのように表示されるのかを確認してみましょう。

```SQL
DELETE FROM qiita_test WHERE id = 1;
INSERT INTO qiita_test VALUES (4, 'Kimura');
ALTER TABLE qiita_test ADD NewColumn VARCHAR(20) NULL;
```

id=1 の行を削除
id=4 の行を追加
新たに「NewColumn」のカラムを追加
この３点が変更されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/0ba8d262-418f-a41f-9947-329c52cffab3.png)


このようにMySQLのテーブルに変更を加えたうえでCDC転送をすると…

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2216922/b8a50a9b-bf7b-fe11-32d8-2dc76ed2fb9e.png)


先ほどのクエリで変更した部分がBigQueryでも反映されていることが分かります。

今回は「スキーマ追従」の設定を有効にしているのでカラムの追加が反映されていますが、スキーマ追従が無効の場合には反映されないのでご注意ください。

### 4-2. おわりに
転送が不要になりましたら、「2-2. Slotの作成」で作成したスロットを削除してください。
スロットを削除し忘れた場合、WALがストレージを圧迫してDBがクラッシュする可能性があります。

## まとめ
いかがでしたでしょうか。[trocco](https://trocco.io/lp/index.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)を使うと、簡単にPostgreSQLのデータを取得し、DWH(BigQuery)に貯めることが出来ました。
ぜひデータ分析の際にご活用ください。

troccoを試してみたい場合は、[フリープラン](https://trocco.io/lp/inquiry_free.html?utm_medium=social&utm_source=qiita&utm_campaign=postgresqlcdc_to_bigquery)がありますので、この機会にぜひ一度お試しください。

その他にも様々な分析データをETL・転送した事例をまとめています。ご活用ください。
[troccoの使い方まとめ（CRM・広告・データベース他）](https://qiita.com/hiro_koba_jp/items/2b2caa040804e402bda7)