- Postgres の変更を別システムへリアルタイムに渡すには CDC(Change Data Capture) が必要で、単純な通知から WAL ベースのレプリケーションまで、選択肢ごとに信頼性と運用負荷が大きく異なる
- Listen/Notify は最も手軽に始められるが、at-most-once 配信、一時的な通知、8000 バイトのペイロード制限があるため、中核的な CDC というより補助的なシグナルに近い
- テーブルポーリングと 監査テーブル(outbox pattern) は標準テーブルとトリガーだけで実装できるが、削除検知・diff・コミット順序・書き込み増幅・バックプレッシャーを自前で解決する必要がある
- 論理レプリケーション(logical replication) は WAL から insert/update/delete をストリーミングする強力な方式だが、replication slot、ack、再起動、スループット対応までアプリケーション側で管理しなければならない
- Sequin は Postgres の論理レプリケーションをベースに、SQS、Kafka、Elasticsearch、Redis、HTTP エンドポイントなどへ変更を渡し、replication slot を直接扱う負担を減らす
Postgres CDC が必要な場面
- Postgres は保存されたデータを扱うのに強いが、テーブル変更をきっかけにワークフローを起動したり、別のデータストア・システム・サービスへリアルタイムにストリーミングしたりするには、データ移動 を別途設計する必要がある
- Change Data Capture(CDC) は、データベースの変更を識別・キャプチャし、その後段のシステムへリアルタイムに渡す方式である
- Postgres で変更を捉える方法はいくつもあり、実装難易度・信頼性・運用負荷がそれぞれ異なる
Listen/Notify: 最もシンプルな pub-sub
- Postgres の Listen/Notify はプロセス間通信機能で、publish-subscribe パターンとして動作する
- セッションは特定のチャネルを
listen し、データベースのアクティビティや別セッションはそのチャネルへ notify を送れる
- 変更キャプチャにはトリガーを付けて利用できる
- 例のトリガーでは、
after insert or update or delete のタイミングで変更されたレコードの table、id、action を JSON にし、pg_notify('table_changes', payload::text) を呼び出す
- 限界も明確である
- at-most-once 配信セマンティクスを持ち、listener は通知発行時点で接続されていなければならない
- listener は購読開始後の通知しか受け取れないため、ネットワーク問題で一時的に切断されただけでも通知を取りこぼす可能性がある
- ペイロードサイズの上限は 8000 バイト で、これを超えると
notify コマンドは失敗する
- ペイロードサイズにはチャネル名も含まれ、Postgres の識別子と同様にチャネル名は最大 64 バイトになりうる
- 基本的な変更検知やテーブルポーリングの最適化には使えるが、複雑な CDC 要件にはあまり向かないことがある
テーブルポーリング: シンプルだが削除と diff に弱い
- 最もシンプルで 堅牢な 変更キャプチャ方式は、テーブルを直接ポーリングすることである
- 各テーブルには、行が更新されるたびに更新される
updated_at のようなカラムが必要で、必要ならトリガーで作れる
updated_at と id の組み合わせをカーソルとして使い、アプリケーションロジックがカーソルを保存・管理する- Notify の購読を併用すれば、レコードの挿入・更新をアプリケーションに知らせてポーリング頻度を下げられる
- Postgres の通知は一時的なので、ポーリングの上に載せる最適化としてだけ使うのが適切である
- 主な欠点は 3 つある
- 削除された行はテーブルに残らないため、削除検知 ができない
- 対策として、delete トリガーが
id と必要なカラムを deleted_contacts のような別テーブルに保存し、アプリケーションがそのテーブルをポーリングできる
- レコードが更新されたことは分かっても、何が変わったか は分からない
- Postgres の datetime と sequence は コミット順序がずれることがある ため、
updated_at 基準のブロックを読んでいる間に、まだコミット中の行を取りこぼす可能性がある
- 削除、diff、まれな取りこぼしが大きな問題でない単純な変更追跡には、妥当な選択肢である
監査テーブル: outbox pattern で変更ログを保存
- 監査テーブル(audit table) 方式は、別の
changelog テーブルに変更を記録するもので、outbox pattern とも呼ばれる
changelog には変更に関するカラムを持たせられる
action: insert、update、delete のどれかold: 変更前レコードの jsonb。insert では空values: 変更されたフィールドの jsonb。delete では空inserted_at: 変更が発生した時刻
- 実装するには、変更が起きるたびに
changelog へ insert するトリガー関数と、監視対象テーブルごとのトリガーが必要になる
changelog をキューのように扱う消費方式も可能である
- アプリケーション worker がテーブルから変更を取得する
- おおむね exactly-once 処理のために、Postgres の
for update skip locked を使える
- worker はトランザクションを開き、
order by timestamp limit 100 for update skip locked でバッチをロックして処理し、処理済みレコードを削除してからコミットできる
- 運用上の欠点もある
- 単一テーブルへの書き込みが監査テーブルへの複数回の書き込みを生む 書き込み増幅(write amplification) が発生する
- 一般には、監査テーブルへの初回 insert、処理中の update、処理後の delete まで最低 3 回の書き込みが生じる
- worker で fan-out する方式は、アプリケーションに合わせて自分で設計しなければならない
- 本番規模のデプロイ前に、トリガー関数やテーブル設計の調整が必要になる可能性が高い
- worker が変更をチェックアウトしたまま保持できる時間制限のような詳細ポリシーも検討できる
- worker が正常に処理できなくても監査テーブルにはデータが溜まり続けるため、バックプレッシャー 管理が不足しがちである
Foreign Data Wrapper: 特定の Postgres 間同期に近い選択肢
- Foreign Data Wrapper(FDW) は、Postgres データベースから外部データソースを読み書きできるようにする機能である
- 最も広くサポートされる FDW ベース拡張は
postgres_fdw である
- 2 つの Postgres データベースを接続し、片方のデータベースからもう片方のテーブルを参照する view に近い構造を作れる
- 内部的には、一方の Postgres データベースがクライアント、もう一方がサーバーになる
- foreign table にクエリを投げると、クライアント側データベースが Postgres の wire protocol を使ってサーバー側データベースへクエリを送る
- FDW は変更キャプチャ方式として一般的ではなく、ごく特定の状況以外では勧めにくい
- ある Postgres データベースの変更を別の Postgres データベースに書き込みたい場合には、FDW が適していることがある
- 例としては、会計用データベースとアプリケーション用データベースを分けて使うケースである
- 中間の変更キャプチャ段階を省き、
postgres_fdw でデータベース間へ直接反映できる
- 独自の FDW を作り、内部 API へ変更を POST する方式も可能である
- コミット内で API に書き込むため、API が変更を拒否してコミットをロールバックさせることもできる
- FDW は強力だが、CDC 用途で最適になることはまれであり、独自 FDW を書くのは変更キャプチャ手法の中でも最も大きな作業に近い
直接の論理レプリケーション: WAL ベースの強力な CDC
- Postgres にはデータベースレプリケーション用のプロトコルがあり、その 1 つが 論理レプリケーション(logical replication) である
- 論理レプリケーションは Postgres の WAL(write-ahead log) の上に構築されている
- データベース内のすべての insert、update、delete が追跡される
- 変更は subscriber へストリーミングされる
- 利用者はまず primary に replication slot を作成する
pg_create_logical_replication_slot('<your_slot_name>', '<output_plugin>') の形式を使う
output_plugin は WAL の変更をデコードするプラグインを指定する
pgoutput はデフォルトのプラグインで、クライアントサーバーが期待するバイナリ形式で出力するtest_decoding は WAL の変更を人が読める形で提供するシンプルな出力プラグインである- Postgres 組み込みではないが人気のあるプラグインとして
wal2json があり、JSON は Postgres のバイナリ形式よりアプリケーションの出発点として扱いやすい
- replication slot を作成した後、開始して消費できる
- replication slot は標準クエリとは異なる Postgres プロトコル領域を使う
- 多くのクライアントライブラリが replication slot 操作を助ける関数を提供している
psycopg2 の例では、cursor.start_replication(...) と cursor.consume_stream(...) で WAL メッセージを消費し、cursor.send_feedback(flush_lsn=msg.wal_end) で ack を送る
- クライアントは受け取った WAL メッセージを ack する必要があり、replication slot はオフセットを持つ Kafka に似た動きをする
- 論理レプリケーションは CDC のために作られた堅牢な方式だが、複雑でもある
- replication slot と replication protocol は通常のテーブルやクエリより開発者になじみが薄い
- 再起動時にメッセージを取りこぼさない戦略が必要である
- Postgres から出てくる大量メッセージを処理できるよう設計しなければならない
Sequin: 論理レプリケーションを包む CDC ツール
- Sequin は、Postgres の変更や行をキュー、ストリーム、検索インデックス、キャッシュ、HTTP エンドポイントなどへ渡す CDC ツールである
- 宛先には SQS、Kafka、Elasticsearch、Redis、HTTP endpoints などが含まれる
- Sequin は内部的に Postgres の 論理レプリケーション を使うが、low-level protocol の複雑さを抽象化する
- insert、update、delete をすべてキャプチャでき、update と delete では行の
new 値と old 値の両方をキャプチャする
- Sequin を検討する条件は次のとおりである
- リアルタイム CDC が必要である
- SQS や webhook のような宛先へ中間システムなしで直接ストリーミングしたい
- 過去データの backfill や SQL の
where 句ベースの変更フィルタリングといった機能が必要である
- replication slot を直接管理する方法よりシンプルな代替が必要である
- exactly-once 処理保証が必要である
- 欠点もある
- Sequin は Postgres の内部拡張ではなく、データベースの横で動く サードパーティツール である
- 拡張ではないため、どの Postgres データベースとも広く互換性を持つが、Sequin Cloud を使わないなら追加インフラを自前で用意する必要がある
選び方の基準
- 立ち上げ段階では Listen/Notify とテーブルポーリングが適している
- Listen/Notify は重要度の低いイベントキャプチャ、プロトタイピング、ポーリング最適化に向いている
- ポーリングはシンプルなユースケースに無難で一直線な解決策である
- もう少し本格的な段階では、監査テーブルが中間的な選択肢になりうる
- 行の
new と old のペイロードをキャプチャできる
- うまく作れば exactly-once 処理システムを得られる
- スケール時には書き込み増幅とバックプレッシャー不足が問題になり、手動構成でミスをするとメッセージを落とす可能性がある
- スケール段階では、論理レプリケーションが堅牢な解決策に最も近い
- ただし slot から直接読むより、Sequin のようなツールを使う方が推奨される
- FDW は興味深い機能だが、一般的な CDC 要件を解決できる可能性は低い
1件のコメント
Hacker News の意見
トリガー + 履歴テーブル（監査テーブル）が98%のケースで正解です。まだ使っていないなら今日から使えばよいです。30年以上にわたって検証されてきた手法です
汎用的に実装する簡単な例は https://gist.github.com/slotrans/353952c4f383596e6fe8777db5d... にあります。空間効率は諦めて「実装のしやすさ」を選んだ方式です
不変データを保存できるなら本当に良いのですが、データベースにはおそらく可変データが非常に多く、日々多くのことを忘れている可能性が高いです。忘れずに履歴テーブルを使えばよいです
参考: https://github.com/matthiasn/talk-transcripts/blob/master/Hi...
Papertrail のようなアプリケーション層の履歴追跡ライブラリや手法は使わない方がよいです。遅く、エラーが起きやすく、アプリスタックを迂回する DB 変更を捕捉できません。
updatedタイムスタンプをアプリで付けようとする試みも根本的に間違っています。各 Web サーバーごとに時計が異なるからです。DB の時計を使うべきで、それだけが正しい時計ですnow()のような呼び出しを入れてDB の時計を使うのが正しいですただし、このタイムスタンプだけで同期するには不十分です。タイムスタンプはトランザクションのコミット時点ではなく、トランザクション開始時点に生成されるためです
テーブルをポーリングし、最近のタイムスタンプでフィルタすると、コミット順が入り乱れたトランザクションの一部を取りこぼす可能性があります。数分前までさかのぼって取得し、重複を除去するバッファ区間を設けることはできますが、PostgreSQL ではトランザクション時間に上限がなく、あまり過去まで取得すると無駄が大きくなります。正確性と効率が重要なら、この方式は適していません
ログシーケンス番号、DB 時刻、
REPLICA IDENTITY FULLを使えば、変更前後の状態まで含まれます。その後 Snowflake などにコレクションをマテリアライズすれば、ソース DB の更新に追従する同期テーブルを基本的に得られます同じ基盤データレイクから、監査目的の全テーブル履歴を変換またはマテリアライズすることもできるため、ソース DB に再度キャプチャや WAL リーダーを接続する必要はありません
似たパターンを JSON ではなくカラムベースで実装した SQLite の方式についても別途説明しています: https://simonwillison.net/2023/Apr/15/sqlite-history/
この記事は、Postgres の基本機能で可能な複数のアプローチを簡潔によくまとめています
「監査テーブルに変更をキャプチャする」の部分では、以前の会社で Temporal Tables パターンをうまく使っていました。他の主要なリレーショナル DBMS と違い、Postgres 自体には組み込まれていませんが、SQL 関数として利用できる単純なパターンがあります: https://github.com/nearform/temporal_tables
特定時点のテーブル状態を確認できるため、「8月12日時点でこのユーザーの設定は何だったか」「昨夜11:55に未処理レコードはいくつあったか」「現在と1週間前の機能フラグの差分を見せてほしい」といった質問に答えられます
以前、非常に大きな モノリシックな SQL Server を持つ会社でコンサルティングしたことがある。Postgres ではなかったが、Postgres だったと仮定しても似たようなものだ。
何十年も運用され、社内のあらゆる用途に使われており、事実上、会社全体のすべてのアプリケーションと業務プロセスがこのデータベースにデータを保存していた。
問題は、この DB を参照するアプリケーションが多く、データを挿入・更新するプロセスや手順も非常に多かったため、上流の挿入・更新プロセスが変更されたり新たに追加されたりすることで、アプリケーションレベルの不変条件を壊すことがあった点だ。正常なプロセスでも、悪いデータがあると違う動きをした。
原因を追跡するのは非常に難しかった。調べる対象の多くが10年前に書かれたもので、その社員たちはすでに会社を去っていたからだ。
Postgres データベースの変更を何らかの DAG 形式でキャプチャして、どのプロセスがデータを挿入・更新・削除しているのか、歴史的にどう振る舞ってきたのか、複数のアプリケーションがこのデータをどう参照しているのか、クエリ統計が時間とともにどう変化しているのかを把握できないかと思っている。
こうした先行事例があるのか、どのようなアプローチならこの種のツールを作れるのかはよく分からない。以前、似たものを作ることを考えたことがあるが、良い選択をするには Postgres コアエンジニア級の理解が必要な領域に思える。
各変更についてクライアントレベルの出所データまで得られるわけではない。
それでも回避策はある。論理レプリケーションストリームには
pg_logical_emit_message関数の情報メッセージも含められるので、クライアントが直接メタデータを入れられる。各トランザクション開始時にクライアント識別子を出力するよう設定できるかもしれない。last updated by）を追跡するカラムを持たせている。アンチパターンかもしれないので、もっと堅牢な解法があれば望ましい。このアプローチは、WAL やトリガーを使うほぼすべての SQL 系ソリューションで動作する。
SQL Server でトリガー方式は何度も使ったことがあるが、すべてのクエリをログに取ると遅くなる傾向がある。運用を妨げない挿入メカニズムを設計するのは完璧ではなく、サンプリングが必要になるかもしれない。
「監査テーブル」の道を行くなら、単に pgaudit を使えばよい。実運用で検証された拡張で、AWS を使っているなら RDS でも利用可能だ。
https://github.com/pgaudit/pgaudit/blob/master/README.md
https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appen...
https://github.com/arkhipov/temporal_tables
https://news.ycombinator.com/item?id=26748096
わざわざやらなくてもよい。これを望むということは、Postgres のリレーションを 契約 に変えるということだ。どのサービスも内部状態を永続化できなくなる。
ドメイン駆動設計に本気で徹底するなら可能かもしれないが、軽量で実用的な イベント駆動システム を使うほうがよい。
イベント駆動の何かは1000倍複雑だ。
updated_atカラムをポーリングする方式は、最も単純な形では堅牢ではない。トランザクションがその順序でコミットされる保証がないためだ。updated_atが2023-09-22 12:00:01に設定される。少し後にトランザクション B が開始され、Row 2 の
updated_atが2023-09-22 12:00:02に設定され、B が先にコミットされる。ポーリングクエリが実行され、Row 2 を最新の変更と見なしてカーソルを
2023-09-22 12:00:02に更新した後、A が後からコミットされると Row 1 を取り逃がす。この問題を避ける単純な方法は、ほぼリアルタイムでポーリングしないことだ。順序は最終的には一貫してそろう。
より堅牢な提案としては シーケンス を使うことかもしれない。たとえば、行が変更されるたびに増加する
updated_at_idxカラムを持たせる方式だ。now()を入れる before トリガーを使っても、2つの行のupdated_atタイムスタンプがトランザクションのコミット順と異なる可能性があるのか気になる。updated_atとコミットタイムスタンプが同じである必要はないが、updated_atはミリ秒・マイクロ秒単位でコミット順を正確に表してほしい。updated_atではなく、トリガーが現在の トランザクション ID に設定する_txidカラムを使っている。その後ポーリングするときに、txid_current()でどのトランザクションがコミット済みで、どれがまだかを確認する。少し危なっかしく、境界値エラーを起こしやすいが、数年にわたって本番で問題なく動いている。
記事が素晴らしい。
Elixir と Postgres を使っているなら、似たアプローチで WAL の変更をリッスンする小さなライブラリを作ってある: https://github.com/cpursley/walex
これらの方式はどれもいまひとつで、個人的にはポーリングが最も実用的だと思う
Postgres がこの領域で革新してくれるとよい
SQL 標準に入るまでは、リレーショナル DBMS のカーネル空間で推進力が生まれにくいと思う。選択肢は多く複雑で、ユーザー空間で成功している解法も、性能面で過度な負担になっているわけではない
ちなみにこの分野を研究している人たちは、おおむね監査テーブル方式に傾いている。データベース内で一貫した ACID 特性を維持でき、プロキシやポーリングジョブを追加する代わりに Postgres を単一障害点のままにできるため
データの世界には大きな空白がある。データストアに結果を問い合わせるのではなく、クエリ結果がインクリメンタルにプッシュされるとよい
リアルタイム／ストリーミング分析を多く扱っているが、ストリーム処理もできるし、データストア内のマテリアライズドビューで一部を処理することもできる。だがデータが DB やデータレイクに入った後、下流で変更を見ようとすると、実質的にはまたポーリングに戻ることになる
データ内で何らかの状況が発生したときに反応したり、ページをリロードせずに画面を更新したりするには、きれいな解法があまりない。この記事の解法も、第一級機能というより回避策に近く見える
ページをリロードせずにリアルタイム更新されるレポートを作りたいなら、たいてい DB からデータをロードしたうえで、Kafka と WebSocket で GUI に変更を流す方式になる。すると、一部の分析はコードで、一部は DB で処理するという奇妙なラムダアーキテクチャを運用することになる
この領域にも革新はある。KSQL と Kafka Streams は変更を出力でき、Materialize にはサブスクリプションがあり、ClickHouse にはライブビューがある。ただし多くの機能は新しいかプレビュー段階で、ぴったり合うわけではない。どれも使ってみたが、開発者にあまりに多くの作業を押しつけていると感じる
[select * from orders with suscribe]のようなオプションで、すぐに変更フィードを受け取れるライブラリがあるとよい。十分に重要な領域なのに、これまであまり注目されてこなかった記事では扱われていないレプリケーションの大きな落とし穴があり、だから私はレプリケーションを使わない
Postgres は、レプリケーションスロットのコンシューマーがデータを取りこぼさないよう、非常に強く保証しようとする。そのためコンシューマーがスロットからデータを消費しないと、Postgres は取りこぼされたデータを親切に保持し続け、最終的にはディスクがいっぱいになって DB が倒れるところまで行く。プロトタイピング中に別々の SaaS DB 2 か所で経験し、復旧にはサポートチケットを入れるしかなかった
レプリケーションスロットのコンシューマーが読み取りを止めたら、必ずアラートが鳴るべきだ
もう一つの理由は、テーブルの初期スナップショットを取得するコードパスと、変更を読むコードパスが完全に異なること。変更を一つも取りこぼさないようにレプリケーションスロットの読み取りを初期化するのは、些細なことではない
残念ながら、変更キャプチャの観点ではレプリケーションが最もハック感の少ない解法だ
私はポーリングを使っているが、
updated_atの代わりに txid を保存しているどのような挙動のほうが望ましいのか気になる
大きなデータ量を扱うなら、初期スナップショットと変更読み取りは別々に処理したくなる。並列初期化や物理バックアップベースの初期化といった作業が可能であるべきだからだ。ただし、スロット作成後に既存データを選択的にストリーミングしてくれる機能が有用かもしれない、という点は理解できる
変更を取りこぼさないようにレプリケーションスロットの読み取りを初期化する部分は、難しいはずはなさそうだが、どこで詰まったのか気になる
すべての変更が必要でない場合は、接続が切れると自動で片付く一時レプリケーションスロットも有用だ。サーバーを落とさないよう、保持 WAL の最大値を設定する構成もある
updated_atの代わりに txid をどう使っているのか、もう少し説明してもらえるとありがたい