PostgresへのCDCの実装方法

トランザクションデータベースからのデータを分析データベースで利用可能にすることは、あらゆるモダンなデータアーキテクチャの不可欠な部分です。それはまた、脆弱なツール、高コスト、複雑な運用との永続的な戦いでもあります。SnowflakeでPostgresサービスを構築し始めたとき、この問題を解決することが自然に最優先事項となりました。この記事では、データミラーリングの背後にあるエンジニアリングについて深く掘り下げます。Postgresレプリケーションをゼロからどのように再考したかについて説明します。 Postgresレプリケーションの最適化 Postgresは素晴らしい運用データベースですが、その変更データキャプチャ（CDC）の仕組みは依然として多くの改善の余地があります。多くのパイプラインは、レプリケーションツールが継続的なデータとスキーマの変更、スナップショット、障害の複雑な相互作用の処理に追われるため、脆弱になりがちです。Snowflake Postgresのために、信頼性の高い、すぐに使えるエクスペリエンスを構築するには、Postgresレプリケーションをゼロから再発明する必要がありました。 データミラーリングは、Snowflake Postgresのパブリックプレビュー中の新機能で、低コスト、低遅延、トランザクション整合性で、Snowflakeへの非常に回復力の高いデータレプリケーションを実行します。内部的には、変更をPostgresから直接Apache Iceberg™テーブルにトランザクションバッチでプッシュすることで機能します。バッチは、トランザクション的かつサーバーレスで、Snowflake内のテーブルに自動的に適用されます。「データレイクへのトランザクションプッシュ、Snowflakeでのトランザクション適用、追加インフラなし」というシンプルさは、レプリケーションを多くの複雑な障害条件を伴う混沌としたプロセスから、永遠に実行される単純な時計仕掛けへと変えます。ボタンを押すだけで、PostgresテーブルがSnowflakeに表示されます。 プルからプッシュへ：変更データキャプチャをPostgresへ移行 変更データキャプチャ（CDC）とは、トランザクションデータベースからの変更を、別のシステムで再生可能な形式でキャプチャするプロセスです。Postgresで利用可能な主な機能は「論理デコーディング」と呼ばれ、WALレコードを論理的な行レベルの挿入/更新/削除操作にデコードすることを指します。これらの操作は、ネットワーク上のストリームとして公開されます。そこから先は、クライアントの負担となります。実際には、レプリケーションにはさらに多くのステップが含まれます。バックフィル、スキーマ変更、CREATE/ADD/REMOVE/DROPテーブル操作の処理、新しいテーブルのスナップショット、障害発生時の再起動、変更の効率的なマージ、トランザクション境界の維持、適切なサイジングなどです。Postgresの組み込み論理レプリケーションでさえ、これらの側面のほんの一部しか処理しません。 論理デコーディングアプローチの1つの問題は、変更を消費する外部システムがPostgresの状態について何も知らないことです。例えば、スキーマ変更がいつ発生するか、テーブルスナップショットが変更とどのように一致するか、あるいはPostgresが稼働しているのか、ネットワークがダウンしているのかさえ知りません。この問題の解決策は非常にシンプルです。Postgresからデータレイクへ、そしてこの場合はIcebergテーブル（圧縮されたParquetを使用）へ変更をプッシュすることです。Amazon S3のようなオブジェクトストアは、非常にスケーラブルで信頼性が高く、Postgresのバックアップにも常に使用されています。変更データキャプチャの適切な宛先でもあります。 ミラーリングは、snowflake_cdc という新しいPostgres拡張機能を使用し、バックグラウンドで（「ベースワーカー」を使用して）変更を継続的にテーブルごとの変更ログと「メタログ」にプッシュします。拡張機能を使用する利点は、Postgresで何が起こっているかを正確に把握できることです。スキーマ変更と複雑なデータ操作言語（DML）およびデータ定義言語（DDL）トランザクションを慎重に調整できます。変更をプッシュしながらスナップショットを取得し、スナップショットを変更と一致させることができます。プッシュベースの変更データキャプチャは、インフラストラクチャとその関連の問題のクラス全体を回避し、オブジェクトストアを介してプロデューサーとコンシューマーを効果的に分離します。 レプリケーションタイムラインの解明 データミラーリングのようなレプリケーションシステムを構築する際、重要な側面はデータベースのタイムラインです。レプリケーションプロセスは、最近の過去のデータベースの状態を扱います。Postgresへの各書き込みは、実質的に4つのステージを経ます。これらはそれぞれ、異なる時点で作動する同じタイムライン上の連続したプロセスを表します。 書き込み：書き込みはテーブルを変更し、WAL（「現在」）に追加します。 デコード：過去のWALが、行レベルの変更に変換されます。 キャプチャ：過去の行レベルの変更が、バッチでキャプチャされます。 適用：過去の変更バッチが、宛先テーブルにマージされます。 デコーダープロセスは、書き込み時点のカタログテーブルを読み取るためのPostgresの特別な機能（「履歴スナップショット」）に依存しています。これにより、レコードがデコードされる時点でテーブルが既に変更または削除されていたとしても、バイナリWALレコードを行レベルの変更として理解できます。データミラーリングの場合、レコードは一時ファイルに入ります。デコーダーは定期的に、現在のバッチを完了し、バッチの準備ができたことをキャプチャプロセスにメッセージで通知する信号を受け取ります。キャプチャプロセスは、完了したファイルをIcebergの変更ログに追加し、メタログにレコードを書き込み、レプリケートされたLSNを追跡します。 スキーマ変更も同じ書き込み→デコード→キャプチャパスに従い、新しい変更ログを生成する可能性があります。新しいメタログと変更ログレコードは、Icebergテーブルに表示されます。Snowflakeの適用プロセスは、メタログ内のすべての命令を実行する有限ステートマシンとして機能します。操作が変更バッチ（一般的なケース）の場合、各テーブルについて隣接するすべての変更バッチが一緒に処理されます。このアプローチにより、スキーマ変更が追加の書き込みを伴うトランザクションの一部であった場合でも、変更ストリームに正しくシーケンスされることが保証されます。予期せぬ障害によりWALが失われた場合、Postgresは自動的に新しいスナップショットをプッシュし、Snowflakeにそれらを消費するように指示できますが、フェイルオーバースロットを使用しているため、実際にはこれは非常にまれです。 分散システム構築ブロックとしてのトランザクション データベースシステムは、トランザクションという1つのシンプルなプリミティブを通じて、システムおよびハードウェア障害に関連する膨大な複雑さを隠蔽できます。トランザクションが低レベルのシステム障害により失敗した場合、何も起こらず、再度試行します。トランザクションが成功した場合、同じ作業を再度行うことは決してないことを保証できます。 Extract, Transform, Load（ETL）やCDCのようなプロセスで感じるフラストレーションは、トランザクションが突然失われ、すべての異なる障害モードを自分で処理しなければならないために発生します。この問題に対する私たちの最初の答えは、マネージドバージョンのオープンソースpg_lake拡張機能である「Postgres for your data lake」であり、現在は一般提供されています。これにより、PostgresはPostgresテーブルとIcebergテーブルを横断したトランザクションを実行する独自の能力を得ます。ETLは通常、外部ツールを必要とし、ユーザーは冪等性を設計し、非常に注意深い記録管理を行う必要があります。今では、Postgresテーブルから削除し、Icebergテーブルに挿入してコミットするためにSQLを使用するだけで済みます。その時点で、データはSnowflakeでクエリ可能になります。 このアプローチは多用途ですが、SQLは高頻度の更新のエンドツーエンドレプリケーションには適さないため、データミラーリングが追加するレイヤーはそれです。内部的には、データミラーリングはpg_lakeとSnowflakeのIceberg実装を最大限に活用します。Postgresテーブルからのデータとスキーマの変更のバッチを、1つのPostgres側トランザクションで複数のIcebergの変更ログにプッシュします。Snowflakeは、Snowflake側トランザクションで一度に複数のバッチをマージします。これは、すべてのSnowflakeテーブルが1つのトランザクションで、Postgresのトランザクション境界に正確に、外部キーと結合の整合性を維持して進むことを意味します。トランザクションレプリケーションアプローチは、非常に高いスループットにスケールします。