Explore the advanced function of Pravega Flink Connector Table API

Flink Forward Asia 2020 で、私たちのチームは「Pravega Flink Connector の過去、現在、未来」というトピックを共有し、Pravega Flink Connector FLIP-95 Table API をゼロから構築するプロセスを紹介しました。これにより、Pravega ストリームを Flink Table API に対応させ、リンクと相互変換を可能にすることで、ユーザーはシンプルな SQL 文で Pravega ストリームのデータをクエリおよび書き込みできるようになりました。それから 1 年以上が経過し、フローとテーブルの統合を推進する中で、利用シナリオをさらに拡充し、使いやすさを向上させる高度な機能を追加してきました。この記事では、そうした新機能の中から Catalog API 統合と Debezium サポートの 2 つを取り上げ、その技術的な詳細について詳しく解説します。

1. Pravega Schema Registry プロジェクトの概要

Pravega Schema Registry は、2020 年に Pravega エコシステムコミュニティがリリースした新機能であり、Confluent Schema Registry や AWS Glue と同様に、Pravega を使用してデータスキーマ構造を保存・管理する汎用ソリューションです。Pravega Schema Registry の機能を活用することで、コネクタ側で Catalog API を実装し、Pravega カタログの機能を実現できます。ユーザーは Flink から SQL を使用して Pravega ストリームのデータに直接アクセスでき、CREATE TABLE の DDL を書き直して接続を確立する必要がありません。

1. プロジェクトの動機

シリアル化された生のバイナリデータは Pravega ストリームに保存されるため、読み取り側と書き込み側の両者が書き込みデータのスキーマについて合意を持っている必要があります。データと開発の規模が小さいうちはこの問題は容易に解決できますが、ビジネスの成長に伴って負荷の規模が拡大し、数千のストリームの読み書きや複数の開発部門間の連携が必要になると、問題はより困難になります。この課題に対応するため、Pravega Schema Registry コンポーネントがスキーマの整合性を保証します。

上図に示すように、書き込み側がイベントを書き込む際、読み取り側は同じスキーマを使用してそのイベントを逆シリアル化する必要があります。これにより、バイナリから正確なデータを取得できます。機密情報などは、メタデータに基づいて追加の暗号化やエンコードが施される場合もあります。

Pravega のストリームごとに、エンコード形式の保存を含むスキーマメタデータを一元管理する必要があります。書き込み側がこれを定義して書き込むことで、すべての読み取り側が読み取ってデータの構造と逆シリアル化方法を把握でき、スキーマの合意を形成できます。

また、この種のメタデータの保存を外部コンポーネントに依存させたくありません。Pravega 自体を使用するのが理想的であり、運用保守のコストを削減できるだけでなく、Pravega 自体の効率的な永続ストレージ機能も活用できます。

さらに、非常に一般的なビジネスシナリオとしてスキーマ変更があります。同じストリーム内で、ビジネスの拡張に伴い、書き込まれる半構造化データに新しいフィールドが追加されたり、既存のフィールドが変更されたりして、より多くのビジネスプロセスやジョブに対応する場合があります。Avro をはじめとする多くのフォーマット規格には、このような変更に対応する互換性設定が用意されています。具体的には、別の書き込み側が新しいスキーマのデータを書き込む場合、既存のオンラインリーダーが引き続き動作することを保証すると同時に、新しいフィールドを必要とする新規リーダーのデータ読み取りもサポートする必要があります。したがって、Pravega Schema Registry の存在により、読み取り側が実際に逆シリアル化を試みるまで変更を認識できないという状況を避け、書き込み側でフォーマットの互換性を設定し、新しいスキーマの登録時に何らかの介入を行うことで、読み書きするクライアントアプリケーションの管理をより迅速に行えることが理想的です。

2. プロジェクトの紹介
こうした動機に基づいて、Pravega Schema Registry プロジェクトを開発しました。

これは、Pravega 内の半構造化データのスキーマを保存・管理し、保存されたスキーマ、データエンコード形式、互換性ポリシーを管理する RESTful インターフェイスを提供します。提供するインターフェイスの抽象化は非常にオープンで、Avro、protobuf、Json などの一般的なシリアル化形式や lz4、snappy などの圧縮アルゴリズムに加えて、カスタムシリアル化方式もサポートしています。この抽象化は、業界の他の類似プロジェクトよりも汎用的なソリューションです。各シリアル化の管理では、対応するエンコード形式と互換性ポリシーを自由にカスタマイズでき、ユーザーは任意のシリアル化方式でデータを処理できます。プロジェクト全体は Pravega Key Value Table の機能を使用して保存されており、これが Schema Registry の基盤実装となっています。Pravega Key Value Table は、以前は Pravega の内部メタデータ保存用コンポーネントの一つに過ぎませんでしたが、その後パブリック API として開発が進められ、バージョン 0.8 で一般公開され、バージョン 0.10 でベータ安定版となりました。

Pravega Schema Registry プロジェクトは実際には比較的独立したプロジェクトです。基盤実装は Pravega Key Value Table を使用していますが、上位層のすべての抽象化は Pravega に限定されない独立した設計になっています。このプロジェクト全体は Pravega 内でオープンソース化されているのではなく、エコシステム内の独立したプロジェクトとして存在しています。これにより、ファイルストレージやオブジェクトストレージなど、より汎用的なストレージシステムもスキーマ管理ソリューションとして利用できます。

3. システムアーキテクチャ

プロジェクト全体のシステムアーキテクチャを図に示します。

Schema Registry は RESTful API と gRPC プロトコルを通じてクライアントとやり取りします。Group はスキーマ管理の単位に対応し、Pravega では Stream に対応します。デフォルトのシリアル化形式、互換性設定、複数バージョンのシリアル化とエンコード情報が保存されています。前述の通り、これらの情報は Key-Value ペアの形式で Pravega セグメントに保存されます。

データリンクの観点からは、書き込み側はプロトコルバージョンとエンコーディング ID を含むヘッダーバイトを持つ特別なイベントシリアル化方式を使用する必要があります。これにより、Schema Registry が介入してデータのスキーマを登録または検証し、エンコードと互換性の要件に適合しているか確認した上で、適法なスキーマのデータを Pravega ストリームへの保存を許可します。同様に、読み取り側もこのような特別な逆シリアル化方式で読み取る必要があります。

2. Catalog API 統合

Schema Registry により、Catalog と Pravega のマッピング関係がより明確になります。

図に示すように、ストリーム内のフローデータと Schema Registry に保存されたスキーマを組み合わせることで、テーブル構造に基づいて Flink Catalog 内のテーブルを正常に逆シリアル化および合成でき、テーブルからストリームへの変換も逆方向に可能です。これでリンク全体が完結します。したがって、テーブルの作成、テーブルの削除、データベースの削除といった従来のカタログ操作は、Pravega と Schema Registry のメタデータ変更として抽象化できます。

この理論に基づいて、カタログインターフェイスを初期実装しました。ユーザーは以下の DDL を使用してカタログを作成し、SQL で Pravega メタデータを操作できます。

実装上の課題
ただし、初期バージョンのプロトタイプをさらに洗練させる段階で、実装の詳細に関する 3 つの課題に直面しました。

1 つ目は、スキーマとシリアル化の処理です。Flink バージョン 1.12 より前では、Json および Avro データと Flink テーブル行レコード間の内部抽象 RowData 変換プロセスは、Flink のフォーマット下の非公開クラスでした。そのため、このコードを再利用して Flink のスキーマおよびシリアル化変換動作と整合性を保つには、クラスライブラリ全体をコピーする必要がありました。そこで、これを抽象化してパブリッククラスに変換できないかコミュニティに提案しました。コミュニティとの議論と調整を経て、対応する JIRA FLINK-19098 を作成し、この問題を解決するコードを貢献しました。これにより、Catalog テーブルでデータシリアル化を変換するリンクが正常にオープンになりました。
2 つ目は、Avro と Json の両フォーマットをサポートすることです。既存のカタログ実装では、シリアル化方式は比較的固定されており、これは Pravega Schema Registry の設計コンセプトにおける汎用性とやや矛盾していました。ユーザーの中には Avro を好む人もいれば、Json を好む人もいます。どうすれば両方のニーズに対応し、カタログがもたらす利便性を享受してもらえるでしょうか。そこで、カタログにシリアル化フォーマットのオプションを導入しました。Avro と Json の指定に加えて、タイムスタンプのシリアル化形式など、Flink が公式にサポートするすべてのシリアル化追加設定を指定でき、カタログ全体に適用されるシリアル化方法をさらに細かくカスタマイズできます。
3 つ目の課題は、Schema Registry を通じて処理するイベントの先頭に 5 バイトのヘッダーが付くため、Flink の既存の Json や Avro フォーマットを直接使用してシリアル化できないことです。Schema Registry が提供するシリアライザー API を使用するには、独自のフォーマットファクトリのセットを開発して呼び出す必要がありました。実装上は、Catalog テーブルのシリアル化時に、Schema Registry API に必要なパラメータ（名前空間、グループなどの情報）をフォーマットファクトリのパラメータにシリアル化します。これは実質的に以下のテーブル作成 DDL に相当します。

その後、実装内でこの情報を取得して Schema Registry API を呼び出し、前述の問題に対する修正を組み合わせることで、完全なバイナリデータと Flink RowData の相互運用を実現しました。これで全体のリンクが正常に機能します。

3. Debezium サポート

まず、大規模 CDC 全体の概念について説明します。CDC の正式名称は変更データキャプチャ (Change Data Capture) で、データの変更を識別して追跡し、それに基づいてアクションを実行する手法です。ただし、これは広義の概念であり、業界のより現代的な実践では、CDC はより狭義の技術用語として使われることが多くなっています。具体的には、データベースのログ分析を行い、それを特定のフォーマットのデータストリームに変換する技術を指します。代表的な実装には Debezium や、国内でよく使われる Canal などがあります。

Debezium は業界で最も広く使用されている CDC 技術であり、Kafka Connect ベースで実装された分散プラットフォームで、データベースの行レベルの変更をイベントストリームに変換します。CDC 技術は現在、バックアップとディザスタリカバリのためのデータ同期、複数の下流システムへのデータ配信、データレイクへの接続のための ETL 統合など、幅広い応用シーンがあります。

1. デプロイ方式

現在、Debezium には 3 つのデプロイ方式があります。

1 つ目は、業界で最も一般的に使用されている方式で、Kafka Connect を使用したデプロイです。前述の通り、これは Debezium が当初からサポートしている方式です。MySQL や Postgres などの従来のデータベースの binlog を分析し、Kafka Connect の Debezium ソースコネクタを通じて Apache Kafka に取り込みます。Kafka の強力なエコシステムを活用して、さまざまな下流エンジンと連携し、集約計算やデータウェアハウス、データレイクのアプリケーションを実現できます。
2 つ目は、Debezium コミュニティがメッセージキューとストリームストレージの分野の変化に対応して、Kafka から徐々に分離してきたことです。現在では、ソースコネクタを使用して独立した Debezium サーバーを起動し、下流のメッセージングシステムに接続できます。Amazon Kinesis や Google PubSub との統合も用意されています。2021 年前半に Debezium コミュニティはバージョン 1.6 をリリースしました。このバージョンで、Debezium は Pravega コミュニティの貢献を正式に受け入れました。Pravega 側のシンクもソースコネクタの実装の一つとなり、Debezium コミュニティとの連携が実現しました。

最後の方式は、Debezium を依存関係ライブラリとして Java プログラムに埋め込んで呼び出す方式です。コミュニティで人気の高い Flink CDC コネクタが代表的です。CDC データの長期保存や再利用が不要なシナリオでは、この軽量な実装により、メッセージキューのデプロイとメンテナンスの複雑さを排除しつつ、計算の信頼性とフォールトトレランスを確保できます。

2. 書き込み方式

Debezium と Pravega の統合において、実装の整合性を保ちつつ、通常の書き込みとトランザクション書き込みの 2 つの方式を提供しました。

Debezium サーバーは、実際にはソースコネクタで定期的にバッチプルするプロセスであり、インターフェイスは Debezium バッチのアップサートストリームを受信します。右側の図から分かるように、各括弧はプルされたバッチを表し、黄色は更新、白は挿入を示しています。

通常の書き込みの場合、挿入と更新を問わず、すべてのイベントを順番に個別のイベントとして書き込みます。分散環境での順序を心配する必要はありません。各 Debezium イベントはデータベーステーブルのキーを持っているため、Pravega への書き込み時に対応するルーティングキーを付けることで、同じキーのイベントが同じセグメントに入るようになり、Pravega は同じルーティングキー上のデータの順序を保証します。

次に、トランザクション書き込みについて説明します。各 Debezium バッチについて、Pravega はバッチ全体を 1 つのトランザクションにパッケージ化し、バッチが完了した時点で Pravega トランザクションをコミットします。これにより、Debezium サーバーでフェイルオーバーが発生しても、Pravega トランザクションの原子性と冪等性の保証により、すべてのイベントを重複なく再生・出力できるため、exactly-once セマンティクスが保証されます。

バージョン 1.6 以降、ユーザーは表中のパラメータを使用して Debezium Server を設定できます。対応する Pravega 接続パラメータを入力し、スコープ名とトランザクション書き込みスイッチを指定するだけで、MySQL などのデータベースからのリアルタイム同期により、データベース内のすべてのテーブルの変更が Debezium メッセージフォーマットでテーブル名と同じ名前の Pravega ストリームに書き込まれます。

3. コネクタ統合

Debezium に対する Pravega の貢献に加えて、Flink のストリーム処理機能を使ってデータを消費するには、Pravega Flink Connector のコンピュート側でも対応する統合を行う必要があります。

既存の FLIP-95 Table API 上に Table Factory を実装し、基本的な読み取りと書き込み機能をサポートしました。Flink Table API をご存知の方は、コミュニティが既に debezium-json のようなフォーマットファクトリの実装を提供しているのだから、対応するフォーマットを指定するだけで簡単に使えるのではないかと思われるかもしれません。どのような課題があるのでしょうか。

当初は同じように考えていましたが、実際にはそう単純ではありませんでした。主に 2 つの課題に直面しました。

1 つ目は、複数イベントへの逆シリアル化機能を追加でサポートする必要があることです。

Debezium はアップサートストリームを提供します。Flink テーブルの RowData 抽象への変換において、挿入イベントは通常の 1 対 1 の逆シリアル化プロセスですが、更新イベントについては、更新前と更新後の 2 つの状態に変換する必要があります。これは、デフォルトの 1 対 1 のコネクタシリアル化実装とは完全に矛盾しています。

Pravega 自身のシリアライザーインターフェイスも 1 対 1 マッピングであるため、Pravega シリアライザーと Flink の DeserializationSchema インターフェースの相互運用性を確保するために、多くのコード修正を行いました。この新しい要件をサポートするため、以前の逆シリアル化リンクを再構築し、元の逆シリアル化プロセスを Pravega クライアント側から Pravega コネクタの内部に移動しました。これにより、以下のような Collector を使用する deserialize メソッドを利用できます。

default void deserialize(byte[] message, Collector out) throws IOException {
同時に、コード修正全体において、元の Pravega シリアライザーと Flink API 変換 API の互換性を維持するよう細心の注意を払い、オンラインユーザーのアップグレード性に影響を与えないようにしました。

2 つ目は、FLIP-107 の Table Source 側のメタデータサポートに関するものです。

Pravega Table Source は、Flink が提供する SupportsReadingMetadata インターフェイスを実装し、このサポートを提供しています。ユーザーは、`from-format` プレフィックスでフォーマットファクトリ自体のメタデータ（Debezium の取り込みタイムスタンプやテーブル名などのメタ情報）を指定して、テーブルの情報を補完・拡充できます。同時に、Pravega 自体のメタデータ（EventPointer と呼ばれ、現在のイベントのストリーム内の位置情報を記録するもの）もサポートしています。この情報を記録することで、ユーザーはこのデータを保存し、後続のランダム読み取りやインデックス作成などのニーズに対応できます。メタデータを含むテーブル作成の DDL は以下の通りです。これらの新規作成列は、元のデータ構造の後に Flink Table の抽象内に順番に並んで表示されます。

以上が、Pravega から Connector までの完全な Debezium サポートのプロセスです。

4. コミュニティ共同ホワイトペーパーのリリース

CDC によるデータベースのリアルタイム処理は、データ同期、配信、リアルタイムデータウェアハウスなど、ビッグデータ業界で非常に重要な応用シーンです。Debezium の統合により、Apache Flink と Debezium を使用して、Pravega を統一されたメッセージ中間ストレージレイヤーソリューションとして活用し、ユーザーのデータベース同期ニーズに対応できます。このソリューションでは Pravega をメッセージミドルウェアとして使用することで、Pravega のリアルタイム永続ストレージの特性を最大限に活用でき、リンクのミリ秒レベルのリアルタイム性能とデータ整合性を確保した上で、他のアプリケーションにも信頼性の高いデータストレージとデータ再利用を提供できます。同時に、Apache Flink の豊富な下流エコシステムを活用して、さまざまなシステムに簡単に接続し、多様なビジネスニーズに対応できます。統合プロセス全体は、Pravega コミュニティと Flink コミュニティの協力によって実現されました。