Real time Data Lake Flink Hudi Practice and Exploration

Apache Hudi の背景

まず、データレイク開発の歴史的な背景と Hudi の基本的な特徴について共有します。

1. データレイク開発の歴史的な背景

個人的な見解として、Hive などの従来のデータウェアハウスソリューションも実質的にはデータレイクであり、Hudi、Iceberg、Delta Lake については、単なるレイクフォーマットではなく、データウェアハウス向けの次世代ソリューションと位置づけています。では、なぜこの 1 年でデータレイクという新しいデータウェアハウス形態が誕生したのでしょうか。

クラウドストレージ、特にオブジェクトストレージの段階的な成熟を背景に、データレイクソリューションは徐々にクラウドネイティブな方向へ移行しています。図 1 に示すように、レイクフォーマットはクラウドベンダーのオブジェクトストレージに対応し、クラウドベンダーやクラウドファクトリー向けの商用ソリューションとして提供されます。同時に、Spark や Flink などの主要なビッグデータコンピューティングフレームワークに対応し、クエリ側では Presto、Trino、従来の Hive エンジンにも対応しています。このような背景から、新世代のデータウェアハウスソリューションが誕生しました。

2. Hudi の 4 つの中核的特徴

上記からわかるように、Hudi は次世代データウェアハウスソリューションとして、上流のコンピューティングエンジンと下流のクエリエンジンを連携させ、従来の Hive ベースのオフラインデータウェアハウスを置き換える新しいソリューションを実現しています。中核的な特徴は以下の通りです。

• オープン性

オープン性は 2 つの側面に表れています。

1 つ目は、上流側で複数のデータソース形式をサポートしている点です。たとえば、従来のデータベースからの変更ログやメッセージキューログなどの転送方法について、ソース側で非常に豊富なサポートが提供されます。

2 つ目は、下流のクエリ側でも複数のクエリエンジンをサポートしている点です。Presto や主流の OLAP エンジンである StarRocks、Amazon Redshift、クラウドベンダーのデータ分析製品である Impala などが、このデータウェアハウスアーキテクチャに接続されます。

このように、オープン性は Hudi の第一の特徴です。

• 豊富なトランザクションサポート

Hudi のトランザクションサポートは、従来の Hive データベースよりも高度で多機能です。中核的な特徴は、ファイルストレージレイアウトの更新をサポートしている点です。従来の Hive ベースの T+1 更新方式では、データの再現性が比較的高く、日単位のデータ新鮮度しか実現できませんでした。しかし、ビジネス要件の複雑化とリアルタイム性の向上に伴い、データウェアハウスストレージシステムには、分単位や秒単位のエンドツーエンドのデータ新鮮度に対するより高い要件が求められています。

次に、更新効率の向上が必要です。毎回テーブル全体やパーティション全体を上書きするのではなく、ファイル粒度での局所更新により、ストレージとコンピューティングの効率を向上できます。Hudi はこれらの要件を十分に満たしています。したがって、ACID セマンティクスの強化が、このデータウェアハウスアーキテクチャの第二の特徴です。

• ACID セマンティクスに基づく増分処理

第三の亮點は、ACID セマンティクスから派生した増分処理、特に Hudi が提唱する TimeTravel の概念や、Flink や Spark Streaming などのストリーミング処理エンジンとの直接的な連携です。ニアリアルタイムであれ常駐型のストリーミングサービスであれ、本質的にはストリーム消費であり、増分 ETL 処理として理解できます。従来のバッチスケジューリングと比較して、コンピューティング効率がより高くなります。特に、Flink のようなステートフルなコンピューティングフレームワークでは、前回の計算結果を再利用し、エンドツーエンドのフルリンク増分処理を直接実現できます。次に、データ新鮮度は桁違いに向上し、「日レベル」から「分レベル」になります。

たとえば、現在、中国国内の実践的なユーザーの中には、Flink コンピューティングフレームワークを使用してレイクテーブルのストリーム消費を行い、ソース側から注入されたデータを一連の増分 ETL リンクで直接分析し、従来のデータウェアハウス階層を構築する試みを行っているケースがあります。さらに、TimeTravel の増分クエリ設計について、2 つのスナップショット間の増分データをクエリする機能に興味を持つ方も多いでしょう。どのような用途があるのでしょうか。バッチモードでのクエリの場合、主流のシナリオは ADS から下流への同期です。たとえば、データウェアハウスの処理結果を他のデータベーステーブル（Elasticsearch や MySQL など）に同期する場合、TimeTravel を使用して定期的にバッチ同期できます。ADS の同期タイミングにそれほど高い要件がない場合は、この冪等な TimeTravel クエリモードを使用して、より効率的に下流エンドポイントに同期できます。

上記の 3 つのポイントは、主流の Hive アーキテクチャとの 3 つの中核的な違いであり、国内外のデータレイクハウスプロジェクトが追求している方向でもあります。

• インテリジェントなスケジューリング

さらに、Hudi ではファイルレイアウトの最適化を試み、フレームワーク内でスモールファイル管理のデータガバナンススキームを実装し、インテリジェントなスケジューリングを実現しています。これが、Hudi を Delta Lake や Iceberg などの他のデータウェアハウスソリューションと区別する中核的な特徴です。

Flink + Hudi の設計

1. Hudi の書き込みパイプライン（複数のオペレーターで構成されるマイクロサービスアーキテクチャ）

図 2 からわかるように、Hudi の書き込みパイプラインはサーバーレスなマイクロサービスアーキテクチャです。中核は、パイプライン全体がサービス化された後、Flink であれ Spark Streaming であれ、サービス一式がテーブル自体に対して自己管理を実現できる点です。したがって、効率的なデータ書き込みだけでなく、書き込みプロセスでのファイル管理も考慮し、スモールファイルの生成を可能な限り避けてクエリ側の効率を最適化する必要があります。定期的なファイルコンパクションとファイルクリーニングにより、スモールファイルの爆発的な増加を回避できます。

一方、ACID はトランザクション性を備えており、特に更新する ACID トランザクションを完了する際には、多くの要素を考慮する必要があります。単一のジョブやノードでフェイルオーバーが発生した場合、Hudi は以前に書き込まれたエラーデータを迅速に特定し、ロールバックを実施できます。したがって、Hudi のトランザクション層のサポートは、3 つのデータレイクフォーマットの中で最も完全で効率的です。

Copy on Write (CoW) の具体的な実装を例にとると、上流の SQL ネイティブなデータ構造を Hudi のデータ構造に変換します。同時書き込みをサポートするため、各シャッフル後にデータをバケットに分割します。

主に 2 つのポイントがあります。

1 つ目は新規データの追加で、できるだけ既存の空き領域が大きいバケットに書き込みます。同時に、スモールファイルの生成を避けるため、各バケットのサイズが期待値と一致するようにします。

2 つ目はデータの更新です。Hudi はキープライマリキーを設計しており、各キーのメッセージはバケット内で管理され、更新は常に元のバケットに書き込まれます。一方、Iceberg や Delta Lake は書き込みのみでファイルレイアウトを考慮しないため、クエリ側で大量のマージとクリーニングが必要となり、クエリ効率が比較的低くなります。対照的に、Hudi の複雑な書き込みプロセスとバケット戦略は、読み取りと書き込みの効率を天秤にかけて考慮したものです。ここで述べるバケットの概念は、Snowflake のマイクロパーティション概念にやや似ており、従来のパーティションの下でさらに細分化され、ファイル粒度の範囲内でメッセージのライフサイクルを管理します。より細かい粒度でライフサイクルを管理することで、データ更新とクエリの効率を効果的に向上できます。

2 つ目のオペレーターの後、データは各バケットごとにパーティション分割されます。バケット ID に基づいてデータをシャッフルし、書き込みタスクに渡して書き込みを行います。なぜバケット ID で再度シャッフルする必要があるのでしょうか。これは主に、2 つの書き込みタスクが同時に同じバケットの更新セマンティクスを変更できないようにするためです。そうしないと、更新の競合が発生しやすくなります。

したがって、全体として、これら 3 つのオペレーターは同時書き込みと更新を効果的に保証できます。2 つ目のオペレーターの同時実行数が、全体の更新の同時実行数と、同時に更新および書き込み可能なバケット数を決定することは明らかです。一方、後続のオペレーターは独立してスケールアウトできます。実務経験からの推奨として、2 つ目と 3 つ目のオペレーターの同時実行設定を同じにすることが推奨されます。スループットがそれほど高くない場合は、1 つのバケットを 1 つの書き込みタスクに渡して書き込みます。スループットが比較的高い場合は、1 つのバケットに複数の書き込みタスクを割り当てることができ、1:2 の比率に調整できます。

バックグラウンドでは、クリーンコミットのクリーニングタスクも開始されます。データコミット操作はコーディネータコンポーネントで発生します。各書き込みタスクのコミットをチェックポイントとおおむね揃えてからデータをフラッシュします。さらに、一部のメタデータ情報もコーディネータに一律で送信されます。コーディネータは統計情報を収集した後、チェックポイントで完了したイベントと組み合わせてコミットを行います。実際のコミットはコーディネータで実行されます。コーディネータがコミットを完了した後、Hudi テーブルは新しいトランザクションを開始します。書き込みタスクは新しいトランザクションを確認して初めて、新しいトランザクションの書き込みアクションを開始できます。したがって、非同期待機プロセスが存在し、小さな状態機械のようなものです。

ただし、Flink のスナップショットが保証するセマンティクスは、実質的にベストビフォアセマンティクスです。チェックポイント成功イベントを受信すると、以前の状態がすべて成功していることを示しますが、中止されているチェックポイントが存在する可能性があります。

Hudi は各書き込みの整合性と完全な exactly once セマンティクスを保証する必要があるため、中間の書き込みが境界を越えないように考慮する必要があります。たとえば、あるチェックポイントのイベントデータを次のチェックポイントに書き込むことはできず、そうでないと exactly once セマンティクスを保証できません。

バージョン 0.11 では、いくつかの最適化を試みています。たとえば、中止されたチェックポイントを再利用できるかどうかです。状態遷移が伴うため、比較的複雑です。Spark Streaming とは異なり、Spark Streaming は毎回マイクロバッチの抽象化であり、毎回まずタスクを起動するため、自然に exactly once を保証します。フォールトトレランスセマンティクスはフレームワークに委ねられます。Flink がこの非同期アルゴリズムと強力な exactly once セマンティクスをどのように組み合わせるかが、このアーキテクチャの難点です。

2. スモールファイル戦略

次に、ファイル書き込みの 2 つ目のオペレーターである bucket assign の具体的な決定について詳しく見ていきましょう。つまり、新しいメッセージをどのバケットに配置するかを選択する方法です。図 3 に示すように、2 つの方法で説明します。

まず、左側のブロック図には 3 つのバケットがあります。青は現在保存されているファイルのサイズを表しています。insert データの場合、毎回最も空き領域が大きいバケットを選択して書き込むポリシーです。なぜ空き領域が最も小さいバケットを選ばないのでしょうか。CoW の書き込み増幅の問題を考慮する必要があり、効率が比較的低いためです。データの更新時は、まず現在のキーを管理しているバケットを見つけてから書き込みます。これによりファイルサイズが無制限に増加することはありません。更新前後の各レコードのサイズは基本的に似ており、ファイルサイズは大きく変化しません。ファイルサイズは主に insert データの影響を受け、約 120 MB のしきい値が設定されます。

図の右側のブロック図は極端なケースです。2 つのバケットに残っている書き込み領域が少量の場合、書き込み増幅の影響を考慮して、新しいバケットを作成して書き込みます。

同時書き込みのスループットを向上させるため、各 bucket assign タスクに独立したバケット管理ポリシーを割り当て、ハッシュアルゴリズムを使用してバケット ID を各 bucket assign タスクの下に固定ルールでハッシュ化し、同時実行の決定を実現します。したがって、bucket assign タスクの同時実行数を制御することで、書き込まれるスモールファイルの数を相対的に制御でき、書き込みスループットとスモールファイルの間のトレードオフとなります。

3. フル + 増分読み取り

データ書き込みプロセスの説明に続き、データ読み取りのストリーム部分について見ていきましょう。ストリーミング読み取りのフル読み取りと増分読み取りはどのように実現されるのでしょうか。図 4 に示すように、Hudi のタイムラインは各トランザクションコミットのミリ秒単位のタイムスタンプを保存しています。各タイムスタンプはスナップショットバージョンに対応し、メタデータに記録されています。フル読み取り時には、テーブル全体のファイルがスキャンされます。組み込みのメタデータインデックステーブルを設定していない場合は、テーブル全体を直接スキャンしてファイルシステム内のすべてのファイルを検索します。メタデータテーブルが有効な場合は、メタデータテーブル内のファイル情報（KV ストレージ）をスキャンし、比較的効率の高い方法でフルテーブルファイルをスキャンしてから下流に送信します。増分部分は、タイムラインを定期的にモニターし（デフォルトで 60 秒）、新しいコミットがあるかどうかを確認し、同期的に下流に送信して読み取りと書き込みを行います。各増分部分は、最後に発行されたタイムラインポイントに基づいて、最新のコミット時刻を検索します。

Split モニターオペレーターは、増分ファイル情報をモニターし、実際に読み取りを実行するタスクに発行する一連のルールの管理を担当します。

最近のマスターバージョンでは、バッチモードの TimeTravel クエリ（特定の期間のスポットチェック）もサポートしています。以前のバージョンでもサポートされていましたが、いくつかの問題がありました。たとえば、増分メタファイルがアーカイブまたはクリーニングされた場合、データ整合性が保証されませんでした。新バージョンでは、読み取り効率を確保した前提で、2 つのスナップショットとコミット間のバッチモード増分読み取りを実装することで、これらの 2 つの問題を解決し、データ整合性を確保しています。

Hudi のアプリケーションシナリオ

現在、Flink + Hudi は中国国内で非常に人気のある技術アーキテクチャです。ここでは 3 つのアプリケーションシナリオを紹介します。

1. 準リアルタイム DB データウェアハウス / データレイク

このアーキテクチャの DB データウェアハウスの中核的な特徴は、従来の T+1 のデータ新鮮度を分レベルまで向上させることです。データ新鮮度は、Debezium や Maxwell に代表される現在主流の CDC 技術を通じて実現されます。データをストリーミングで準リアルタイムにデータウェアハウスへ同期します。従来の Hive データウェアハウスでリアルタイム性を確保すること、特にファイル更新とレイクテーブルのリアルタイム書き込み更新は非常に困難です。CDC 技術はログウェアハウス自体のストレージに要件を課します。まず、更新効率が十分に高く、ストリーミング書き込みをサポートし、非常に効率的な更新を実現する必要があります。特に、更新プロセス中に CDC ログの順序が入れ替わる可能性もあります。このような順序入れ替え更新の ACID セマンティクスをどのように保証するかは、非常に高い要件です。現在、順序入れ替え更新の要件を満たせるのは Hudi のみです。さらに、Hudi は更新の効率性も兼ね備えており、現在、先進的なアーキテクチャと言えます。

上記の方式と比較して、図 5 の下部に示す方式は、大規模な環境（1 日あたり最大 1 億件の増分）でデータプラットフォームが整っている企業により適しています。中間に統一されたデータ同期方式があり（異なるソーステーブルからのデータを集約し、メッセージキューへ同期）、メッセージキューはデータのフォールトトレランス、ディザスタリカバリ、キャッシュを担当します。同時に、この方式の拡張性も優れています。Kafka のトピックサブスクライブモードを通じて、データを柔軟に配信できます。

2. 準リアルタイム OLAP

2 つ目のシナリオは、準リアルタイム OLAP のシナリオです。分単位のエンドツーエンドのデータ新鮮度と、非常にオープンな OLAP クエリエンジンへの適応が可能です。実際、これは Kappa アーキテクチャや従来のストリーミングデータウェアハウスアーキテクチャに対する新しいソリューションです。このアーキテクチャ以前は、リアルタイム分析では Hudi をスキップして、ClickHouse、Elasticsearch、MongoDB などの OLAP システムにデータを直接二重書き込みしていました。ウェアハウスストレージが効率的な階層更新をサポートし、OLAP エンジンに接続できるようになると、アーキテクチャが大幅に簡素化されます。まず、二重書き込みが不要になります。1 つのデータで唯一の真実のセマンティクスを保証でき、二重書き込みによるデータ整合性の問題を回避できます。次に、レイクフォーマット自体が非常にオープンなため、クエリエンジンの選択肢が広がります。たとえば、Hudi は Presto、Trino、Apache Flink、StarRocks、クラウドベンダーの Amazon Redshift エンジンをサポートしており、非常に柔軟です。

したがって、この準リアルタイム OLAP アーキテクチャは次のように要約できます。① 上流ストレージ側の統一。② 下流クエリ側の開放。

ただし、このアーキテクチャのデータ新鮮度は約 5 分です。秒単位の Kappa アーキテクチャのようなアーキテクチャを実現したい場合、現在の Hudi は適していません。Flink のチェックポイントメカニズム（エンドツーエンドの exactly once セマンティクスをサポート）に依存しているため、頻繁にコミットできません。

3. 準リアルタイム ETL

3 つ目のシナリオは、現在、先進的なアーキテクチャであり、中国国内で徐々に試みられています。データソース自体のデータ量が小さい場合、たとえばソースが Kafka ではなく MySQL の binlog で、QPS が秒間数百程度の場合、このアーキテクチャは非常に安定しており便利なアーキテクチャです。エンドツーエンドの増分処理を実現するだけでなく、中間データウェアハウスのニーズにも対応しています。実際、2 セットの抽象化を提供しています。1 つ目は、データウェアハウスの中間ストレージとしてのストレージ抽象化を請け負い、データをレイク形式で直接保存します。2 つ目は、キュー機能の提供で、Kafka のメッセージキュー機能に似た機能です。ストリーミングモードで増分消費し、その上で一部の増分計算を実行できます。このアーキテクチャは、従来の Lambda アーキテクチャと Kappa アーキテクチャを直接統一します。つまり、Kafka のストレージ抽象化とウェアハウスファイルのストレージ抽象化を 1 つのストレージ抽象化に統合し、ストレージコストを大幅に増やすことなく実現します。後段では、オブジェクトストレージや HDFS などの低コストストレージを使用できます。

このアーキテクチャ全体が 2 つの問題を解決します。1 つ目は二重書き込みの問題です。Lambda アーキテクチャでは、データがまず Kafka に書き込まれてからストレージに保存されるため、2 つのデータの一貫性セマンティクスを保証するのが困難です。さらに、Kafka で exactly once 書き込みを有効にすると、スループットが大幅に低下します。Kafka と HDFS 間のデータの一貫性をどう保つのでしょうか。ストリーミング以外で Kafka を読み取り、別のジョブを起動して Kafka データを HDFS に同期する方法もありますが、その場合、コンピューティングリソースのコスト、メンテナンスジョブ、同期のコストが元の 2 倍になります。

解決する 2 つ目の問題は、M サーバーのクエリ要件です。M サーバーのデータは直接保存され、効率的な方法で更新されません。M サーバーの DWD テーブルで JOIN 操作が必要な場合、エンジンと直接接続でき、Lambda アーキテクチャの T+1 更新効率の問題を考慮する必要はありません。レイク形式の分レベルの適時性により、この問題は大幅に緩和されます。

さらに、このアーキテクチャにはもう 1 つの利点があります。アプリケーションシナリオに応じて豊富な OLAP クエリエンジンを選択でき、データベースストレージに外観の形で直接アクセスできるため、OLAP 分析に非常に便利です。

4. Alibaba Cloud VVP のリアルタイムレイク統合

次に、Alibaba Cloud VVP プロダクトの現在のリアルタイムレイク統合について簡単に説明します。主にレイクエントリーの状態にあります。Alibaba Cloud の組み込み Apache Flink バージョンには Hudi コネクタが内蔵されており、FlinkSQL でレイクエントリータスクを迅速に構築できます。Hudi CDC コネクタや Kafka CDC フォーマットと直接連携し、データを迅速にレイクへ取り込めます。

さらに、レイクエントリープロセスでは、スキーマ進化などの商用機能が提供されています。CE や CTAS 構文はスキーマ進化をサポートしています。同時に、DLF カタログメタデータ管理コンポーネントを推奨しています。DLF カタログは EMR DLF とシームレスに統合されます。EMR で Apache Flink 経由で書き込んだ場合でも、Flink のレイクエントリータスクを書き込み後に管理できます。DLF コンポーネントを通じて、Hudi 形式のデータを EMR クエリエンジンで直接分析できます。

これが現在、商用ユーザーに推進している技術ソリューションです。VVP サービス経由でレイクにエントリーし、EMR 経由で分析した後、VVP は後段でストリームとバッチの統一など、より多くの機能を統合し、ユーザーのストリーム読み取りニーズに応える可能性があります。

最近の Hudi ロードマップ

図 9 に示すように、Hudi 0.12 と 1.0 で近いうちに実装予定の機能を簡単に紹介します。

まず、Delta 2.0 と同様の CDC フィード機能を導入します。現在サポートされている CDC は、入力が CDC 形式である必要があり、Hudi も CDC 形式で保存します。CDC フィード機能との違いは、全体の入力が CDC 形式であることを保証する必要がない点です。たとえ UPSERT セマンティクスがあったり、CDC の中間データが欠落していたりしても、CDC を完全にプライマリエンドに復元できます。この機能は、読み取りと書き込みのスループットとリソースの間に一定のトレードオフがあり、現在の CDC 処理アーキテクチャほど効率的ではありません。

2 つ目はメタサービスです。メタデータ管理をプラグイン化し、Hudi 上のテーブルとタスクを統一されたメタサービスプラグイン形式で管理します。

3 つ目は、セカンダリインデックスの構築を計画しています。現在のマスターバージョンでは、Apache Flink と Apache Spark がデータスキッピング機能を実現しています（書き込み時に、ユーザーがメタデータテーブルを有効にし、同時にデータスキッピングを有効にすると、各列の統計が追加で記録されます）。最も典型的なのは、各列に Max と Min を作成してメタデータによるアクセラレーションを有効にし、ファイルレベルのクエリ効率を向上させることです。将来的には、データベースに似たセカンダリインデックスをサポートします。特定の列に対して LSM に似た抽象化インデックスを実装し、ポイントクエリシナリオに適した効率的なインデックススキームを構築します。

最後に、特徴量エンジニアリングに似た列ごとの更新機能を開発します。これは ClickHouse の Merge Tree 抽象化に似ており、列を独立して保存します。機械学習の特徴量エンジニアリングでは、大量の特徴量が数千のフィールドを必要とし、特徴量が生成されるたびに列を更新する必要があるため、単一の列に効率的な更新能力が求められます。このようなシーンに対応するため、Hudi は引き続き探求を続けます。

質疑応答

Q1: Hudi を直接使用して更新を保存するのと、CDC を直接使用して StarRocks に送信するのとではどちらが良いですか。StarRocks の QPS は Hudi の更新速度よりも速いと思います。

A1: 確かにその通りです。StarRocks は LSM に似た効率的なプライマリキーインデックスを使用し、メモリ内でパーティションポリシーを使用してより多くのセカンダリインデックスとメタデータ情報を維持しています。さらに、最も重要な点は、メインテーブルへの書き込みと更新時にバッチ保存操作を使用することです。まず複数の書き込みをバッファーにインポートしてからフラッシュし、データフラッシュ時にもバッチ処理を行えます。これが、StarRocks の更新効率が高い理由です。

ただし、同時に、サーバークラスターであるため、2 つの問題が生じます。1 つ目は、高い運用コストがかかることです。2 つ目は、Hudi のサーバーレス形式と比較して、メモリモードではより多くのオーバーヘッドが発生することです。

Hudi 形式は StarRocks に対してオープン性で一定の優位性があります。StarRocks だけでなく、Presto や Apache Flink などの主要な OLAP エンジンとも連携できます。

これが両者の違いです。2 つの方式は重点が異なり、実際のアプリケーションシナリオに応じて選択する必要があります。StarRocks は OLAP アプリケーションにより適しています。ただし、データウェアハウスを構築する場合、StarRocks を Hudi の代わりに使用するコストは高すぎるでしょう。Hudi のシナリオは主に時系列データであり、従来のデータ系列には Hive の方が有利です。

Q2: フローデータがレイクにエントリーする場合、MoR (Merge on Read) テーブルと CoW (Copy on Write) テーブルのどちらが適切ですか。

A2: フローデータの量が多い場合は、MoR テーブルを推奨します。現在の計測方式では、QPS が 20000 を超えない限り、CoW テーブルでも対応可能です。20000 を超えた場合は、MoR テーブルを推奨します。QPS がさらに高い場合、コンパクションタスクを分離する必要があるかもしれません。これが現在提供可能なソリューションです。