Presto on-disk processing

Sadayuki Furuhashi @frsyuki

MapReduceとPrestoなどの分散SQL処理系は、相互に機能を取り込みつつあるので、スループット重視か遅延重視で選ぶ話は近いうちに無くなり、スループット重視の設定と遅延重視の設定を同時にこなせるスケジューラをどのように実装するかという問題になると思うところ。

Sadayuki Furuhashi @frsyuki

HadoopやHiveは、ソースコードは汚くて”お世辞にも”クズコードが溢れている大変なシステムだけど、利用者も開発者も多い。スケジューラは中央集権型なので（運用大変というデメリットの一方で）、異なるスケジューリング戦略を要求するケースが混在するケースにも対応しやすい。

Sadayuki Furuhashi @frsyuki

なにより実行は完全にディスクをベースにした設計なので、メモリ上なら速いがディスクに落ちると遅い…みたいな予想困難な性能の不均一さが無いから、スケジューラの設定が簡単で理解しやすい。ただしデータが小さければ当然遅い。

Sadayuki Furuhashi @frsyuki

なにより実行は完全にディスクをベースにした設計なので、メモリ上なら速いがディスクに落ちると遅い…みたいな予想困難な性能の不均一さが無いから、スケジューラの設定が簡単で理解しやすい。ただしデータが小さければ当然遅い。

Sadayuki Furuhashi @frsyuki

なにより実行は完全にディスクをベースにした設計なので、メモリ上なら速いがディスクに落ちると遅い…みたいな予想困難な性能の不均一さが無いから、スケジューラの設定が簡単で理解しやすい。ただしデータが小さければ当然遅い。

tagomoris @tagomoris

@frsyuki クエリの対象データのサイズを見積もるのはシステムにはなかなか難しいし、そこで人間が走らせるエンジンを選択するというのは、それなりにコストパフォーマンスのいい構成なんじゃないかという気はします

tagomoris @tagomoris

@frsyuki クエリの対象データのサイズを見積もるのはシステムにはなかなか難しいし、そこで人間が走らせるエンジンを選択するというのは、それなりにコストパフォーマンスのいい構成なんじゃないかという気はします

Sadayuki Furuhashi @frsyuki

Prestoは最初から拡張性を最優先にした設計になっていて、ソースコードが圧倒的に読みやすく開発もデバッグもやりやすい。コンポーネントの密結合化で性能を稼いでいる部分は多少あるけども、全容を把握するのはがんばれば何とかなるレベル。

Sadayuki Furuhashi @frsyuki

中間処理の結果をディスクに落とすのは難しくないのだけど、入れるとディスクに落ちるケースとメモリに収まるケースで不均一さが生まれるので、データの統計情報が取るAPIが無い現状では、賢いスケジューラを実装するのは困難というか無理なので、たぶん全体のスループットがだいぶ落ちる。

Sadayuki Furuhashi @frsyuki

@tagomoris スケジューラに全部任せるのは今のところ現実的じゃないから、しばらくはエンジンを人間が選ぶ必要がありますね。ただTezが入ればHiveもMPPとMapReduceの混合になるし、エンジンの選択はシステムの選択ではなくパラメータの選択になるはず。

Sadayuki Furuhashi @frsyuki

@tagomoris スケジューラに全部任せるのは今のところ現実的じゃないから、しばらくはエンジンを人間が選ぶ必要がありますね。ただTezが入ればHiveもMPPとMapReduceの混合になるし、エンジンの選択はシステムの選択ではなくパラメータの選択になるはず。

tagomoris @tagomoris

@frsyuki Tezが充分賢ければそうなるけど、それが現実的になるのにあとどのくらいかかるかなーという感じですね。まあPrestoは導入も楽だし、繋ぎの解法としても現実的かな。

Sadayuki Furuhashi @frsyuki

@tagomoris あとどれくらいかかるでしょうねw　繋ぎの解法として現実的なのは賛成！

Sadayuki Furuhashi @frsyuki

Prestoにどういうスケジューラを実装すると、ディスクIOにも対応できるかを考えるのは結構おもしろい。ディスクを使うとクエリは落ちないけど、使えば使うほど性能は際限なく落ちる。メモリを使うと性能は落ちないけど、使えば使うほどクエリが落ちる。

Sadayuki Furuhashi @frsyuki

まず考えられるのは、1taskあたりのメモリ使用量の制限と、1taskあたり何GBまでディスクの使用を許可するか、というパラメータ。max.task-disk=2GBのような。max.task-diskはデフォルトでは0で、性能は予測可能だが、クエリはデータ量次第で落ちる。

Sadayuki Furuhashi @frsyuki

それから、1worker内で何GBまでのディスク使用を許可するか。これで性能的には結構制御可能になる。クラスタの利用効率の面では、snapshotをどう作るかのパラメータが必要になる。これはSQL内にコメントを入れる方法が現実的かな。

Sadayuki Furuhashi @frsyuki

select … from T1 join T2 group by …みたいなクエリで、表(T1 JOIN T2)の作成と、表(T1 JOIN T2)のGROUP BY処理は、MapReduceでは2段に分解できるわけだけども、MPPでは1段というか段という概念が無い。

Sadayuki Furuhashi @frsyuki

なので1クエリあたりの瞬間リソース消費量が増えるという形でシステムに影響する。瞬間リソース消費量の大きいクエリは、クラスタの利用効率を下げ、トータルスループットを落とすので、1クエリあたりの性能=クエリの遅延を犠牲にしても良いから、クラスタの利用効率を上げてね！という設定が必要。

Sadayuki Furuhashi @frsyuki

SQLを見ただけでMapReduceの段数が分かるのは訓練された人間だけだけども、単純化すればJOIN、GROUP BY、サブクエリごとにsnapshotの作成を許可するか否か、という設定になるのでは無いかな。

Sadayuki Furuhashi @frsyuki

許可するというか、1クエリの瞬間リソース消費量を抑えるのが目的なので、強制的にsnapshotを作成させて同期を入れさせる。クエリ内で指定する方法では、例えば select … from T1 shuffle join T2 shuffle group by … みたいな。

Sadayuki Furuhashi @frsyuki

でもそういう構文はSQL標準にはないので、特殊SQLを入れてしまうと既存ツールとの連携で困ったりするから良くない。なのでパラメータで制御したいけど、そのためにはデータの統計情報が必要そう。カラムのカーディナリティとか。

Sadayuki Furuhashi @frsyuki

JOINやGROUP BYで生成されるハッシュテーブルのサイズや、サブクエリの中間結果サイズが予測できるなら、task.shuffle-threshold=2GBみたいな指定で制御すればいい。でも予測できないだろうなぁ。Hiveでカーディナリティを保存する話ってあるんだっけか。

Sadayuki Furuhashi @frsyuki

指定したサイズを超えたらshufleにフォールバックする…というのは投機的実行と同じでクラスタ全体のスループットを落とすので良くない。どうしても人間任せになるなぁ。

Sadayuki Furuhashi @frsyuki

query.max-stagesならいけるな。stage数が3段を超えるならsnapshotを1回はさんで同期させる。どのstage間で同期させるかは、JOIN→サブクエリ→GROUP BYの順でテキトーに選ぶ。1クエリ辺りの瞬間リソース消費量に最大値を設定できる。

Sadayuki Furuhashi @frsyuki

query.max-stages=1は、全stage間でsnapshotが挟まることになる。MapReduceとほぼ同程度のクラスタ利用効率を期待できる。たまに遅延重視のクエリを走らせたい場合は、query.max-stages=3くらいのパラメータを付けてクエリを投げればいい。