メインコンテンツまでスキップ
バージョン: 4.x

並列実行

Dorisの並列実行モデルはPipeline実行モデルであり、主にHyper論文で説明されている実装にインスパイアされています。Pipeline実行モデルは、Doris内のクエリスレッド数を制限しながらマルチコアCPUの計算能力を最大限に活用し、実行時のスレッド爆発の問題に対処します。その設計、実装、および有効性の詳細については、[DSIP-027](DSIP-027: Support Pipeline Exec Engine - DORIS - Apache Software Foundation)および[DSIP-035](DSIP-035: PipelineX Execution Engine - DORIS - Apache Software Foundation)を参照してください。

Doris 3.0以降、Pipeline実行モデルは元のVolcanoモデルを完全に置き換えています。Pipeline実行モデルに基づいて、DorisはQuery、DDL、およびDMLステートメントの並列処理をサポートしています。

物理プラン​

Pipeline実行モデルをより良く理解するために、まず物理クエリプランにおける2つの重要な概念であるPlanFragmentとPlanNodeを紹介する必要があります。以下のSQLステートメントを例として使用します：

SELECT k1, SUM(v1) FROM A,B WHERE A.k2 = B.k2 GROUP BY k1 ORDER BY SUM(v1);

FEは最初にこれを次の論理プランに変換します。各ノードはPlanNodeを表します。各ノードタイプの詳細な意味は、物理プランの紹介で確認できます。

pip_exec_1

DorisはMPPアーキテクチャ上に構築されているため、各クエリはクエリレイテンシを削減するために、可能な限りすべてのBEを並列実行に関与させることを目指します。そのため、論理プランは物理プランに変換される必要があります。この変換は本質的に、DataSinkとExchangeNodeを論理プランに挿入することを含みます。これら2つのノードは、複数のBE間でのデータシャッフルを促進します。

変換後、各PlanFragmentはPlanNodeの一部に対応し、独立したタスクとしてBEに送信できます。各BEは、PlanFragment内に含まれるPlanNodeを処理し、DataSinkとExchangeNode演算子を使用して、他のBEにデータをシャッフルして後続の計算を行います。

pip_exec_2

Dorisのプランは3つのレイヤーに分かれています：

  • PLAN: 実行プラン。SQL文はクエリプランナーによって実行プランに変換され、実行エンジンに提供されて実行されます。

  • FRAGMENT: Dorisは分散実行エンジンであるため、完全な実行プランは複数の単一マシン実行フラグメントに分割されます。FRAGMENTは完全な単一マシン実行フラグメントを表します。複数のフラグメントが組み合わさって完全なPLANを形成します。

  • PLAN NODE: 演算子であり、実行プランの最小単位です。FRAGMENTは複数の演算子から構成され、各演算子は集約や結合操作などの特定の実行ロジックを担当します。

Pipeline実行​

PlanFragmentは、FEがBEに実行のために送信するタスクの最小単位です。BEは同じクエリに対して複数の異なるPlanFragmentを受信する場合があり、各PlanFragmentは独立して処理されます。PlanFragmentを受信すると、BEはそれを複数のPipelineに分割し、複数のPipelineTaskを開始して並列実行を実現し、クエリ効率を向上させます。

pip_exec_3

Pipeline​

Pipelineは、SourceOperator、SinkOperator、およびいくつかの中間演算子から構成されます。SourceOperatorは外部ソースからのデータ読み取りを表し、テーブル（OlapTableなど）またはバッファ（Exchangeなど）のいずれかです。SinkOperatorはデータ出力を表し、ネットワーク経由で他のノードにシャッフル（DataStreamSinkOperatorなど）するか、ハッシュテーブルに出力（集約演算子、結合ビルドハッシュテーブルなど）することができます。

pip_exec_4

複数のPipelineは実際に相互依存しています。JoinNodeを例にとると、それは2つのPipelineに分割されます。Pipeline-0はExchangeからデータを読み取ってハッシュテーブルを構築し、Pipeline-1はテーブルからデータを読み取ってprobe操作を実行します。これら2つのPipelineは依存関係によって接続されており、Pipeline-1はPipeline-0が完了した後にのみ実行できます。この依存関係はDependencyと呼ばれます。Pipeline-0が実行を完了すると、DependencyのSet_readyメソッドを呼び出してPipeline-1に実行準備ができていることを通知します。

PipelineTask​

Pipelineは実際には論理的な概念であり、実行可能なエンティティではありません。Pipelineが定義されると、さらに複数のPipelineTaskにインスタンス化される必要があります。読み取りが必要なデータは異なるPipelineTaskに分散され、最終的に並列処理が実現されます。同じPipelineの複数のPipelineTask内の演算子は同一ですが、状態が異なります。例えば、異なるデータを読み取ったり、異なるハッシュテーブルを構築したりする場合があります。これらの異なる状態はLocalStateと呼ばれます。

各PipelineTaskは最終的にスレッドプールに独立したタスクとして送信されて実行されます。Dependencyトリガーメカニズムにより、このアプローチはマルチコアCPUをより有効活用し、完全な並列性を実現できます。

Operator​

ほとんどの場合、Pipeline内の各演算子はPlanNodeに対応しますが、例外のある特別な演算子もあります：

  • JoinNodeはJoinBuildOperatorとJoinProbeOperatorに分割されます。
  • AggNodeはAggSinkOperatorとAggSourceOperatorに分割されます。
  • SortNodeはSortSinkOperatorとSortSourceOperatorに分割されます。 基本原則は、特定の「ブレイキング」演算子（計算を実行する前にすべてのデータを収集する必要がある演算子）については、データ取り込み部分がSinkに分割され、演算子からデータを取得する部分がSourceと呼ばれることです。

並列スキャン​

データのスキャンは非常に重いI/O操作です。ローカルディスク（またはデータレイクシナリオでHDFSやS3）から大量のデータを読み取る必要があり、さらに長いレイテンシが発生するため、大量の時間を消費します。そのため、ScanOperatorに並列スキャン技術を導入しました。ScanOperatorは複数のScannerを動的に生成し、それぞれが約100万から200万行のデータをスキャンします。スキャンを実行中、各Scannerはデータ解凍、フィルタリング、およびその他の計算などのタスクを処理し、ScanOperatorが読み取るためにDataQueueにデータを送信します。

pip_exec_5

並列スキャン技術を使用することで、不適切なバケッティングやデータスキューにより特定のScanOperatorが過度に長時間かかる問題を効果的に回避でき、そうでなければクエリレイテンシ全体が遅くなります。

Local Shuffle​

Pipeline実行モデルにおいて、Local ShuffleはPipeline Breakerとして機能し、異なる実行タスク間でデータをローカルで再分散する技術です。HASHやRound Robinなどの方法を使用して、上流Pipelineが出力するすべてのデータを下流Pipelineのすべてのタスクに均等に分散します。これにより、実行中のデータスキューの問題を解決し、実行モデルがデータストレージやクエリプランによって制限されなくなります。Local Exchangeがどのように動作するかを説明する例を示します。

前の例のPipeline-1を使用して、Local Exchangeがどのようにデータスキューを防ぐかをさらに説明します。

pip_exec_6

上図に示すように、Pipeline-1にLocal Exchangeを挿入することで、Pipeline-1をPipeline-1-0とPipeline-1-1にさらに分割します。

現在の並行レベルが3（各Pipelineに3つのタスク）で、各タスクがストレージ層から1つのバケットを読み取ると仮定します。3つのバケットの行数はそれぞれ1、1、7です。Local Exchangeを挿入する前後の実行は次のように変化します：

pip_exec_7

右の図からわかるように、HashJoinとAgg演算子が処理する必要があるデータ量が（1, 1, 7）から（3, 3, 3）に変化し、データスキューを回避します。

Local Shuffleは一連のルールに基づいて計画されます。例えば、クエリがJoin、Aggregation、Window Functionsなどの時間のかかる演算子を含む場合、データスキューを可能な限り最小化するためにLocal Shuffleが使用されます。