本エントリでは、RDBMSからIcebergへのCDCユースケースでの活用が期待されるIceberg Sink Connectorの現状と留意点を整理します。
CDCが必要とされる背景
データの分析基盤の一般的な要件として、業務システムのRDBMS上のデータを連携したい場面がしばしばあります。
これを実現する方法として様々なアプローチが考えられますが、その一つにChange Data Capture(CDC)が挙げられます。CDCによるデータ連携では、ソースDB上のイベント(MySQLのbinlogなど)をDebeziumなどの仕組みでキャプチャして、分析基盤に反映します。
CDCの重要な特徴は、ソーステーブルでのUPDATEやDELETEなどの変更差分をターゲットとなるテーブルへ効率よく反映できることです。
CDCを使用せず、ソースとなるテーブルのある時点での断面を取得して反映する場合、あるレコードがソースから更新された/消えた事実を直接的に把握できないため、UPDATEやDELETEを反映する際はソースとターゲットの全レコードを突き合わせるか、テーブル全体を洗い替える必要があります。ある程度規模の大きいテーブルでこうした処理を運用するのは処理速度やリソース効率の観点で現実的ではありません。
これに対して、CDCではDB上の変更イベントをベースにターゲットのテーブルを更新するため、変更差分のみを効率よく処理できるというわけです。
なお、CDCの特徴としてリアルタイム性にも注目されることが多いものの、CDCは本質的には変更差分をベースとするレプリケーションを指すもので、必ずしもその処理がリアルタイムである必要はありません。ある期間の変更イベントを集積しておき、バッチベースでまとめて反映していくような実装も検討できます。ただ、リアルタイムなデータ連携を実施する際には、差分データの取り込み方式としてCDCを用いる効率が良いため、結果的に併せて使用されることが多い構図となっています。
Apache Iceberg Sink Connector
RDBMSからIcebergテーブルへのCDCを実現する際に用いられるコンポーネントとして、Icebergプロジェクトが提供するKafkaのIceberg Sink Connectorがあります。
Iceberg Sink ConnectorはKafkaのメッセージを外部と連携するためのフレームワークであるKafka Connectとして開発されています。たとえば、DebeziumなどのDBの変更イベントをキャプチャする仕組みを用いてイベントをKafkaへ配信し、そのメッセージをIceberg Sink ConnectorによってIcebergテーブルに配信していく連携フローを構成できます。
Kafka Iceberg Sink Connectorについての詳細は、yassanが詳しく解説しています。
developers.microad.co.jp
また、Slack社は、SlackへのあらゆるメッセージをアプリケーションのMySQLからIcebergへCDCするユースケースでこのコネクタを活用しています。
ただし、上記の2つはIcebergプロジェクトで提供されているIceberg Sink Connectorではなく、Tabular社によって提供されているIceberg Sink Connector(後述)をベースとしている点に注意が必要です。
2つのIceberg Sink Connector
開発の歴史的経緯
Iceberg Sink Connectorは、当初Icebergのコミッターが多数所属するTabular社によって開発され、同社のリポジトリで公開されていました。
github.com
このコネクタは後にASF側に寄贈され、これが現在Icebergプロジェクトで提供されているIceberg Sink Connectorとなっています。
github.com
Tabular社によるオリジナルなコネクタは現在でもGithubで公開されていますが、こちらは現在では開発は止まっており、新規の改善や機能追加はIcebergプロジェクト側のリポジトリで行われています。
従って、新規にIceberg Sink Connectorの利用を検討する場合には、基本的にはIcebergプロジェクト側の実装を利用することになると思います。
Tabular側の実装とIcebergプロジェクト側の実装での挙動の違い
ここからが本題です。
Iceberg Sink Connectorには、TabularからIcebergプロジェクト側へ移管された後にいくつかのBreaking Changeが入っています。
従って、各種挙動がTabular実装時代と異なる部分がいくつかあります。たとえば、Tabular側のコネクタでUpsertを有効化する際に指定するconfigであるiceberg.tables.upsert-mode-enabledはIcebergプロジェクト側の実装には存在しません。
頭の痛い状況として、現在インターネットなどで流通しているIceberg Sink Connectorに関する情報は、多くがTabular側の実装を前提としています。(少し前までそちらしか存在していなかったので、仕方がないことではあるのですが)
よって、コネクタの利用を検討する際は、その点に留意した上で、参照している情報がどちらの実装を前提にしているものかを把握しておく必要があります。
Icebergプロジェクト側のコネクタはUPDATE / DELETEを反映できない
特に大きなTabular側の実装との差分として、Icebergプロジェクト側のIceberg Sink Connectorには、delta writerが実装されていません。
delta writerというのは、データソースの更新イベントをIcebergのテーブルに反映するための仕組みです。つまり、delta writerが実装されていないということは、Icebergプロジェクト側のコネクタはUPDATE / DELETEを反映できないということです。
元々、Tabular側の実装はdelta writerがあるため、更新を問題なく反映できました。
しかし、Icebergプロジェクトへ移管される際に「変更の反映がequality deletesに依存していることによる性能への懸念」を理由として削除されています。
The initial Kafka Connect sink submission did not include delta writer support that the Tabular version has, as there are performance concerns over relying on equality deletes. We should address those concerns and add back the delta writer support, for features like change data capture and upsert mode.
現在、新たにIcebergプロジェクト側に改善版のdelta writerを導入する動きがあり、開発が進んでいます。
github.com
よって、上記issueが解決するまでは、Icebergプロジェクト側のKafka Iceberg Sink ConnectorをUPDATE / DELETEを含むテーブルのCDCを目的として使用することはできません。一方でTabular側の実装は既に開発が止まっているため、今から利用するのは難しい状態にある、というのが現状です。
ワークアラウンドとして、SparkやFlinkなどによってDebeziumがキャプチャしたDBの変更イベントを反映する処理を実装するアプローチが考えられます。カスタム実装が求められるものの、処理内容を細かくコントロールできる利点もあるため、状況が許すのであれば悪くない方法かと思います。
参考までに、以下の筆者のOpenSearchのデータ取り込みに関する発表では、IcebergへのCDCにSparkを用いるアプローチを紹介しています。
まとめ
RDBMSからIcebergへのCDCを設計する際の道具として、IcebergプロジェクトはKafka Iceberg Sink Connectorを提供しています。このコネクタを活用する際には、Tabularから寄贈された歴史的経緯によって２つの実装が存在しており、両者に差分がある点を理解しておくことが重要です。特に大きな差分として、Icebergプロジェクト側の実装では現状はUPDATE / DELETEを反映できません。従って、それらを含むCDCを行いたい場合は、SparkやFlinkなどによる実装を検討する必要があります。
ただ、Iceberg Sink Connectorについても活発に改善が進んでいるため、今後のIcebergへのデータ連携を支える強力なソフトウェアとして今後に期待したいところです。
