Use Flink Hudi to build a streaming data lake platform

1. Apache Hudi の基礎

データレイクといえば、誰もがこのような疑問を持つだろう。データレイクとは何か。なぜこの 2 年でデータレイクが非常に人気になったのか。データレイクは実際には新しい概念ではない。データレイクの最初の概念は 1980 年代に提唱された。当時のデータレイクの定義はオリジナルデータ層であり、さまざまな構造化、半構造化、さらには非構造化データを格納できるものだった。機械学習やリアルタイム分析など多くのシナリオでは、データのスキーマはクエリ時に決定される。

レイクストレージは低コストと高い柔軟性という特徴を備え、クエリシナリオにおける集中管理に非常に適している。近年のクラウドサービスの台頭、特にオブジェクトストレージの成熟に伴い、ますます多くの企業がクラウド上にストレージサービスを構築することを選択している。レイクストレージのストレージとコンピューティングの分離アーキテクチャは、現在のクラウドサービスアーキテクチャに非常に適している。スナップショット分離による基本的な ACID トランザクションを提供し、複数の分析エンジンとの連携をサポートして、さまざまなクエリシナリオに対応できる。レイクストレージはコスト効率とオープン性の両面で大きな優位性を持っていると言える。

現在のレイクストレージはデータウェアハウスの機能も担い始めており、計算エンジンとの接続によりレイクとウェアハウスの統合アーキテクチャを実現している。レイクストレージは、元のデータ形式をベースにテーブルの高度なセマンティクスをカプセル化したテーブルフォーマットである。Hudi は 2016 年からデータレイクを実践に移してきた。当時、ビッグデータシナリオにおけるファイルシステム上のデータ更新問題を解決することが目的だった。Hudi の LSM ライクなテーブルフォーマットは現在のレイクフォーマットの中でも独自の存在であり、準リアルタイム更新に適している。また比較的完成度の高い実装となっている。

テーブルフォーマットは現在人気のある 3 つのデータレイクフォーマットの基本的な属性であり、Hudi はプロジェクト開始当初からプラットフォーム方向へ進化してきた。データガバナンスやテーブルサービスなどの機能が比較的充実している。たとえば、ユーザーが書き込みを行う際に、同時にファイルのレイアウトを最適化でき、メタデータテーブルによりクエリ側のファイル検索効率を大幅に最適化できる。

以下では Hudi のいくつかの基本概念を紹介する。

タイムラインサービスは Hudi のトランザクション層の中核抽象化である。Hudi 内のすべてのデータ操作はタイムラインサービスを中心に展開される。各操作はインスタント抽象化を通じて特定のタイムスタンプにバインドされる。一連のインスタントがタイムラインサービスを構成する。各インスタントレコードには対応するアクションと状態が記録される。Hudi はタイムラインサービスを通じて、現在のテーブル操作のステータスを把握できる。ファイルシステムビューの一連の抽象化とタイムラインサービスを組み合わせることで、特定のタイムスタンプ時点のファイルレイアウトビューを現在のテーブルのリーダーとライターに公開できる。

ファイルグループはファイルレイアウト層における Hudi の中核抽象化である。各ファイルグループはバケットに相当し、ファイルサイズに基づいて分割される。各書き込み動作は新しいバージョンを生成し、バージョンはファイルスライスとして抽象化される。ファイルスライス内部では対応するバージョンのデータファイルが管理される。ファイルグループが指定されたファイルサイズに達すると、新しいファイルグループに切り替えられる。

Hudi のファイルスライスへの書き込み動作は、Copy on Write (CoW) と Merge on Read (MoR) の 2 つのセマンティクスに抽象化できる。

Copy on Write (CoW) は毎回フルデータを書き込み、新しいデータは前のファイルスライスのデータとマージされてから、新しいファイルスライスとして書き出され、新しいバケットファイルが生成される。

Merge on Read (MoR) はより複雑である。そのセマンティクスは追記書き込み、つまり毎回差分データのみを書き込むため、新しいファイルスライスは書き出されない。まず前のファイルスライスへの追記を試み、追記されたファイルスライスがコンパクションプランに含まれた時点で、新しいファイルスライスが生成される。


2. Flink Hudi 統合

Flink Hudi の書き込みパイプラインはいくつかのオペレーターで構成される。最初のオペレーターはテーブル層の RowData を Hudi のメッセージフォーマットである HudiRecord に変換する。次に Bucket Assigner を通じて、変換された HudiRecord を特定のファイルグループに割り当てる。ファイルグループに割り当てられたレコードは Writer オペレーターに流れ込み、実際のファイル書き込みが行われる。最後にコーディネーターがあり、Hudi テーブル層のテーブルサービススケジューリングと新規トランザクションの開始およびコミットを担当する。さらに、バックグラウンドで古いバージョンのデータをクリーンアップする役割もある。


現在の設計では、各バケット割り当てタスクはバケットアサイナーを保持し、独自のファイルグループ群を管理する。新規データや更新なしの挿入データを書き込む際、バケット割り当てタスクはファイルビューをスキャンし、小規模バケットと判定されたファイルグループを優先的に選択してこの新規データを書き込む。

たとえば、上の図ではファイルグループのデフォルトサイズが 120 MB である。左図の task1 はまず file group1 と file group2 に書き込む。file group3 には書き込まない点に注意する。file group3 は既に 100 MB のデータがあり、ターゲットしきい値に比較的近いため、過大な書き込み増幅を避けるためにこのバケットには書き込まない。右図では、task2 は直接新しいファイルグループを書き込み、既に書き込まれた大きいファイルグループには追記しない。

次に、Flink Hudi 書き込みプロセスの状態遷移メカニズムを紹介する。ジョブ起動直後、コーディネーターはまずファイルシステム上にこのテーブルを作成しようとする。現在のテーブルが存在しない場合、ファイルディレクトリにメタ情報を書き込み、テーブルを作成する。すべてのタスクの初期化メタ情報を受信した後、コーディネーターは新しいトランザクションを開始し、書き込みタスクはトランザクションの開始を確認してから現在のデータのフラッシュ動作を有効にする。

書き込みタスクはまずデータをバッファーに蓄積する。ここには 2 つのフラッシュ戦略がある。1 つは現在のデータバッファーが指定サイズに達した場合に、メモリ内のデータがフラッシュされる。もう 1 つは上流からのチェックポイントバリアがスナップショット取得を要求した場合に、メモリ内のすべてのデータがディスクにフラッシュされる。各フラッシュの後、メタ情報がコーディネーターに送信される。コーディネーターはチェックポイントの成功イベントを受信すると、対応するトランザクションをコミットし、次の新しいトランザクションを開始する。ライタータスクが新しいトランザクションを確認すると、次のラウンドのトランザクションの書き込みが有効になる。このようにして、書き込みプロセス全体が直列に接続される。

Flink Hudi の書き込みは非常に豊富なシナリオをサポートしている。現在、更新なしデータ型のログデータへの書き込みをサポートし、小ファイルマージにも対応している。さらに、更新ストリームや CDC データといった Hudi のコア書き込みシナリオもサポートされている。同時に、Flink Hudi は過去データの高効率なバッチインポートもサポートしている。バケット挿入モードにより、Hive やデータベース内のオフラインデータをバッチクエリを通じて一度に Hudi フォーマットへ効率的にインポートできる。さらに、Flink Hudi はフルインデックスとインクリメンタルインデックスのロードも提供している。ユーザーはバッチデータをレイクフォーマットに一度に効率的にインポートした後、ストリーム書き込みプログラムとの連携によりフルデータおよび増分データの取り込みを実現できる。

Flink Hudi の読み取り側も非常に豊富なクエリビューをサポートしている。現在、主にフル読み取り、過去の時間範囲の増分読み取り、およびストリーム読み取りをサポートしている。

上図は Flink SQL を通じて Hudi に書き込む例である。Hudi は多くのユースケースをサポートし、ユーザーが設定すべきパラメータを可能な限り簡素化している。テーブルパス、同時実行数、操作タイプを簡単に設定するだけで、ユーザーは上流データを Hudi フォーマットに容易に書き込める。

3. Flink Hudi のユースケース

以下では Flink Hudi の代表的なアプリケーションシナリオを紹介する。

1 つ目の代表的なシナリオは、DB からデータレイクへのデータ取り込みである。現在、DB データをデータレイクに取り込む方法は 2 つある。CDC コネクタを通じてフルデータと増分データを一度に Hudi フォーマットに取り込むことができる。また、Kafka 上の CDC 変更ログを消費して、Flink の CDC フォーマットを通じてデータを Hudi フォーマットに取り込むこともできる。

2 つ目の代表的なシナリオは、ストリームコンピューティングによる ETL（準リアルタイム OLAP 分析）である。上流ストリームを接続したシンプルな ETL（デュアルストリーム Join やデュアルストリーム Join を使った集計など）を計算し、変更ストリームを Hudi フォーマットに直接書き込む。その後、下流の読み取り側で Presto や Spark などの従来の代表的な OLAP エンジンに接続することで、エンドツーエンドの準リアルタイムクエリを実現できる。

3 つ目の代表的なシナリオは 2 つ目とやや似ている。Hudi はネイティブの変更ログをサポートしており、Flink 計算内で行レベルの変更を保存できる。この機能に基づき、ストリーム読み取りによる変更の消費を通じて、エンドツーエンドの準リアルタイム ETL 処理を実現できる。

今後、コミュニティの 2 つのメジャーバージョンはストリーム読み取りとストリーム書き込みに注力し、ストリーム読み取りのセマンティクスを強化する。さらに、カタログとメタデータの面では自己管理を実装する。また、近い将来 Trino ネイティブコネクタをリリースし、現在の Hive 経由の読み取り方式を置き換えて効率を向上させる計画である。

4. Apache Hudi ロードマップ

以下は MySQL から Hudi への大量テーブル取り込みのデモである。

まず、データソースとして benchmark1 と benchmark2 の 2 つのデータベースを用意した。benchmark1 配下には 100 テーブル、benchmark2 配下には 1,000 テーブルがある。大量テーブル取り込みはカタログに強く依存するため、まずカタログを作成する必要がある。データソース用に MySQL カタログを作成し、ターゲット用に Hudi カタログを作成する。MySQL カタログはすべてのソーステーブルに関する情報（テーブル構造、テーブルデータなど）を取得するために使用する。Hudi カタログはターゲットの作成に使用する。

2 つの SQL ステートメントを実行すると、2 つのカタログが正常に作成される。

次に、ジョブ開発画面に移動して大量テーブル取り込みのジョブを作成する。わずか 9 行のシンプルな SQL で実現できる。1 つ目の構文は CREATE DATABASE AS DATABASE である。その機能は、MySQL の benchmark1 ライブラリ配下のすべてのテーブル構造とテーブルデータを Hudi の CDS demo ライブラリに一括同期することである。テーブル間の関係は 1 対 1 のマッピングである。2 つ目の構文は CREATE TABLE AS TABLE である。その機能は、MySQL の benchmark2 ライブラリ配下で sbtest 正規表現に一致するすべてのテーブルを Hudi の DB1 配下の ctas_dema テーブルに同期することである。多対 1 のマッピング関係であり、データベース分割テーブルのマージが行われる。

その後、実行してオンラインにし、ジョブ運用画面に移動してジョブを起動する。設定情報が更新されており、正常にオンラインになったことがわかる。次にスタートボタンをクリックしてジョブを開始する。その後、ジョブ概要画面に移動してジョブ関連のステータス情報を確認できる。

上図はジョブのトポロジーであり、非常に複雑で 1,100 のソーステーブルと 101 のターゲットテーブルがある。ここでいくつかの最適化を行った。ソースマージにより、すべてのテーブルを 1 つのノードに統合し、増分 binlog プルフェーズで 1 回だけプルできるようにすることで、MySQL への負荷を軽減している。

次に OSS ページをリフレッシュすると、cdas_demo パスが追加されているのがわかる。subtest1 パスに入ると、メタデータが既に書き込まれており、データが実際に書き込まれていることがわかる。

その後、ジョブ開発画面に移動して簡単な SQL クエリを作成し、特定のテーブルに問い合わせて、データが実際に書き込まれているかを検証する。上図の SQL ステートメントを実行すると、データが正常にクエリでき、挿入されたデータと一致していることがわかる。

カタログが提供するメタデータ機能と CDAS および CTAS 構文を組み合わせ、数行のシンプルな SQL を通じて、数千のテーブルのデータを容易にレイクに取り込むことができる。これにより、データ取り込みのプロセスが大幅に簡素化され、開発およびメンテナンスのワークロードが軽減される。