FivetranでAurora PostgreSQLのLogical Replicationを設定してみた

かわばたです。

今回は、Aurora PostgreSQL + Fivetran の構成で、Fivetran の増分同期方式として Logical Replication を選択してConnectionの設定を試してみました。

Fivetran の Aurora PostgreSQL の 増分同期方式は2種類あります。

• logical replication：PostgreSQL の WAL（Write-Ahead Log） を logical decoding で読み取り、Fivetran が変更データを取得する方式
• Query-Based：Fivetran が定期的に SQL を実行し、PostgreSQL の system columns (xmin, 必要に応じて ctid) を読んで変更を検知する方式

Logical Replication は変更差分ベースのため、大規模テーブルでも効率がよく、Fivetran でも基本的にこちらの方式が推奨されています。

https://fivetran.com/docs/connectors/databases/postgresql/aurora-setup-guide

概要

背景

Aurora PostgreSQL を Fivetran に接続し、増分同期方式として Logical Replication を利用する場合は、Aurora 側で publication と replication slot の作成が必要です。Fivetran の Aurora PostgreSQL セットアップガイドでも、publication を作成し、さらに pg_publication_tables で選択したテーブルが publication に含まれていることを確認するよう案内されています。

検証環境

Aurora PostgreSQL の logical replication は publisher 側で WAL を読み出し、既定では pgoutput を使います。

Aurora PostgreSQLの設定ですが、今回は検証用のためエクスプレス構成で環境を作成しました。

https://dev.classmethod.jp/articles/aurora-postgresql-express-configuration/

事前準備

検証用スキーマ・テーブルの作成

まず、publication に含めるテーブルと含めないテーブルを 1 つずつ用意します。

UPDATE の挙動を見やすくするため、主キー付きテーブルを作成します。PostgreSQL では、publication が UPDATE / DELETE を含む場合、対象テーブルには replica identity が必要で、通常は主キーがその役割を果たします。

https://www.postgresql.org/docs/current/logical-replication-publication.html

CREATE SCHEMA IF NOT EXISTS ft_lr_pub_test;
DROP TABLE IF EXISTS ft_lr_pub_test.included_table;
DROP TABLE IF EXISTS ft_lr_pub_test.excluded_table;
CREATE TABLE ft_lr_pub_test.included_table (
    id          BIGINT PRIMARY KEY,
    note        TEXT,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE ft_lr_pub_test.excluded_table (
    id          BIGINT PRIMARY KEY,
    note        TEXT,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);
INSERT INTO ft_lr_pub_test.included_table (id, note)
VALUES
    (1, 'included: initial row 1'),
    (2, 'included: initial row 2');
INSERT INTO ft_lr_pub_test.excluded_table (id, note)
VALUES
    (1, 'excluded: initial row 1'),
    (2, 'excluded: initial row 2');

Fivetran ユーザーへの権限付与

この検証では、publication の有無だけを差分条件にしたいため、Fivetran ユーザーには included_table と excluded_table の両方に対して SELECT 権限を付与します。

Fivetran は初回同期時に、Fivetran 側で同期対象として選択され、かつ接続ユーザーに必要な参照権限があるテーブルを取り込みます。Logical Replication を使う場合は、そもそも publication に含まれていないテーブルは Fivetran 側で同期対象として扱えませんでした。

GRANT USAGE ON SCHEMA ft_lr_pub_test TO your_fivetran_user;
GRANT SELECT ON ft_lr_pub_test.included_table TO your_fivetran_user;
GRANT SELECT ON ft_lr_pub_test.excluded_table TO your_fivetran_user;

将来このスキーマに追加するテーブルにも自動で SELECT 権限を付けたい場合は、必要に応じて ALTER DEFAULT PRIVILEGES を併用してください。

publicationおよびreplication slotの設定

publication の作成

今回は比較のために included_table のみを publication に含めます。

DROP PUBLICATION IF EXISTS ft_lr_pub_test_pub;
CREATE PUBLICATION ft_lr_pub_test_pub
FOR TABLE ft_lr_pub_test.included_table;

publication の内容確認

pg_publication_tables を使うと、どのテーブルが publication に含まれているか確認できます。このビューは FOR ALL TABLES や FOR TABLES IN SCHEMA の定義も展開して確認できます。Fivetran も Aurora セットアップガイドで、このビューを使って選択したテーブルが publication に入っていることを確認するよう案内しています。

SELECT
    pubname,
    schemaname,
    tablename
FROM pg_publication_tables
WHERE pubname = 'ft_lr_pub_test_pub'
ORDER BY schemaname, tablename;

replication slot の作成

Aurora PostgreSQL の Logical Replication を Fivetran で利用する場合、pgoutputを使った logical replication slot を作成します。

SELECT pg_create_logical_replication_slot('ft_lr_pub_test_slot', 'pgoutput');

レプリケーション権限の付与

GRANT rds_replication TO your_fivetran_user;

replication slot の確認

Aurora 側で slot を確認するには pg_replication_slots を参照します。Fivetran の Setup Test でも、slot の存在や pgoutput plugin、replication 権限などが確認されます。

https://beta.fivetran.com/docs/connectors/databases/postgresql/setup-guide

SELECT
    slot_name,
    plugin,
    slot_type,
    database,
    active
FROM pg_replication_slots
WHERE slot_name = 'ft_lr_pub_test_slot';

Fivetran ユーザーが slot を読めるか確認

Fivetran の Aurora セットアップガイドにある確認 SQL です。これが成功すれば、slot / publication / 権限の最低限の成立確認になります。

https://fivetran.com/docs/connectors/databases/postgresql/setup-guide

SELECT count(*)
FROM pg_logical_slot_peek_binary_changes(
    'ft_lr_pub_test_slot',
    null,
    null,
    'proto_version', '1',
    'publication_names', 'ft_lr_pub_test_pub'
);

Fivetran 側の設定

Destination の設定を行います。
今回は Snowflake で行いましたので、下記ブログが参考になるかと思います。

https://dev.classmethod.jp/articles/fivetran-destination-setting-snowflake/

Connectionの設定についても下記ドキュメントを参考に設定しました。

https://fivetran.com/docs/connectors/databases/postgresql/aurora-setup-guide

Update MethodはLogical Replicationを設定しています。

下記は上述の設定内容を記載しました。

• Replication Slot：ft_lr_pub_test_slot
• Publication Name：ft_lr_pub_test_pub

Save&Testを押下して、接続確認を行います。

その後、Select Data to Sync の画面に移動し、同期対象を選択します。

同期の確認

同期を実行したところ、Destination に設定した Snowflake にデータが連携され、publication に含めたテーブルのデータのみが格納されていました。

まとめ

Aurora PostgreSQL + Fivetran で増分同期方式に Logical Replication を選んだ場合、Aurora 側の publication に含まれていないテーブルは、Fivetran の同期対象として選択できず、結果として同期されませんでした。

Aurora PostgreSQL側で設定項目がいくつかあるので、何かの参考になれば幸いです！