Zenn
😊

ストリーミングデータをSQLでIcebergに管理する：RisingWave における Iceberg テーブルエンジンの紹介

に公開
SQL
DB
データエンジニアリング
iceberg
リアルタイムデータ分析
tech

Apache Iceberg は、モダンなデータレイクにおけるテーブルフォーマットのオープン標準として急速に広まっています。その強力な機能であるスキーマ進化、隠れたパーティショニング、タイムトラベル、および ACID準拠により、スケーラブルで信頼性が高く、ベンダーニュートラルなデータアーキテクチャが可能になります。

しかし、Icebergはストレージフォーマットを定義するだけであり、データの取り込みや処理、特にリアルタイムのストリーミング処理の複雑性については別システムに任されています。Trino や Athena のようなクエリエンジンは静的データセットを扱うのには優れていますが、ストリーミングデータをIcebergに連続して低遅延で取り込み、変換するためには設計されていません。そのため、複数の複雑なツールを統合する必要が生じ、運用のオーバーヘッドと脆弱性が増すことが多々あります。

Icebergへのストリーミングにおける課題

様々なエンジニアリングチームとの協力から、Icebergにシンクするリアルタイムパイプラインを構築する際に共通の課題があることが分かっています。

1. リアルタイムの ETL は必要だが複雑

生のストリーミングデータはIcebergに取り込む前に変換を要することが多いです。コンプライアンス要件として、機微情報のフィールドをフィルタリングすることが義務付けられたり（GDPRのためのPIIマスキングなど）、ビジネスロジックとしてイベントを参照データと結合してエンリッチする必要があります。こうした連続的なETLを従来のバッチ処理ツールや特化したストリーミングシステムで堅牢に実装するのは、複雑かつリソースを多く消費します。

2. CDC (Change Data Capture) の適切な取り扱い

オペレーショナルデータベース（Postgres、MySQLなど）からCDC経由でIcebergへ変更を伝搬するには注意が必要です。Icebergテーブルは通常ストレージ層でのプライマリーキー強制がないため、更新と削除はしばしば**等価削除（equality deletes）**で実装する必要があります。ストリーミングの取り込みエンジンは以下を確実に行う必要があります。

  • CDCフォーマット（Debeziumなど）を解釈する。
  • 変更を決定論的に適用する。
  • 障害やスケーリング時にデータ損失や重複を防ぐため、Exactly-once処理を保証する。

3. 低遅延なデータ可視化の必要性

エンジニアやアナリストは、検証、デバッグ、品質保証、リアルタイム監視のために取り込み中のデータを迅速にクエリする必要があります。従来のパイプラインはマイクロバッチ間隔や非同期のコンパクションプロセスにより、Icebergでデータがクエリ可能になるまでに大きな遅延が発生します。

一般的なアプローチの限界

Icebergへストリーミングする一般的な手法には、KafkaやFlinkを使うものがありますが、それぞれにトレードオフがあります。

  • Kafka： データ転送には優れていますが、ネイティブなデータ処理機能を欠いています。変換、結合、CDCの処理などを行うには、他のツール（Kafka Streams、ksqlDB、カスタムアプリケーション）との統合が必要であり、複雑性を増す要因となります。
  • Flink： ETLとIcebergへの書き込みが可能な強力なストリームプロセッサです。しかし、習得に急な学習曲線（複雑なジョブではJavaやScalaが必要）があり、またSQL方言が標準的なSQLと異なります。さらに、大規模な多方向結合のような複雑な操作に対して状態を管理することは難しい場合があります。

こうしたシステムをスケールさせ、信頼性を持って運用するには、かなりのエンジニアリング投資が必要です。そのため、Icebergへのストリーミング投入と管理をシンプルにし、統合されたSQLネイティブな体験を提供できるソリューションが求められています。

RisingWaveのIcebergテーブルエンジンの紹介

RisingWave の Icebergテーブルエンジン は、このような課題を直接的に解決するよう設計されています。RisingWave内でIcebergテーブルをネイティブのテーブルとほぼ同じように扱うことができ、標準的なSQLでテーブルの定義、取り込み、変換を行えます。RisingWaveの分散型ストリーム処理エンジンの上に構築されており、Iceberg向けのリアルタイムデータ処理を統合的に行うことが可能です。

主な機能

  • SQLによる統合テーブル管理： 標準的なSQL DDL (CREATE TABLE ... ENGINE = 'iceberg';) を使ってRisingWave内でIcebergテーブルを直接作成・管理できます。他のテーブル同様、SQLのDMLである INSERT、UPDATE、DELETE 操作が可能です。
  • ネイティブCDC取り込みおよび処理： CDCソース（例：Postgres、MySQLなどDebezium経由）に直接接続できます。RisingWaveは変更イベントを自動的に解釈し、Icebergへの書き込み時に UPDATE と DELETE 操作をequality deleteとして正しく適用し、データの一貫性を保証します。また、RisingWaveの障害耐性メカニズムにより、Exactly-once処理のセマンティクスが維持されます。
  • SQLによるリアルタイムETL： 複雑なストリーミング変換（フィルター、結合、集約）を標準SQLで定義可能です。RisingWaveはこれらを増分的に更新されるストリーミングジョブとして実行し、入力データを継続的に処理して結果を対象となるIcebergテーブルに書き込みます。
  • 低レイテンシでのデータクエリ可能性： RisingWave内のテーブルやマテリアライズドビューの状態を低遅延でクエリできるため、データがIcebergファイルにコミットされ圧縮処理される前でも即時の検証や検査が可能です。
  • ネイティブIcebergフォーマット統合： このエンジンは公式のIcebergライブラリを使用してデータを書き込むため、完全な互換性を提供します。パーティショニングやスキーマ進化といったIcebergの機能もサポートしています。テーブルは標準のIcebergフォーマットで保存され（例えば S3）、共有カタログ（AWS Glue、AWS S3 Tables、REST、JDBC経由）を通じ、他のIceberg互換エンジン（Trino、Spark、Athena、DuckDBなど）からもクエリ可能です。必要に応じて外部ツールによるコンパクションも行えます。

構成例：PostgresのCDCをRisingWave経由でIcebergにストリーミングする場合

一般的な構成例としては以下のようになります。

  1. ソース： PostgresデータベースからCDCイベントを出力（Debeziumフォーマット経由）。
  2. 取り込みおよびストレージ： CDCデータをRisingWaveのみで取り込み、Icebergテーブルへ格納します。通常は数行のSQLコマンドと簡単な設定だけで可能です。
--- テーブルエンジン用のコネクションを作成
CREATE CONNECTION ...

--- テーブルエンジンのための接続設定
SET iceberg_engine_connection = 'public.conn';
ALTER system SET iceberg_engine_connection = 'public.conn';

--- ソースの作成
CREATE SOURCE pg_source WITH (
    connector='postgres-cdc',
    hostname='localhost',
    port='5432',
    username='your_user',
    password='your_password',
    database.name='your_database',
    schema.name='public' -- オプション、省略時は'public'
);

-- ソース内のテーブルからIcebergテーブルを作成
CREATE TABLE my_table (
    id INT PRIMARY KEY,
    name VARCHAR
)
FROM pg_source TABLE 'public.my_upstream_table'
ENGINE = iceberg;

このアーキテクチャでは、Kafka、Flink、またはカスタムコードを個別に展開する際の典型的な複雑性を避け、リアルタイムでのデータ取り込みとIcebergへのストレージを提供します。

まとめ

Apache Icebergは、データレイクの堅牢な基盤を提供しますが、リアルタイムのストリームデータの統合は依然として課題です。RisingWaveの Icebergテーブルエンジン は次のような利点でこの課題をシンプル化します。

  • SQLベースの管理 をストリーミングの文脈で提供。
  • CDCストリームとequality deleteを 内蔵で処理。
  • 標準SQLを使用した リアルタイムETL 機能。
  • ストリーミングデータパイプラインを 低レイテンシで可視化。
  • Icebergエコシステムとの 完全な互換性。

Icebergをターゲットにしたストリーミングデータパイプラインを構築していて、より統合的で効率的かつSQLフレンドリーな手法を探しているなら、まもなく登場するバージョン2.3のRisingWave Icebergテーブルエンジンをぜひお試しください。

詳しくはこちら：

Discussion