Presto: SQL on Everything


モチベーション
SQLクエリ実行におけるレイテンシーを最小限するための分散クエリエンジン
Connector APIによるハイパフォーマンスな多数のデータソース対応

クライアントがSQL StatementをCoordinatorへHTTPリクエスト送信
cipepser HTTPなんですね
Coordinator node
クエリのパース・実行計画・最適化、オーケストレーション
Statement(クエリ)の実行レイテンシーが最小限になるように適切なworkerノードに分散させるか
Worker nodes
クエリ実行
External Storage Systemからのデータ読み込み


実行計画に対する最適化

論理プラン


Data Layout Propertiesの活用

Data Layout
実行計画でoptimizerがshuffle数を最小限にするためにphysical data layoutが使われる
physical data layoutに基づいた具体アルゴリズムの決定
例：joinカラムにインデックスが貼られていたらnested loop joinがコスト最適か検討など
stage
worker node間で分散処理される実行単位
shuffle （中間データ）
stage間でのインメモリのデータ送信
HTTP long polling
レイテンシーを増やし、メモリ・CPUオーバーヘッドも増やすのでoptimizerが最小限に最適化
一方、hiveやsparkは非同期にdiskに書き込まれる
shuffle fileがメモリに収まらない場合はディスクにReduce処理時にページキャッシュからではなくディスクから読み出される。

以下の図では上記クエリ実行のために4つのshuffleが必要
この段階ではまだdata layout propertiesを活用していなく、shuffle数の最適化も適用されていない。

Data Layoutを活用することでshuffle数を最小化


pipelines
シーケンシャルな一連の処理
task
各タスクは異なる入力データのセットに対して同じ計算を実行
同じstageのtaskは同じ計算
splits
外部データストレージからのデータチャンク
外部データストレージからデータを読み取るtaskに割り当てられる
例：hash joinは最低2つのpipelinesが必要
ハッシュテーブルの生成
探索し、join
LocalShuffle
pipeline1とpipeline2は内側テーブルのハッシュテーブル生成のために複数スレッドで計算



appendix

スケジューリング
一つの処理単位としてstageをcoordinatorがworkersに分散させる
クエリ実行のためにはstageの順番とどのくらいの量のタスクをスケジュールしどのwoerkerノードで実行するか
stage scheduling：2種類のスケジューリングポリシー
all-at-once
全ての処理を並行にスケジューリングしwall clock timeを最小限にする
アドホック分析などレイテンシーsensitiveなユースケース
phased
Task scheduling
task schedulerはplan treeを見てleaf stagesとintermediate stagesに分類
leaf
connectorからデータを取得するのでnetwork制約を考慮してスケジュール
ストレージが同居しているshared-nothing
CPUの大半はconnectorから読み取るデータのdecompressingやでコーディング、フィルタリング、変換などの処理
なので、できるだけ最大限に並列化するためにleaf stageは全てのworker nodeにスケジューリングされる
shared-storage modeではnetwork intensiveに実行計画になりうる
intermediate
クエリエンジンはいくつスケジュールするか。
Split Scheduling
外部データソースから取得するメタデータを含むデータsplit
redis: table info, key value, hosts list
Related