Flink Kylin Hudi Lake warehouse integrated big data ecosystem

1. レイクとウェアハウスの統合アーキテクチャ

データレイクとデータウェアハウス

湖倉一体について語るにあたり、まずレイクとウェアハウスがそれぞれ何かを理解しましょう。データレイクは非常に歴史のある概念で、近年になって再び注目されています。現在に至るまで、業界にはデータレイクの統一された定義が存在しません。AWS はクラウド上でデータレイクソリューションを初めて提供したクラウドサービスプロバイダーです。ここでは AWS のデータレイクの定義を参照します。「データレイクは、任意の構造のデータを保存し、ビッグデータ処理、リアルタイム分析、機械学習などの関連アプリケーションシナリオに適用できる中央リポジトリです。」同様に、AWS のデータウェアハウスの定義も使用します。「データウェアハウスは情報の中央リポジトリです。」ここでの情報は分析可能であり、より情報に基づいた意思決定を行うことができます。

この定義にはさらに詳細な展開があります。AWS は、レイクからウェアハウスへのデータフローの関係を示すことで、データレイクとデータウェアハウスの違いと関連性を説明しています。まず、データは元々データレイクまたはデータベースに存在し、データスクリーニングと準備を経て、高価値な分析のためにデータウェアハウスに流れ込みます。この比較表は、データ、スキーマ、コストパフォーマンス、Data Quality、ユーザー、分析という 6 つの観点からデータレイクとウェアハウスを直感的に比較しています。

湖倉一体の前例

今年、Alibaba が提唱する「湖倉一体」という概念を耳にしました。業界に湖倉一体の成功例があるかどうか、考えたことはあるでしょうか。個人的にはあると思います。2020 年 9 月、Snowflake という会社がニューヨーク証券取引所に上場しました。Snowflake はクラウドベンダーが提供するインフラ上に SaaS プラットフォームを構築し、中小企業向けにデータホスティングと分析サービスを提供するクラウドデータウェアハウス企業です。Snowflake は自らをクラウドデータウェアハウス企業と位置づけ、2016 年のデータサミットでエラスティックデータウェアハウスアーキテクチャとその技術的な詳細を紹介する論文を発表しました。

Snowflake は実際、クラウド上のオブジェクトストレージを基盤とするアーキテクチャであり、1 つのコピーで複数のコンピューティングを保存し、コンピューティングとストレージを分離しています。これはまさに AWS や現在の主要なクラウドベンダーが推進しているデータレイクのアーキテクチャです。Snowflake の上場初日、時価総額は 700 億ドルに急騰しました。個人的には、Snowflake は湖倉一体を実装した最も成功した前例と考えられます。先ほど紹介した論文について詳しく調べてみてください。以下の 5 つのポイントを選んで簡単に共有します。

まず 1 つ目のポイントは、従来のデータウェアハウスで広く採用されている Shared Nothing アーキテクチャを採用せず、Shared Data アーキテクチャへ移行したことです。

2 つ目は、ストレージとコンピューティングの分離です。論文で強調されている点であり、最も価値のある観点だと思います。統一ストレージの概念を提唱し、その後にエラスティックコンピューティングが続くという構成です。

3 つ目は、データウェアハウジングとサービスです。商業化において最も成功した点だと思います。データウェアハウスに対して SaaS のような体験を提供し、データウェアハウスは大規模で重厚であるという従来のバイアスを打ち破りました。

4 つ目は、高可用性です。ユーザーエクスペリエンスとフォールトトレランスの向上において重要な側面です。

最後に、構造化データから半構造化データへの拡張です。当時既に、レイク側の汎用データを探求する能力を実証していました。

これは 2016 年の論文ですが、その概念は古くなることなく、今なお学ぶ価値があります。次は、私たちが取り入れて実践する予定のポイントをいくつか紹介します。これらは T3 出行が湖倉一体を実現する上で重要な要素です。

Shared Nothing アーキテクチャの利点

まず、従来のデータウェアハウスで広く採用されているアーキテクチャとして、Shared Nothing にはいくつかの構造的な利点があります。

1 つ目は、テーブル上のデータをノード間で水平方向にパーティショニングでき、各ノードが独自のローカルストレージを持つことができる点です。各ノードのコンピューティングリソースは、そのノード自体に保存されたデータの処理に集中できます。

もう 1 つの利点は、処理メカニズムが比較的シンプルであることであり、データウェアハウス分野における典型的なアーキテクチャです。

Shared Nothing アーキテクチャの欠点

このアーキテクチャにはいくつかの欠点があります。

最も大きな問題は、コンピューティングとストレージのリソースが結合されていることです。

同時に、2 つ目の問題として拡張性の不足があります。これは 2 つの側面で現れます。

a. クラスターの拡張または縮小時に、データの大幅な再配置が必要になる

b. 未使用のコンピューティングリソースを簡単にアンインストールできない

3 つ目の問題は、結合されたコンピューティングとストレージのリソースが可用性の制約をもたらすことです。いわゆるステートフルな計算により、障害やアップグレード時にパフォーマンスが大幅に影響を受け、サービス全体の可用性が損なわれます。

最後に、同質リソースと異質ワークロードの問題があります。データウェアハウスのシナリオでは、バッチロード、クエリ、大規模なレポートの計算と分析など、多くの異質ワークロードが存在します。しかし、Shared Nothing アーキテクチャのリソースは同質であるため、この 2 つの間に矛盾が生じます。

Shared Data アーキテクチャ

これらの問題に基づいて、Snowflake は Multi Cluster Shared Data と呼ばれるアーキテクチャを提案しました。ここでは公式のイメージを簡単に調整しています。

このアーキテクチャの 1 つ目の利点は、データサイロが存在せず、統合ストレージであることです。これによりストレージとコンピューティングのデカップリングも実現します。

2 つ目の利点は、現在のオブジェクトストレージを基盤として、構造化データと非構造化データの両方に対応できることです。

3 つ目は、クラスターサイズを柔軟にスケールできることです。

4 つ目は、上記の特徴によりオンデマンドコンピューティングの低コスト化が実現できることです。

次に、このアーキテクチャを階層的に確認します。全体として、その構造はおおよそ 3 つのレベルに分類できます。

最下層はクラウドベンダーが提供するオブジェクトストレージ、すなわちユーザーストレージです。

中間層は多目的コンピューティングクラスターです。

データレイクの管理サービスであり、大規模な SaaS プラットフォームが基盤ストレージとコンピューティングクラスター全体の管理を担っています。

Shared Data の継続的な高可用性

次のポイントはこのアーキテクチャの高可用性です。これは大きく 2 つの側面に分解できます。1 つ目はフォールトトレランス、2 つ目はオンラインアップグレードです。

まず、SaaS ベースのアプリケーションとして、そのフォールトトレランスはアーキテクチャ全体に反映される必要があります。ここでも階層ごとに確認します。

最下層のストレージレイヤーはクラウドベンダーのオブジェクトストレージ機能を活用しており、データセンター間でのレプリケーションとほぼ無限のスケーラビリティを実現するメカニズムであるため、ユーザーは基本的に気にする必要がありません。

その上は多様なコンピューティングクラスターです。各コンピューティングクラスターは同じデータセンター内に配置され、ネットワーク伝送パフォーマンスを確保しています。ここで 1 つの課題として、あるコンピューティングクラスターでノード障害が発生する可能性があります。クエリ中にノードが障害を起こした場合、これらのコンピューティングノードは上位のサービス層にそのステータスを返します。サービス層はこの障害を受け取ると、計算を再利用可能なノードに転送してクエリを再実行します。つまり、ストレージとコンピューティングが分離された Shared Data アーキテクチャでは、ノードは計算においてほぼステートレスです。このアーキテクチャでは 1 台のノード障害は大きな問題ではありません。

さらに、サービス層はオブジェクトストレージの機能をメタデータ保存にも活用しています。したがって、このサービス層は実質的にステートレスなサービスと見なせます。

最上位層はロードバランサーであり、サービスの冗長性と負荷分散を実現します。

2 つ目のポイントは、オンラインアップグレードは主に 2 つの設計を活用していますが、これは特に新しい手法ではありません。1 つはコンピューティング層とサービス層の間に複数のマッピングを作成し、カナリアリリースを切り替える方法です。コンピューティング層には複数のバージョンが存在し、これらのバージョン間でローカルキャッシュが共有されています。サービス層のメタデータ管理も複数のマッピング間で共有されています。これはアーキテクチャ内のサブ Shared Data であり、複数バージョン間でのデータ共有とスムーズなカナリアリリースによるアップグレード機能を持っています。

次に、同僚の Wang Xianghu がこれら 3 つのフレームワークの統合方法と、最終的に T3 の湖倉一体をどのようにサポートするかについて説明します。2 つ目のトピックに入る前に、まずメインフレームワークである Hudi と Kylin を紹介します。そして、これら 3 つのフレームワークがどのようにペアで統合されるかを説明し、最後に T3 出行が湖倉一体システムをどのように構築するかを説明します。

2. Flink/Hudi/Kylin の導入と統合

Hudi

まず Hudi について理解しましょう。Hudi は元々 Uber のエンジニアがデータ分析のニーズを満たすために設計・開発したデータレイクフレームワークです。2019 年 1 月に Apache インキュベーターに参加し、2020 年 5 月に卒業して Apache トップレベルプロジェクトになりました。Hudi の名前は Hadoop Upserts Delete and Increments の略称です。つまり、Hudi はデータの挿入、更新、削除、および増分処理をサポートするデータレイクフレームワークです。さらに、ACID トランザクションに基づく増分処理、ストレージ管理、タイムライン管理もサポートしています。Hudi はクラウド上の 100 ペタバイト級の大規模分析データセットを管理でき、各種クラウドサービスとすぐに連携できるため非常に便利です。また、Uber 社内では約 4 年間安定的に稼働しています。

下図は Hudi のプラグイン型アーキテクチャを示しています。Hudi はストレージ、データ処理エンジン、テーブルタイプ、インデックスタイプ、クエリビュー、クエリエンジンなど、各コンポーネントに対して比較的疎結合なサポートを提供しています。つまり、特定のコンポーネントに縛られることはありません。

ストレージ面では、Hudi は HDFS、OSS、S3 をサポートします。

データ処理エンジンに関しては、Hudi は Flink と Spark をサポートしています。コミュニティでは既に Java クライアントと Python クライアントがサポートされています。Hudi は COW テーブルと MOR テーブルの 2 種類のテーブルタイプをサポートしており、それぞれ低レイテンシのクエリシナリオと高速なデータインジェストシナリオに対応しています。

インデックスに関しては、Hudi は Bloom や HBase を含む 4 種類のインデックスをサポートしています。基盤層では Parquet と Avro を使用してデータを保存しており、コミュニティでは ORC フォーマットと SQL のサポートも進められています。近い将来実現する見込みです。

Hudi は読み取り最適化クエリ、増分クエリ、スナップショットクエリの 3 種類のクエリをサポートしています。クエリエンジンとしては Spark、Presto、Hive、Impala があり、他のいくつかのコンポーネントもサポートされています。

以下はストレージモードとビューの詳細な説明です。

まず Copy On Write モードです。これは Hudi の COW テーブルに対応し、低レイテンシのデータクエリシナリオに重点を置いたテーブルで、基盤層では Parquet データファイルを使用してデータを保存しています。スナップショットクエリと増分クエリの 2 種類のクエリ方法をサポートしています。クエリエンジンに関しては、5 つのエンジンがスナップショットクエリ、増分クエリ、読み取り最適化ビューに対してさまざまなレベルでサポートしています。

Merge On Read テーブルは Copy On Write をさまざまなレベルで補完するもので、高速なデータインジェストシナリオに重点を置いています。Parquet ファイルを使用して実際のデータを保存し、行ベースの Avro 増分ファイルを使用して操作ログを保存します。これは HBase の Write Ahead Log に似た仕組みです。Hudi の 3 つのビューすべてをサポートしており、Hive、Spark SQL、Spark Datasource、Presto、Impala は読み取り最適化クエリをサポートしています。Hive と Spark SQL はスナップショットクエリのみをサポートしています。各コンポーネントの対応状況は今後公式サイトで確認できます。

T3 出行のビジネスでは、注文にロングテール決済という属性があります。つまり、注文が発生した後、決済プロセスが長い期間遅延する可能性があります。言い換えれば、ユーザーの次の旅行の前に決済が行われるか、それ以上かかるか、あるいは決済が行われないこともあります。このロングテール属性により、ビジネスのクローズドループウィンドウが過度に長くなり、データ更新のタイミングを正確に予測できなくなります。多段階の更新がある場合、リンクがさらに長くなり、更新コストが非常に高くなります。

下図はロングテール更新によるコールドデータの頻繁な更新を示しています。左側はビジネスデータベースで、右側は依存関係を持つ 3 つの概略テーブルです。ビジネスデータベースのデータが更新されると、右側の更新が必要なデータは既にパフォーマンスの低いデバイスにアーカイブされている可能性があり、データ更新コストが増加します。さらに、このデータ更新がロングチェーンのカスケード更新を引き起こすと、この遅い I/O がさらに増幅されます。

データの信頼性もデータ ETL において避けられない課題です。マシンの故障や計算ロジックの問題により、処理データが歪んだり完全に不正確になったりする可能性があり、ビジネス判断に大きな影響を及ぼします。データレイテンシの面では、Hive ベースの従来のアーキテクチャでは、Hive にインデックスメカニズムがないため、データ更新はデータパーティション全体の書き換えを引き起こし、その場での削除ができません。また、小ファイルの問題は NameNode のストレージとクエリの負荷を増加させ、処理を遅延させ、ある程度データ遅延を増加させます。

Kylin フレームワーク

次に Kylin フレームワークについて説明します。Hudi と比べて、Kylin は比較的よく知られているでしょう。これはオープンソースの分散分析データウェアハウスで、Hadoop や Spark SQL 上でデータクエリウィンドウを提供できます。当初 eBay によって開発され、オープンソースコミュニティに貢献されました。巨大なテーブルをサブ秒でクエリできるのが特徴です。その仕組みは事前計算にあります。スタートポロジーデータキューブのメトリクスの多次元組み合わせを計算し、結果を出力テーブルに書き出してクエリインターフェイスを公開することでリアルタイムクエリを実現しています。つまり、スペースとアクセスタイムを交換するアプローチです。

Kylin は今年の 9 月にバージョン 4.0 alpha をリリースしました。これは Kylin 3 以降の主要なアーキテクチャアップグレードです。Parquet を使用して HBase ストレージを置き換えることで、ファイルスキャンパフォーマンスを向上させ、HBase のメンテナンス負荷を削減または完全に排除しました。Kylin 4 は Spark のビルドエンジンとクエリエンジンを再設計し、コンピューティングとストレージを分離することで、クラウドネイティブな技術トレンドに適した設計になっています。

Flink/Hudi/Kylin フレームワーク間の融合

Kylin 3.1 のリリースにより、Kylin と Flink の統合が完了しました。この機能は 2019 年に完成し、Kylin と Flink の統合作業は昨年の 1 月に Flink Batch を通じて開始されました。Hudi との融合については、Kylin と Hudi は本質的に互換性が高いと言えます。Hudi は自身を Hive テーブルとして公開でき、ユーザーは Hive と同様に Hudi のデータを使用できるため、Kylin にとって非常にフレンドリです。Kylin は Hudi を使用したデータをシームレスに Hive テーブルとして扱えます。Hudi と Flink の統合は、今年のコミュニティへの主な貢献です。これら 2 つのスクリーンショットは、Hudi と Flink の統合に関する 2 つのマイルストーン PR に対応しています。

1 つ目は、Hudi クライアントが複数エンジンをサポートし、Hudi を Spark からデカップリングすることで、Hudi が複数エンジンをサポートすることを可能にしました。

2 つ目は、Flink クライアントの基本的な実装をコミュニティに貢献し、Hudi が Flink データテーブルに実際に書き込めるようにしました。これら 2 つの変更は合わせて 10,000 行以上のコードに達し、今年の Hudi コミュニティの顕著な特徴となっています。

Hudi と Flink の融合プロセス

以下は Hudi と Flink の融合プロセスの詳細な説明です。Hudi は元々 Spark エンジンのみをサポートしていたため、最初のステップは Hudi を Spark からデカップリングし、その後目的のエンジンを統合することでした。

デカップリングの難しさは、Hudi が当初複数エンジンのサポートを考慮していなかったため、データソースからの読み取りから Hudi テーブルへの書き出しまで RDD が広く使用されていた点にあります。通常のユーティリティクラスでも RDD を基本操作単位として使用していました。Spark からのデカップリングを評価したところ、変更が非常に大きいことが分かりました。次に、Flink と Spark のコア抽象化には違いがあります。Spark はデータを有限データセットとして扱うのに対し、Flink はデータを有界でないデータストリームとして扱います。この抽象化の違いにより、汎用抽象化の統一が困難になります。

この変更は Hudi にとって大きな挑戦であるため、元の Hudi の機能とパフォーマンスを維持することを優先し、Flink Stream API の一部を犠牲にすることにしました。Flink はリストを操作し、Spark は RDD を操作することで、ジェネリックを抽出して統一抽象レイヤーを形成できます。

抽象化の原則：

ジェネリック I、K、O を統一使用して置き換えます。

Spark からのデカップリングのため、抽象レイヤー API はエンジン非依存であり、抽象レイヤーでの実装が難しい部分は抽象メソッドに変更して Spark サブクラスの実装に委任します。

元のバージョンに影響を与えないよう、抽象レイヤーへの変更を可能な限り最小限に抑えて機能を固定します。

JavaSparkContext の代わりに HoodieEngineContext を導入してランタイムコンテキストを提供します。

次に、Flink クライアント DAG について説明します。主に 5 つの部分で構成されています。

1 つ目の部分は Kafka ストリーミングソースで、主に Kafka データを受信してリストに変換するために使用されます。

2 つ目は InstantGeneratorOperator で、グローバルにユニークなインスタントを生成する Flink オペレーターです。

3 つ目は KeyBy パーティション操作で、partitionPath に基づいてデータをパーティションに分割し、複数のサブタスクが同じパーティションにデータを書き込む際の競合を回避します。

4 つ目は WriteProcessOperator で、これもカスタムオペレーターです。このオペレーターで実際の書き込み操作が行われます。

5 つ目は CommitSink で、上流の WriteProcessOperator から送信されたデータを受け取り、上流のデータに基づいてトランザクションをコミットするかどうかを判断します。

以下は Flink の更新に関するコード例です。左側は元のバージョンから簡略化された HoodieWriteClient です。

insert 関数の入力パラメーターが RDD で戻り値も RDD であることが分かります。右側の抽象化後は、入力パラメーターがジェネリック I に、戻り値がジェネリック O に変更されています。興味のある方は詳しく調べてみてください。

もう 1 つの Flink 統合のアイデアがあります。ストリーミングソースを作成し、Flink を使用して完全な ETL パイプラインを構築する方法です。Hudi テーブルからデータを読み取り、再度 Hudi テーブルに書き出します。

次に初期設計のアイデアです。左側のグレーの画像には Hudi メタデータの 5 つの列が含まれています。最も左側は Hoodie_commit_time のトランザクションリストです。各 Hoodie_commit_time は 1 つのトランザクションに対応し、各トランザクションは 1 バッチのデータに対応します。各バッチのデータの各 Record にはコミットシリアル番号が割り当てられ、それが 2 列目の Hoodie_commit_seqno です。Hoodie_commit_time と Hoodie_commit_seqno のマッピング関係は、Kafka のパーティションとオフセットのマッピング関係に非常に似ています。将来的にはこの特徴に基づいて Hoodie Streaming Source を実装できる可能性があります。

これら 3 つのフレームワークの融合関係に基づいて、計算、分析、ストレージに使用される 3 つのエンジンが相互に互換性を持つことが分かりました。そして、湖倉一体をサポートし、クラウドネイティブなシステムへ近づいています。

3. T3 出行の湖倉一体構築の実践

最後に、T3 出行がどのように湖倉一体を構築するかを見ていきましょう。これは T3 出行の車聯網システムアーキテクチャで、基盤サポートから上位レベルまで継続的に力を与え、自動車企業や国家情報プラットフォームのシステムと連携しています。車聯網駆動型の交通企業として、人、車、道路などに関連するデータを収集しています。各データタイプには独自のアプリケーションシナリオがあり、データは孤立せず相互に力を与え合い、T3 スマート交通を共同で支えています。

これはストレージとコンピューティングを分離したデータベースアーキテクチャです。アーキテクチャ全体はコンピューティング層とストレージ層の 2 つの層に分かれています。

コンピューティング層では Flink、Spark、Kylin、Presto を使用し、Elasticsearch (ES) と連携してタスクスケジューリングに使用しました。データ分析と表示には Da Vinci と Zeppelin を使用しました。

ストレージ層では Alibaba Cloud OSS を使用し、HDFS と組み合わせてデータストレージを実現しました。データ形式としては、Hudi を主要ストレージフォーマットとして使用し、Parquet、ORC、Json ファイルと組み合わせています。コンピューティングとストレージの前に Alluxio を追加し、データ処理のパフォーマンスを加速・向上させています。リソース管理には Yarn を使用しており、時期が来れば K8s に移行する予定です。

ストレージとコンピューティングの分離という現在のトレンドの中で、レイクストレージを中心に、レイクアクセラレーション、レイクコンピューティング、OLAP 分析、インタラクティブクエリ、可視化などを含む完全なビッグデータエコシステムを構築しています。

T3 における Hudi のアプリケーションシナリオ

以下は T3 出行内での Hudi のアプリケーションシナリオです。

1 つはニアリアルタイムストリーミングデータパイプラインです。Log、MySQL、Kafka などからビジネスデータを左側のデータパイプラインにインポートし、Flink または既存の DeltaStreamer を使用してストリーミングデータを Hudi テーブルに取り込みます。

ニアリアルタイムストリーミングデータ処理の Flink UI には、source や instant_generator など、前述した DAG オペレーターがいくつか含まれていることが確認できます。

もう 1 つのシナリオはニアリアルタイムデータ分析です。Hive、Spark、Presto を使用してデータをクエリし、最終的に Da Vinci や Zeppelin を使用してレポートを作成します。

これは Hudi を使用して構築した増分データパイプラインです。左端の CDC データをキャプチャした後、後続の一連のテーブルを更新する必要があります。Hudi の導入後は、Hudi がインデックスと増分データ処理をサポートしているため、更新が必要なデータのみを更新すればよく、以前のようにパーティション全体やテーブル全体を更新する必要はありません。

最後のシナリオは、Flink を使用してオンラインデータやビジネスデータを Hudi テーブル内の ETL にサブスクライブし、機械学習に活用する方法です。機械学習にはデータ基盤が必要なため、Hudi を使用してオンラインデータをオフライン環境に増分公開し、モデルトレーニングやパラメータチューニングを行います。その後、トレーニングしたモデルをオンラインに公開して、ビジネスにサービスを提供します。