Alibaba Cloud EMR StarRocks Extreme Data Lake Analysis

より多くのユーザーの迅速なデータ分析ニーズに応え、StarRocks の強力な分析機能をより幅広いデータセットに適用するため、Alibaba Cloud E-MapReduce (EMR) OLAP チームと StarRocks コミュニティは 2021 年に協力を開始しました。

双方は共同で StarRocks のデータレイク分析機能を強化し、StarRocks にローカルに保存されたデータだけでなく、Apache Hive (以下 Hive)、Apache Iceberg (以下 Iceberg)、Apache Hudi (以下 Hudi) などのオープンソースデータレイクやデータウェアハウスに保存されたデータに対しても、同等の優れたパフォーマンスで分析を実行できるようにしました。

Alibaba Cloud EMR StarRocks は、StarRocks が Alibaba Cloud にライセンスを付与したオープンソース OLAP プロダクトであり、エンタープライズユーザーの多様なデータ分析シナリオに対応するため、高速で統一された分析体験の構築に取り組んでいます。本記事では、Alibaba Cloud EMR StarRocks がデータレイク分野で行った取り組み、その実際の成果、およびデータレイク分析分野における StarRocks の今後の計画について主に解説します。

Alibaba Cloud EMR StarRocks 全体アーキテクチャ

ストレージレイヤーでは、Alibaba Cloud のオブジェクトストレージ OSS をデータレイクの統一ストレージとして使用し、Parquet / ORC / CSV などの一般的なファイル形式を保存できます。

レイク管理および最適化レベルでは、EMR は Data Lake Formation (DLF) を通じて、データレイク全体のメタデータ管理と統合構築を行います。同時に、データレイク分析の実践において、HDFS は従来の Apache Hadoop (以下 Hadoop) と比較していくつかのパフォーマンス上の課題を抱えています。この課題を解決するため、Alibaba Cloud EMR では Jindo FS システムを開発し、データレイクストレージレイヤーへのアクセスを高速化および最適化しました。

同時に、Parquet や ORC 形式などの一般的なデータレイクストレージ形式を対象として、Hudi や Iceberg では、インデックス統計のバージョン情報、バージョン管理、小ファイル統合、ライフサイクルの面で最適化と機能強化が行われています。ストレージとデータベース管理の最適化を組み合わせることで、分析レイヤー、すなわちデータ開発およびガバナンスレイヤーを構築できます。

データ開発およびガバナンスレベルでは、StarRocks は Alibaba Cloud EMR において固定ノードとエラスティックノードの 2 つのロールに分かれています。StarRocks データレイク分析エンジンにより、EMR 上のオープンソースである Apache Airflow (以下 Airflow) や Jupyter、および Alibaba Cloud の DataWorks を接続して、データ開発とスケジューリングを行えます。

Iceberg における StarRocks の実装

StarRocks は主に FE と BE コンポーネントで構成されています。これらは RPC を通じて相互に通信し、クエリのスケジューリングと配布、結果の集約など一連のタスクを実行します。

Iceberg のデータレイク分析をサポートするため、FE 側と BE 側の両方で多くの改修を行いました。まず FE 側では、外部テーブルタイプ IcebergTable を追加しました。実行計画の生成後、RPC プロトコル (Thrift プロトコル) を修正して実行計画の関連情報を BE に送信します。BE 側では、HDFS スキャナーを使用して実際のデータスキャンを実行します。

上記の一連の研究開発作業を完了した後、TPC-H ベンチマークと Trino をベースにしたパフォーマンス比較テストを実施しました。StarRocks は Trino と比較して非常に優れたパフォーマンスを発揮していることが確認できます。

では、StarRocks はなぜ Trino よりもはるかに優れたパフォーマンスを実現できるのでしょうか。

StarRocks のパフォーマンス分析

StarRocks の既存の包括的なベクトル化実行エンジンと新しい CBO オプティマイザーにより、シングルテーブルおよびマルチテーブルの両レベルでパフォーマンスを大幅に向上できます。これに加えて、データレイク分析シナリオ向けの新しい最適化ルールも追加しました。

まず、最適化ルールの面でいくつか簡単な例を挙げます。たとえば、一般的なプレディケートプッシュダウンがあります。プレディケートプッシュダウンをサポートすることで、col_a > X などの条件をデータソース側にプッシュダウンでき、データスキャン時のスキャンデータ量を削減できます。

プレディケートがプッシュダウンされない場合 (上記の図の左上に示す通り)、まずデータ全体をスキャンしてから、エンジン自体の上流にあるフィルターオペレーターでフィルタリングを行います。これにより多大な IO オーバーヘッドが発生します。

スキャンデータ量をさらに削減するため、パーティションクリッピングもサポートしています (上記の図の中央に示す通り)。最適化前は 3 つのパーティションをスキャンする必要がありましたが、パーティションクリッピングの最適化により、不要な 2 つのパーティションを FE 側で除外できます。BE には残りのパーティションのデータのみをスキャンするよう指示します。BE では、グローバルランタイムフィルターもサポートしており、Join シナリオのパフォーマンスを大幅に向上できます。StarRocks の優れた実行エンジンと組み合わせることで、CPU 集約型のデータレイク分析シナリオで高いパフォーマンスを発揮します。ただし、実際のシナリオの実装において、FE 側の最適化ルールや前述のグローバルランタイムフィルターだけでは、IO オーバーヘッドを完全に削減できない場合があります。

IO オーバーヘッドの削減方法が重要になります。ほとんどの場合、データレイクで分析対象となるデータとコンピューティングノードは同じ物理マシン上に存在しません。分析プロセスにおいて、非常に大きなネットワーク IO の課題に直面しています。このため、StarRocks コミュニティは、遅延マテリアライゼーション、IO 統合、ネイティブ Parquet / ORC リーダーのサポート、オブジェクトストレージ向け SDK の最適化など、IO に関する多くの最適化を実施しました。

次に、2 つの例を通じて実際の最適化の詳細を紹介します。

IO 統合

IO 統合以前は、Parquet ファイルに関連するデータを読み取る場合、まず FE 側から BE に送信されたスキャンデータパスに基づいて、ファイルレベルのファイルリーダーを構築する必要がありました。FE 側でのプランニング時に、実際にどの列のデータをスキャンするかも指定できます。実際の顧客導入プロセスにおいて、小ファイルが IO 消費の増大を招いていました。

ColumnReader について、ある SQL ステートメントが同時に 3 つの列を読み取る必要があるとします。2 つの列のデータ量が比較的小さい場合があります。このとき、これら 2 つの列の IO を統合できます。たとえば、以前は 2 回のネットワーク IO が必要だったデータを 1 回で読み取れるようになります。Row Group についても、小さい Row Group に対して IO 統合を行い、IO 回数を削減できます。

ファイル自体が非常に小さい場合は、ファイルを一度にメモリへロードすることもサポートしています。実際、テストプロセスにおいて、小規模な IO が大量に発生するシナリオで明確なパフォーマンス向上が確認されています。

遅延マテリアライゼーション

遅延マテリアライゼーションとは何でしょうか。遅延マテリアライゼーションはどのような課題を解決する必要があるのでしょうか。

遅延マテリアライゼーション以前、Parquet の実装原理に戻ります。たとえば、3 つの列を読み取りたい場合、これら 3 つの列を同時に読み取り、いくつかのプレディケートを適用してから、上流のオペレーターに結果を返します。ここで明らかな課題は、3 番目の列にプレディケートがない場合、3 番目の列のデータをすべて読み取る必要はないということです。

図の左部分を見てください。SQL に最初の 2 列 c0 と c1 に対するプレディケートがあるため、まずこれら 2 列のデータをメモリに読み込みます。そして、これら 2 列に基づいてセレクションマスクを構築します。これら 2 つのマスクはタグ配列と呼ばれます。これら 2 つのタグ配列を用いて、3 番目の列は Lazy 列として定義されます。

最初の 2 列のタグ配列を取得した後、これら 2 つのタグ配列に基づいて新しいフィルタータグ配列を構築します。そして、新しいフィルタータグ配列に基づいて Lazy 列を読み取ります。実際の使用では、Lazy 列には複数の列が含まれる可能性があり、不要な IO 読み取りを大幅に削減できます。包括的なベクトル化、CBO オプティマイザー、データレイク分析向けの IO 自体の最適化を含む、エンジンの既存の機能強化により、テストおよび実際の実装プロセスで優れたパフォーマンスを達成しています。

実践において、もう 1 つの課題はメタデータアクセスです。データレイクシナリオでは、ファイルのリスト操作がネットワークアクセス全体のボトルネックになる場合があります。この課題を解決するため、StarRocks の FE 側に、Hive のパーティション情報とファイル情報をキャッシュできる細粒度のインテリジェントキャッシュスキームを完全セットで設計しました。

キャッシュの設計において、キャッシュの更新は大きな課題です。イベント駆動モードをベースにすることで、キャッシュ更新の課題を解決し、ユーザーのクエリパフォーマンスを確保すると同時に、手動でのキャッシュ更新なしに非常に優れた使用体験を提供できます。同時に、クエリのプランニングとスケジューリングを高速化するため、統計情報のキャッシュもサポートしています。

StarRocks のエコロジー分析

初期バージョンでは、新しいデータソースをサポートする場合、多くの冗長な開発が必要でした。開発者は多くの他のモジュールについて深い理解を持つ必要があり、使用時に外部テーブルを作成する必要もありました。この課題をどう解決するか。私たちのソリューションは、新しいコネクタフレームワークを設計することです。

以前のバージョンでは、ユーザーが StarRocks 上で分析する必要のある 100 以上のテーブルを含むライブラリを持っている場合、100 以上の外部テーブルを手動で作成し、FE を通じてメタデータを管理してからユーザーに使用させる必要がありました。ユーザーがスキーマの変更を行った場合、外部テーブルを再作成する必要があり、使用上の負担が大幅に増加していました。

コネクタフレームワークの設計において、カタログの概念を導入しました。ユーザーは手動で外部テーブルを作成する必要がなくなりました。たとえば、現在 Hive カタログと Iceberg カタログがあります。ユーザーは外部テーブルを作成する必要がなく、カタログを作成するだけでテーブルのメタデータ情報をリアルタイムで取得できます。Hive、Iceberg、Hudi に対して完全なサポートを提供しています。同時に、前述のメタデータ管理用の DLF、OSS、MaxCompute などのプロダクトが EMR プロダクトエコシステムに統合されています。

StarRocks の弾力的分析

プロダクト全体の紹介で、主要なプロダクト機能の 1 つが弾力性であると述べました。弾力性はどのように実現されるのでしょうか。中核となるソリューションは、StarRocks でコンピュートノード (以下 CN) をサポートすることです。以下の図の左部分は固定 StarRocks クラスターです。これらの固定 BE ノードには実際の SSD ストレージがあります。

緑色の部分が CN です。CN と BE は同じ実行エンジンコードを共有し、ステートレスなノードです。CN は Kubernetes 上にデプロイでき、データはオブジェクトストレージまたは HDFS に保存できます。Kubernetes の HPA 機能を通じて、クラスター負荷が高いときには CN を動的にスケールアウトし、クラスター負荷が低いときにはスケールダウンします。

上記の変革により、EMR StarRocks は弾力的スケーリングをサポートし、コスト削減を最大化できます。柔軟性を備えた上で、もう 1 つ解決する必要がある課題はリソース分離です。データレイク上のクエリワークロードは通常多様で、BI と直接連携してレポートを生成するものや、アナリストが詳細を問い合わせるアドホッククエリなどがあります。一般的に、ユーザーは物理的な分離ではなく、ソフト分離によるテナントリソースの弾力的な分離を望んでいます。たとえば、クラスターリソースがアイドル状態のときは、クエリがクラスターリソースを十分に利用することを許可し、クラスターリソースが逼迫しているときは、各テナントが自身のリソース制限に従ってリソースを使用します。そのため、StarRocks はリソースグループベースのリソース分離も実装しており、ユーザー、クエリ、IP アドレスのレベルで CPU / MEM / IO などのリソース使用を制限できます。

パフォーマンス最適化、エコロジー統合、柔軟性などの側面の紹介を通じて、Alibaba Cloud EMR StarRocks がデータレイク分析シナリオで何を行い、どの程度の成果を達成したかがわかります。まとめると、Alibaba Cloud EMR StarRocks データ分析の核心は「高速」と「統一」の 2 つのキーワードです。

超高速：Trino と比較して、パフォーマンスが数倍向上しています。このページの上述のテストデータは Hudi に対するものです。

統一：前述の JDBC データソースを含む多様なデータソースをサポートしています。現在、Trino から StarRocks への移行において多くの導入実績があり、基本的にシームレスな移行を実現できます。

Alibaba Cloud EMR StarRocks データレイク計画

ユーザーとの継続的なコミュニケーションと議論を通じて、データレイク分析がポピュラーなデータ分析技術となるためには、少なくとも以下の 4 つの要件を満たす必要があると考えています。

・Single Source of Truth。データは 1 つのコピーのみ存在し、ユーザーはデータフローを意識する必要がありません。

・高パフォーマンス。クエリレイテンシは秒レベル、あるいはサブ秒レベルに迫ります。

・弾力性。ストレージとコンピューティングのアーキテクチャを分離します。

・コスト効率に優れた。オンデマンドでスケールアップおよびスケールダウンします。

現在、データレイク分析が上記 4 つの要件を満たすことを妨げている 3 つの条件があります。

・データレイクストレージシステムは一般的に IO パフォーマンスが低く、ユーザーの低レイテンシクエリ要件を満たせません。

・データレイクとデータウェアハウスは明確に区別されています。一般的に、データレイククエリを高速化するために、その上にデータウェアハウスレイヤーを構築する必要があり、これが Single Source of Truth の原則を損ないます。

・複雑なデータスタック構造により、柔軟性、コスト効率、使いやすさを保証できません。

幾度もの検討、オープンな議論、慎重な検証を経て、新しいデータレイク分析のアプローチを提案しました。上記の課題を克服し、理想的なデータレイク分析の状態を達成することを目指しています。

新しいデータレイク分析のアプローチは、キャッシュ + マテリアライズドビューに等しいと考えています。

OSS などのデータレイクストレージシステムの IO パフォーマンスが低いため、データレイク分析のボトルネックは通常、データのスキャンに起因します。

データレイク分析のパフォーマンスをさらに向上させるため、ローカルディスクやメモリキャッシュを使用して IO パフォーマンスを高速化し、リモートストレージがパフォーマンスのボトルネックにならないようにします。キャッシュの導入はユーザーに対して透過的です。ユーザーは追加の運用保守作業なしに、キャッシュ高速化のメリットを享受できます。

リモートストレージと比較して、ローカルディスクやメモリは一般的にコストが高くなります。リソースを有効活用したいと考えています。ユーザーの分析に必要な列データのみがキャッシュに入り、コールドデータを自動的に排除して、キャッシュのスペース使用率を向上させます。

CPU キャッシュアーキテクチャと同様に、階層キャッシュ戦略も採用しています。第 1 レベルはメモリ、第 2 レベルはローカルディスクです。メモリにキャッシュされた非常にホットなデータについては、すべての読み取りがキャッシュ自体のメモリを直接参照でき、メモリコピーは不要です。データが絶えず更新されるシナリオでは、新しいデータが通常キャッシュミスを引き起こし、クエリレイテンシのジッターが発生します。

現在、いくつかの POC を実施しています。POC の結果、SSB マルチテーブル性能テストにおいて、キャッシュありのパフォーマンスはキャッシュなしよりも 3 倍以上高速であり、StarRocks のローカルテーブルにほぼ匹敵することが示されています。キャッシュは、Single Source of Truth を維持しながら高パフォーマンスを確保するのに役立ちます。キャッシュの特性により、ユーザーは真に弾力的スケーリングとコスト効率を実現できます。レイテンシに敏感なシナリオでは、キャッシュスペースを増やしてクエリレイテンシを低減します。レイテンシに敏感でないシナリオでは、キャッシュを削減または使用せず、コストを節約します。

ユーザーは通常、データのさらなる処理、事前集約、またはモデリングを行い、ビジネスのデータ分析に対するパフォーマンスと品質の要件をさらに満たしたいと考え、同時に繰り返し計算のコストも節約できます。しかし、Lambda アーキテクチャでも Kappa アーキテクチャでも、データレイク上のデータをさらに処理するために複雑なデータスタックを構築する必要があります。同時に、ユーザーはデータの整合性に対応するため、メタデータと処理済みの複数コピーのデータを個別に保守する必要があります。

ユーザーのデータ処理およびモデリングのニーズに応え、レイクとウェアハウスをさらに統合するため、より強力なマテリアライズドビュー機能を提供し、上記の課題を解決します。

まず、マテリアライズドビューは SQL で定義されるため、データ処理とモデリングが非常に簡単になります。次に、マテリアライズドビューは異なるデータのメタデータを統合し、外部に統一されたビューを提供します。ユーザーはクエリ SQL を書き換えることなく、クエリの自動ルーティングによる透過的な高速化を実現できます。StarRocks のビューはリアルタイムの増分更新をサポートし、ユーザーによりリアルタイムな分析機能を提供します。最後に、マテリアライズドビューは StarRocks のネイティブ機能として、運用保守コストを大幅に削減します。マテリアライズドビューを通じて、データレイクは真に Single Source of Truth を実現し、ユーザーがデータレイク上でデータをより簡単に処理およびモデリングできるようにし、レイクとウェアハウスの壁を打破し、データスタック全体のアーキテクチャを簡素化します。

まとめと展望

StarRocks データレイク分析の核心は、高速、統一、シンプルで使いやすいことです。

コネクタとデータカタログにより、データソースへのアクセスが非常に簡単になります。キャッシュにより、データレイクストレージシステムの IO パフォーマンスがボトルネックになることはありません。マテリアライズドビューにより、レイクとウェアハウスのデータフローがより自然になり、レイクとウェアハウスのビューが統一され、クエリの透過的な高速化が可能になり、データスタックのアーキテクチャがよりシンプルになります。最後に、クラウドと K8S の柔軟性を活かして、StarRocks データレイク分析は真の柔軟性とコスト効率を実現します。