Analysis principle of EMR StarRocks extreme speed data lake

StarRocks は強力なデータ分析システムです。その主な目的は、ユーザーに極めて高速で統合的かつ使いやすいデータ分析機能を提供し、低い運用コストでデータの価値をより迅速に把握できるようにすることです。シンプルなアーキテクチャ、効率的なベクトル化エンジン、そして新たに設計されたコストベースオプティマイザ（CBO）により、StarRocks の分析性能、特にマルチテーブルの JOIN クエリは同種の製品をはるかに上回っています。

より多くのユーザーの超高速データ分析のニーズに応え、StarRocks の強力な分析機能をより幅広いデータセットに適用するため、Alibaba Cloud のオープンソースビッグデータ OLAP チームとコミュニティが連携して、StarRocks のデータレイク分析機能を強化しています。これにより、StarRocks にローカルに保存されたデータだけでなく、Apache Hive、Apache Iceberg、Apache Hudi などのオープンソースデータレイクやデータウェアハウスに保存されたデータも、同じく優れたパフォーマンスで分析できます。

この記事では、StarRocks がデータレイク分析を高速に実現する技術的な背景、パフォーマンス、および今後の計画に焦点を当てて解説します。

1. 全体アーキテクチャ

データレイク分析のシナリオでは、StarRocks は主にデータの計算と分析を担当し、データレイクは主にデータの保存、整理、およびメンテナンスを担当します。上図は、StarRocks とデータレイクで構成される完成された技術スタックを示しています。

StarRocks のアーキテクチャは非常にシンプルです。システム全体のコアは FE（フロントエンド）と BE（バックエンド）プロセスのみで、外部コンポーネントに依存しないため、デプロイとメンテナンスが容易です。FE は主にクエリ文（SQL）の解析、クエリの最適化とスケジューリングを担当し、BE は主にデータレイクからデータを読み取り、Filter や Aggregate などの一連の操作を実行します。

データレイク自体は技術概念の集合体です。一般的なデータレイクは通常、Table Format、File Format、Storage の 3 つのモジュールで構成されます。Table Format はデータレイクの「UI」であり、構造化、半構造化、さらには非構造化データを整理して、HDFS などの分散ファイルシステムや OSS、S3 などのオブジェクトストレージに保存できるようにし、テーブル構造の関連セマンティクスを外部に公開することが主な役割です。Table Format には主に 2 つの系統があります。1 つはメタデータを一連のファイルとして整理し、実際のデータと一緒に分散ファイルシステムやオブジェクトストレージに保存する方式で、Apache Iceberg、Apache Hudi、Delta Lake がこれに該当します。もう 1 つは専用のメタデータサービスを使用してメタデータを別途保存する方式で、StarRocks のローカルテーブル、Snowflake、Apache Hive がこれに該当します。

File Format の主な機能は、データユニットの効率的な取得と圧縮の方法を提供することです。現在、一般的に使用されているオープンソースファイル形式には、列指向の Apache Parquet と Apache ORC、および行指向の Apache Avro があります。

Storage はデータレイク内でデータを保存するモジュールです。現在、データレイクで最も広く使用されているストレージは、分散ファイルシステムの HDFS、オブジェクトストレージの OSS、S3 などです。

FE

FE の主な機能は、SQL 文を BE が認識できるフラグメントに変換することです。BE クラスタを分散スレッドプールと見なすと、フラグメントはそのスレッドプール内のタスクに相当します。SQL テキストから分散物理実行計画への変換には、主に以下のステップが必要です。

• SQL Parse：SQL テキストを AST（抽象構文木）に変換する

• SQL Analyze：AST に基づく構文解析とセマンティック分析

• SQL Logical Plan：AST を論理計画に変換する

• SQL Optimize：関係代数、統計情報、コストモデルに基づいて論理計画を書き換え、コストが「最も低い」物理実行計画を選択する

• Generate Plan Fragment：Optimizer が選択した物理実行計画を、BE が直接実行できる Plan Fragment に変換する

• 実行計画のスケジューリング

BE

バックエンドは StarRocks のバックエンドノードで、データの保存と SQL 計算の実行を担当します。

StarRocks の BE ノードはすべて同等です。FE は特定のポリシーに従ってデータを対応する BE ノードに割り当てます。データインポート時、データは FE を経由せずに直接 BE ノードに書き込まれます。BE はインポートされたデータを対応する形式で書き込み、関連するインデックスを生成します。SQL 計算の実行時、SQL 文はまず特定のセマンティクスに従って論理実行ユニットに計画され、次にデータの分布に応じて物理実行ユニットに分割されます。物理実行ユニットはデータが保存されているノード上で実行されるため、データ転送やコピーを回避し、究極のクエリパフォーマンスを実現できます。

2. 技術的な詳細

StarRocks が高速な理由

CBO オプティマイザ

一般的に、SQL が複雑であればあるほど、結合されるテーブルが多く、データ量が大きいほど、クエリオプティマイザの重要性は増します。異なる実行方法によるパフォーマンスの差は数百倍から数千倍にも達する可能性があるからです。StarRocks のオプティマイザは主に Cascades と ORCA の論文に基づいて実装され、StarRocks の実行エンジンとスケジューラに合わせて深くカスタマイズ、最適化、革新されています。TPC-DS の 99 の SQL 文を完全にサポートし、共通式の再利用、相関サブクエリの書き換え、Lateral Join、CTE の再利用、Join Router、Join 分散実行戦略の選択、Runtime Filter のプッシュダウン、低カーディナリティ辞書の最適化などの重要な機能と最適化を実現しています。

CBO オプティマイザの重要なポイントの 1 つは、コスト見積もりが正確かどうかであり、その鍵は統計情報が適時かつ正確に収集されているかどうかです。StarRocks は現在、テーブルレベルと列レベルの統計情報をサポートし、自動収集と手動収集の両方に対応しています。自動収集と手動収集の両方で、完全収集とサンプリング収集が可能です。

MPP 実行

MPP（超並列処理）は、超並列計算の略称です。核心的なアプローチは、クエリ計画を単一ノードで実行可能な多数の計算インスタンスに分割し、複数ノードで並列実行することです。各ノードは CPU、メモリ、ディスクリソースを共有しません。MPP データベースのクエリパフォーマンスは、クラスタの水平スケールとともに継続的に向上します。

上図に示すように、StarRocks はクエリを論理的に複数のクエリフラグメントに分割します。各クエリフラグメントは 1 つ以上のフラグメント実行インスタンスを持つことができ、各フラグメント実行インスタンスはクラスタの BE 上でスケジュールされて実行されます。上図に示すように、フラグメントには 1 つ以上のオペレータを含めることができます。図のフラグメントには Scan、Filter、Aggregate が含まれています。上図に示すように、各フラグメントは異なる並列度を持つことができます。

上図に示すように、複数のフラグメントはメモリ内でパイプライン方式により並列に実行され、バッチエンジンのようにステージごとに順次処理されることはありません。

上図に示すように、Shuffle 操作は MPP データベースのクエリパフォーマンスをクラスタの水平スケールとともに継続的に向上させる鍵であり、高基数の集約や大規模テーブル結合を実現する上でも重要な要素です。

ベクトル化実行エンジン

データベース実行のボトルネックが IO から CPU へと徐々に移行するにつれ、StarRocks は CPU の実行性能を最大限に引き出すため、ベクトル化技術に基づいて実行エンジン全体を再実装しました。オペレータと式のベクトル化の核心は、列指向のバッチ実行にあります。行単位の実行と比較して、仮想関数呼び出しが少なく、分岐判断も少なくなります。また、行指向よりも列指向の実行の方が CPU キャッシュに優しく、SIMD 最適化も容易です。

ベクトル化は、データベース内のすべてのオペレータと式のベクトル化にとどまらず、ディスク、メモリ、ネットワークにおけるデータの列指向整理、データ構造とアルゴリズムの再設計、メモリ管理の再設計、SIMD 命令の最適化、CPU キャッシュの最適化、C++ の最適化などを含む、大規模で複雑なパフォーマンス最適化プロジェクトです。以前の行単位実行と比較して、ベクトル化により全体のパフォーマンスが 5 倍から 10 倍向上しました。

StarRocks がデータレイク分析を最適化する方法

ビッグデータ分析の分野では、データはデータウェアハウスだけでなく、データレイクにも保存されます。従来のデータレイク実装には Hive/HDFS があります。近年では、レイクハウスの概念がより普及しており、一般的な実装には Iceberg/Hudi/Delta があります。StarRocks は、ユーザーがデータレイク内のデータ価値をより深く掘り下げるのに役立つでしょうか。答えはイエスです。

前述の内容で、StarRocks がどのように超高速分析を実現するかを紹介しました。これらの機能をデータレイクに適用すれば、確実に優れたデータレイク分析体験をもたらします。このセクションでは、StarRocks がどのようにデータレイクの超高速分析を実現するかを紹介します。

まず、グローバルアーキテクチャを見てみましょう。StarRocks とデータレイク分析に関連する主要モジュールを下図に示します。データ管理はデータレイクが提供し、データストレージはオブジェクトストレージ OSS/S3 または分散ファイルシステム HDFS が提供します。

現在、StarRocks がサポートしているデータレイク分析機能は、以下の部分にまとめられます。

• Iceberg v1 テーブルのクエリをサポート https://github.com/StarRocks/starrocks/issues/1030

• Hive の外部テーブルクエリをサポート external table @ External_table @ StarRocks Docs (dorisdb.com)

• Hudi COW テーブルのクエリをサポート https://github.com/StarRocks/starrocks/issues/2772

次に、クエリ最適化とクエリ実行の観点から、StarRocks がどのようにデータレイクに高速分析機能を与えるかを見ていきましょう。

クエリ最適化

クエリ最適化の部分は、主に前述の CBO オプティマイザによって実現されます。データレイクモジュールはオプティマイザに統計情報を提供する必要があります。これらの統計情報に基づいて、オプティマイザは一連の戦略を用いてクエリ実行計画を最適化します。例を通じて、いくつかの代表的な戦略を見ていきましょう。

統計情報

以下の例を見てみましょう。生成された実行計画では、HdfsScanNode にカーディナリティ、平均行サイズなどの統計情報が表示されています。

MySQL [hive_test]> explain select l_quantity from lineitem;

これらの統計情報は、CBO オプティマイザに入る前に計算されます。たとえば、Hive の場合はメタデータキャッシュを利用してこれらの情報をキャッシュします。Iceberg の場合は、Iceberg のマニフェスト情報を使用してこれらの統計情報を計算します。これらの統計情報を取得した後、後続の最適化戦略の効果が大幅に向上します。

パーティションプルーニング

パーティションプルーニングは、ターゲットテーブルがパーティションテーブルの場合にのみ実行できる最適化手法です。パーティションプルーニングは、クエリ文のフィルタ条件を分析し、条件を満たす可能性のあるパーティションのみを選択して一致しないパーティションをスキャンしないことで、計算データ量を大幅に削減できます。たとえば、以下の例では、パーティション列として a_sold_date_sk を持つ外部テーブルを作成しています。

結合順序の最適化

複数テーブルの結合クエリの効率は、各テーブルが結合に参加する順序と密接に関連しています。たとえば、select * from T0, T1, T2 where T0.a=T1.a and T2.a=T1.a という SQL では、以下の 2 つの実行順序が考えられます。

• T0 と T1 を先に結合し、次に T2 と結合する

• T1 と T2 を先に結合し、次に T0 と結合する

T0 と T2 のデータ量とデータ分布に応じて、これら 2 つの実行順序は異なるパフォーマンスを示します。この状況に対応するため、StarRocks はオプティマイザに DP と貪欲法に基づく Join Reorder の仕組みを実装しました。現在、Hive のデータ分析では Join Reorder をサポートしており、他のデータソースへのサポートも開発中です。

述語プッシュダウン

述語プッシュダウンは、クエリ文のフィルタ式の計算をデータソースにできるだけ近い位置まで押し下げ、データ転送や計算のコストを削減します。データレイクシナリオでは、Min/Max などのフィルタ条件を Parquet 内にプッシュダウンする実装を行っています。Parquet ファイルを読み取る際に、不要な行グループを素早くフィルタリングできます。

たとえば、以下のクエリでは、l_count=1 に対応する条件が Parquet 側にプッシュダウンされます。

その他の戦略

上記の戦略に加えて、Limit プッシュダウン、TopN プッシュダウン、サブクエリ最適化などの戦略もデータレイク分析向けに適応しました。これにより、クエリパフォーマンスをさらに最適化できます。

クエリ実行

前述の通り、StarRocks の実行エンジンは全方位型のベクトル化と MPP アーキテクチャを採用しており、データレイク内のデータ分析に大きな向上をもたらします。次に、StarRocks がデータレイク分析クエリをどのようにスケジュールし、実行するかを見ていきましょう。

クエリスケジューリング

データレイクのデータは通常 HDFS と OSS に保存されており、混在環境と非混在環境の両方を考慮して、フラグメントスケジューリング用の負荷分散アルゴリズムを実装しました。

• パーティションプルーニング後に、クエリ対象のすべての HDFS ファイルブロックを取得する

• 各ブロックに対して THdfsScanRange を構築する。hosts にはブロックのすべてのレプリカが存在するデータノードのアドレスが含まれ、最終的に List を取得する

• コーディネータは、すべての BE に現在割り当てられているスキャン範囲の数のマップ、各データノードのディスクに割り当てられている読み取り対象ブロック数のマップ、および各 BE に割り当てられたスキャン範囲の平均数 numScanRangePerBe を管理する

• ブロックレプリカが存在するデータノードに BE がある場合（混在環境）

• 各スキャン範囲は、レプリカが存在する BE の中で最もスキャン範囲数が少ない BE に割り当てられる。BE に割り当てられたスキャン範囲数が numScanRangePerBe を超える場合は、リモート BE から最もスキャン範囲数が少ないものを選択する

• 複数の BE でスキャン範囲数が同じ場合は、BE 上のディスクの状況を考慮し、レプリカが存在するディスク上で既に割り当てられた読み取り対象ブロック数が少ない BE を選択する

• ブロックレプリカが存在するデータノードに BE がない場合（分離デプロイまたはリモート読み取り可能）

• スキャン範囲数が最も少ない BE を選択する

クエリ実行

BE 側にスケジュールされて実行に移されると、全体の実行プロセスはベクトル化されています。以下の Iceberg の例を参照してください。BE 側の IcebergScanNode に対応するのは現在 HdfsScanNode のベクトル化実装で、他のオペレータについても同様です。BE 側はベクトル化実装となっています。

3. ベンチマーク

TPC-H は、TPC（Transaction Processing Performance Council）が意思決定支援アプリケーションをシミュレートするために開発したテストセットです。ビジネス指向のアドホッククエリと同時データ操作のスイートで構成されています。

TPC-H は、実際の運用環境に基づいて販売システムのデータウェアハウスをシミュレートします。テストには 8 つのテーブルが含まれ、データ量は 1 GB から 3 TB まで設定できます。ベンチマークテストには 22 のクエリが含まれ、主な評価指標は各クエリの応答時間、すなわちクエリの送信から結果の返却までの時間です。

StarRocks はローカルストレージクエリと Hive 外部テーブルクエリを使用してテストを行います。StarRocks On Hive と Trino On Hive は同じデータをクエリします。データは ORC 形式で保存され、zlib 形式で圧縮されています。テスト環境は Alibaba Cloud EMR を使用して構築されています。

テストの結果、StarRocks のローカルストレージクエリの合計時間は 21 秒、StarRocks の Hive 外部テーブルクエリの合計時間は 92 秒でした。Trino クエリの合計時間は 307 秒でした。StarRocks On Hive のクエリパフォーマンスは Trino をはるかに上回っていますが、ローカルストレージクエリとはまだ差があります。主な要因は、リモートストレージへのアクセスによるネットワークオーバーヘッドの増加と、リモートストレージのレイテンシと IOPS が一般的にローカルストレージに劣ることです。今後はキャッシュなどの仕組みでこの問題を補い、StarRocks のローカルテーブルと StarRocks On Hive の差をさらに縮めていく計画です。

具体的なテスト手順については、StarRocks vs Trino TPC-H パフォーマンス比較レポートを参照してください。

4. 今後の計画

フルベクトル化実行エンジン、CBO オプティマイザ、MPP 実行フレームワークなどのコア技術により、StarRocks は現在、他の同種の製品をはるかに上回る超高速なデータレイク分析機能を実現しています。長期的には、StarRocks のデータレイク分析におけるビジョンは、ユーザーに極めてシンプルで使いやすく高速なデータレイク分析機能を提供することです。この目標を達成するために、StarRocks にはまだ多くの作業が残っています。

• Push Based のパイプライン実行エンジンを統合し、クエリ応答速度をさらに向上させる

• ホットデータとコールドデータの自動階層化ストレージ。ユーザーは頻繁に更新されるホットデータを StarRocks のローカルテーブルに保存でき、StarRocks はコールドデータをローカルテーブルからデータレイクへ定期的に自動移行する

• 明示的な外部テーブル作成の手順をなくし、ユーザーはデータレイクに対応するリソースを作成するだけで、データレイクのデータベーステーブルの自動同期を実現できるようにする

• Apache Hudi の MOR テーブルや Apache Iceberg の v2 テーブルのサポートを含む、データレイク製品の機能への対応をさらに拡充する。データレイクへの直接書き込みをサポートし、Time Travel クエリをサポートし、カタログのサポートを改善するなど

• 階層キャッシュを通じて、データレイク分析のパフォーマンスをさらに向上させる