Zerobus Ingest と Debezium を使った Unity Catalog へのプッシュ型 CDC

オペレーショナルデータベース内のデータがビジネス上の意思決定を動かしており、その分析の多くはレイクハウス内で行われています。しかし、データベースがプライベートな企業ネットワーク、顧客管理環境、または外部サービスからのインバウンド接続をブロックするネットワークポリシーを持つ制限されたサブネット内にある場合、データをそこに取得することは困難です。

多くのチームが Databricks で PostgreSQL の変更データを必要としていますが、データベースを管理するチームは、そのキャプチャの管理も行いたいと考えています。そのチームは、先行書き込みログ（WAL）を読み取り、どの接続がネットワーク外に出るかを決定するプロセスを実行します。その一方で、変更データは、パイプラインやアナリストが構築できる管理された Delta テーブルに格納されます。

Debezium は、チームが完全なデプロイメントを管理してデータベースから変更データキャプチャ（CDC）フィードを抽出できるようにする、一般的なオープンソースのツールです。データベースの隣に配置され、先行書き込みログを読み取って外部へイベントをプッシュします。Zerobus Ingest は、Unity Catalog Delta テーブルに直接レコードを書き込むプッシュインターフェイスであり、その送信先となります。オープンソースの Debezium Server Zerobus シンクは、キャプチャとレイクハウスの間に従来の Kafka クラスターを挟むことなく、この2つを接続します。

NVIDIA と Databricks のエンジニアが協力してこのシンクを構築し、オープンソースの Debezium プロジェクトに提供しました。Databricks は、型付きの Zerobus Ingest シンクを構築しました。その後、NVIDIA は、ソースのオフセットが進む前に、完全な Debezium 変更イベントを、べき等性キーや耐久性のある確認応答などの配信セーフガードとともに格納するエンベロープモードを提供しました。Zerobus Ingest シンクは、Debezium Server 3.7 リリースに同梱されています。

この投稿では、Debezium と Zerobus のそれぞれの役割、NVIDIA がシンクに提供した生の変更エンベロープ、および格納されたイベントがどのように現在状態テーブルまたは履歴テーブルになるかについて説明します。ローカルの PostgreSQL テーブルから現在状態テーブルへのライブ実行を含む実践的なチュートリアルについては、Databricks Community の関連ハウツーブログをご覧ください。 

課題

オペレーショナルデータベースの変更をレイクハウスに信頼性高くレプリケートするには、いくつかの問題を同時に解決する必要があります。

• 挿入、更新、削除を信頼性高くキャプチャする。 作成、変更、および後で削除された行は、各イベントに対して忠実なストリームを生成する必要があります。これには、クエリベースの「テーブルのコピー」アプローチでは通常見落とされる削除も含める必要があります。
• 順序付けに必要な証拠を保持する。 現在の状態を再構築するダウンストリームロジックは、ソーステーブルを正しくレプリケートするために、ソースの順序で変更を適用する必要があります。
• ソースデータベースの過負荷を回避する。 定期的なポーリング (SELECT ... WHERE updated_at > ?) はクエリの負荷を追加し、スケーラビリティに欠けます。
• プライベート / 顧客管理ネットワークに対応する。 データベースが厳重なセキュリティ境界の背後にある場合、これらのシステムへの接続やデータの取り込みが困難になることがあります。
• 純粋なトランスポートとしての Kafka の立ち上げを回避する。 Kafka は強力なイベントバスですが、データベースからレイクハウスに CDC イベントを移動させるためだけにクラスターを運用することは、他にその必要がないチームにとって大きなオーバーヘッドとなります。
• 簡単に処理できるようにデータを格納する。 変更ストリームは、別のエクスポートステップを必要とするサイロではなく、レイクハウスが処理およびクエリを実行できる形式で到着する必要があります。

これらの要件を考慮すると、データがソースからシンクへどのように流れるかというアーキテクチャから始める必要があります。

全体的なターゲットアーキテクチャ

このブログでは、従来のメッセージバスアーキテクチャと Zerobus Ingest を使用したアーキテクチャを比較します。

メッセージバスを使用するアーキテクチャ:

多くのエンタープライズ環境では、セキュリティ境界の内側からのアウトバウンド接続は、インバウンドアクセスよりも許可することがはるかに容易です。変更フィード抽出エージェントのデプロイメントを実行する場合、ローカルプッシュモデルを使用すると、データを外部の宛先へスムーズに流すことができます。従来のセットアップでは、これは Kafka などのメッセージバスになります。Kafka を実行するには、物理スケーリングのためにピークロードを処理するのに必要なブローカーの数を予測するためのキャパシティプランニングのリソースや、プロデュースのレートにコンシューマーが追従するために必要なパーティションの数を見積もるリソースが必要です。これには、Kafka のデプロイメントと管理に関する経験を持つ専任チームが必要です。

データが Kafka に格納されたら、次の作業として、それらのイベントをレイクハウスにレプリケートする必要があります。これは、Spark、Kafka Connector、またはその他のツールを使用したインジェストジョブのデプロイメントによって実現できます。その結果、Debezium からレイクハウスまでの2ホップのアーキテクチャになります。

Zerobus Ingest を使用したアーキテクチャ:

唯一の宛先がレイクハウスである場合、Zerobus を使用することで、メッセージバスなどの中間ホップをバイパスできます。これにより、エンドツーエンドのアーキテクチャが合理化され、データ転送のために Kafka インフラストラクチャをデプロイおよび管理する必要がなくなります。Debezium のシンクとして Zerobus を使用すると、チューニングを行わなくても、ピーク時間帯に稼働し続けます。

Debezium とは

Debezium は、オープンソースのログベースの変更データキャプチャプラットフォームです。テーブルをポーリングする代わりに、PostgreSQL の先行書き込みログ (WAL)、MySQL の binlog、MongoDB の変更ストリームなど、データベースのトランザクションログを読み取ります。その後、各変更を構造化イベントとして出力します。このプロジェクトは 2015 年に開始され、現在では幅広いコネクターエコシステム（PostgreSQL、MySQL、SQL Server、Oracle、MongoDB など）をサポートしています。

各変更イベントには、一貫したエンベロープが含まれています。

• before — 変更前の行の状態（挿入の場合は null）
• after — 変更後の行の状態（削除の場合は null）\
• op — 操作: c (作成/挿入)、u (更新)、d (削除)、r (読み取り/スナップショット)
• source — メタデータ: データベース、スキーマ、テーブル、WAL ログシーケンス番号 (LSN)、トランザクション ID、スナップショットフラグ
• ts_ms — イベントのタイムスタンプ

Debezium は、ポーリングではなくログテイルイングを使用して低レイテンシで変更をキャプチャするため、プライマリデータベースへの負荷を最小限に抑え、コミット順序を保持します。

スタンドアロン (Debezium Server) と Kafka Connect のデプロイメントの比較

Debezium には 2 つの運用モデルがあります。1 つは Kafka Connect上で実行されるもので、もう 1 つは通常のコンピュート上で立ち上げることができる「スタンドアロンサーバー」です。

スタンドアロンランタイムである Debezium Server は、トランザクションログをテイル処理し、トランスポートとしての Kafka クラスターを介さずに、変更イベントを Databricks の Zerobus Ingest に直接送信する、ソース側のエージェントとして機能します。

ソリューション: Debezium + Zerobus Ingest

ログテイラーとして Debezium を選定しました。次は、イベントを Databricks に取り込む必要があります。

Zerobus Ingest は、メッセージブローカーを実行することなく、データを Unity Catalog 管理下の Delta テーブルに直接書き込む、サーバーレスでプッシュ型の取り込みサービスです。プロューサーは Zerobus エンドポイントに接続し、オープンフォーマットの Delta テーブルに格納されるレコードをストリーミングします。高スループットで同時実行性の高いワークロードを処理できるように構築されています。メッセージブローカーのステップをスキップすることで、レイクハウスへデータを移行するために取り込みパイプラインを拡張および実行する手間や、アーキテクチャの保守をシンプルにすることができます。

Debezium と組み合わせた場合、CDC パイプラインは次のようになります。

• Debezium がデータベースのログを監視し、各変更を CDC イベントに変換します。
• Zerobus Ingest がそれらのイベントを Databricks に直接転送し、中間トランスポート層としての Kafka クラスターなしで Delta テーブルに書き込みます。
• Databricks が Delta テーブルで変更ストリームを受け取り、ダウンストリームのキュレーション済みテーブルへと処理できます。

Zerobus Ingest は追記型（append-only）です。変更イベントが格納されたら、次のステップは、マージを実行して各プライマリキーの組み合わせの最新バージョンを取得し、Type-1 ディメンションテーブルを構築することです。

NVIDIA による貢献: エンベロープモード

Debezium の Zerobus シンクは、2 つのテーブルコントラクトを提供します。デフォルトの Typed モードは、各変更イベントをソース自体のカラムを持つテーブルにマップします。NVIDIA が貢献した Envelope モードでは、完全な Debezium 変更イベントが固定の 9 カラムのテーブルに格納されます。

1. target_table
2. destination
3. partition
4. operation
5. idempotency_key
6. key
7. value
8. source_position
9. headers

レコードは JSON (VARIANT カラムに保存) または Protobuf としてエンコードできます。

エンベロープモードは、ダウンストリーム処理に必要なエビデンスを保持します。source_position は、PostgreSQL LSN を含め、ソースパーティションとオフセットを持ち運びます。idempotency_key は、宛先、パーティション、キー、操作、およびソース位置から派生した各イベントの SHA-256 アイデンティティであるため、再配送されたイベントでも同じキーを取得します。キーはイベントを識別するものであり、それ単体でイベントの順序付けやテーブルの重複排除を行うわけではありません。 

後のグループが失敗した場合、以前の永続化されたグループが再配信される可能性があるため、配信は少なくとも 1 回（at-least-once）行われ、重複は idempotency_key を使用してダウンストリームで処理されます。また、エンベロープモードでは、宛先と操作によるフィルタリングを行い、取り込み前にサイズオーバーしたレコードを拒否するため、そのオフセットがコミットされることはありません。

Databricks でのデータモデリング

CDC イベントが Delta に格納されると、一般的なメダリオンアーキテクチャスタイルのフローによって、生の変更フィードがアナリストやアプリケーションが使用できるものに変換されます。

• Bronze：受信したままの未加工の変更イベント。エンベロープモードでは、イベントごとに1行となり、操作、変更前後の行イメージ、ソース位置（LSN）、および idempotency_key が含まれます。
• Silver：パース済みの変更レコード。idempotency_key で重複排除され、ソース位置で順序付けられています。挿入、更新、削除が解釈されて現在の状態が再構築された、ソースデータの使用可能な表現です。
• Gold：財務やパフォーマンス指標などのレポーティングに向けて、業務データとビジネスデータが変換されています。

変更フィードを形作る2つの方法について簡単に触れておきます。

• タイプ1（現在の状態）：変更ログを圧縮し、各キーが最新の値を反映するようにします。ソーステーブルの現在の状態のレプリカです。
• タイプ2（履歴）：時間の経過に沿って各バージョンを保持するため、任意の時点におけるデータの状態をクエリできます。

Databricks Communityの参考ブログ記事では、ローカルのPostgreSQLテーブルを使用して、これを最初から最後まで実際に手順を追って説明しています。Debezium ServerをZerobusシンクで設定し、エンベロープテーブルでの挿入、更新、削除を確認した後、ソースLSNで変更をシーケンス処理しながらAUTO CDCでタイプ1（現在の状態）のテーブルを構築します。

まとめ

適切なCDCアーキテクチャの選択は、最終的には組織のネットワーク制約、運用上の好み、および既存のインフラストラクチャによって決まります。

Debezium ServerとZerobus Ingestを組み合わせることで、中間メッセージトランスポートとしてのKafkaやクラウドオブジェクトストレージ上のステージングエリアに依存することなく、変更をDeltaテーブルに直接ストリーミングするローカルプッシュアーキテクチャをデプロイでき、エンドツーエンドの取り込みパスがシンプルになります。

これにより、PostgreSQLのキャプチャはデータベースを管理するチームの制御下に置いたまま、データをUnity Catalogに取り込むことができます。

はじめに

このパイプラインをご自身で構築するには、Databricksコミュニティのステップバイステップガイドに従ってください。PostgreSQLの変更をDebezium ServerとZerobus Ingestを経由してUnity Catalogテーブルに流し込み、AUTO CDCを使用して最新状態のテーブルを構築します。Zerobus Ingestのセットアップやその他の取り込みインターフェースについては、Zerobus Ingestのドキュメントをご覧ください。

よくある質問（FAQ）

Debezium Zerobus Ingestシンクとは何ですか？

これは、スタンドアロンのDebeziumランタイムであるDebezium Server内のシンクであり、変更イベントをZerobus Ingestに直接送信します。Zerobus Ingestは、それらをUnity CatalogによってガバナンスされたDeltaテーブルに書き込むため、キャプチャとレイクハウスの間にKafkaクラスターを配置する必要はありません。NVIDIAとDatabricksがオープンソースへの貢献として共同開発しました。

エンベロープモードとは何ですか？

エンベロープモードでは、Debeziumの各変更イベントが、ソース自体のカラムにマッピングされるのではなく、固定された9カラムのテーブルに格納されます。NVIDIAがこのシンクに貢献しました。操作、キー、値、ソース位置、およびダウンストリーム処理のためのべき等性キーが保持されます。デフォルトは型付きモードです。

シンクは各変更を正確に1回配信しますか？

いいえ、配信は少なくとも1回（at-least-once）です。後のレコードグループの処理が失敗した場合、以前の永続的なグループが再配信されることがあります。各イベントにはidempotency_keyが含まれているため、ダウンストリーム側で重複排除を行います。

変更フィードから最新状態のテーブルを作成するにはどうすればよいですか？

AUTO CDCを使用して、各キーの最新バージョンを保持するType-1テーブルを構築します。ソースLSNで変更の順序を整理し、ソースの順序通りに適用されるようにします。代わりにすべてのバージョンを履歴として保持したい場合は、Type-2テーブルを構築します。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事