負けられない戦いがそこにある。アドベントカレンダー3日目。
Arm Treasure Data Advent Calendar 2019 - Adventar
Prestoのバージョンアップに際して、トレジャーデータのPrestoを利用して検証するのは簡単なんですが、あくまでトレジャーデータのPrestoの検証になってしまうところが気になります。 ブログにふと書こうと思った時に、なるべくOSSとしてのPrestoでの機能による性能差を書きたいなというのがポリシーなところです。 また、再現性がないことはよくないので、PrestoのDockerイメージを利用して、ローカルでTPCHを利用して検証してみます。また、今回の検証対象として、Distributed Sortの仕組みを試してみたいので、複数Workerを持つPrestoを利用してみました。
そこで、TDのPrestoチームメンバーが公開しているPresto複数ワーカーをDockerで立ち上げるツールを使って試してみます。
GitHub - Lewuathe/docker-presto-cluster: Multiple node presto cluster on docker container
（結論としては、パフォーマンス検証がローカルでちゃんとできているわけではないです。）
検証環境
- MacBook Pro (13-inch, 2016)
- 2 GHz Intel Core i5
- 8 GB 1867 MHz LPDDR3
検証項目
- Distributed Sortを利用した場合 Order byおよびWindow関数でのOrder byの性能差のチェック: 4.7. Distributed Sort — Presto 326 Documentation
Treasure DataのPrestoのバージョンは、0.205となっており、2020年1月にPresto317に移行予定です。そのためここではPresto317での性能を比較します。
Distributed Sortとは何か？
分散クエリエンジンにおいて、ORDER BYという処理は鬼門です。 ORDER BYはデータの順番を単一に保証しているため、同じデータは同じ順序で結果を返す必要があります。 そのため、全てのデータを一つのサーバ（ノード）に集めて、並び替えを行うため、数千万〜数億件以上のデータがある場合に１ノードに全てのデータが集まるため、処理パフォーマンスがよくないだけなく、Prestoのようなメモリにデータを保存する処理する場合に非常に大きなメモリ消費を行う必要があります。
そのため、基本的にはORDER BYを使わないというのが鉄則です。
一方で、そうした問題を解決するための仕組みの一つがDistributed Sortです。これは１ノードでソートする前段階で複数のノードを利用してソートし、さらにソートした結果を元に１ノードでソートをするような仕組みです。 これにより、一度に全量データのソートをする必要がないことで、ピークメモリの消費量を減らすメリットがあります。
Prestoクラスタの立ち上げ
下記設定でPresto 317を3 Worker付きで立ち上げることができます。
version: '3'
services:
coordinator:
image: "lewuathe/presto-coordinator:317"
ports:
- "8080:8080"
container_name: "coordinator"
command: coordinator
worker0:
image: "lewuathe/presto-worker:317"
container_name: "worker0"
ports:
- "8081:8081"
command: worker0
worker1:
image: "lewuathe/presto-worker:317"
container_name: "worker1"
ports:
- "8082:8081"
command: worker1
worker2:
image: "lewuathe/presto-worker:317"
container_name: "worker2"
ports:
- "8083:8081"
command: worker2
そして下記を実行。
$ docker-compose up -d
Dockerが立ち上がったら、127.0.0.1:8080に接続することでPresto Web UIが参照できます。
Prestoクラスタへの接続
下記のjarファイルをダウンロードすることで、CLIからPrestoクラスタに接続できます。
2.2. Command Line Interface — Presto 326 Documentation
// 実行権限の設定 $ chmod +x presto-cli-326-executable.jar // tpchデータベースへの接続 $ ./presto-cli-326-executable.jar --server localhost:8080 --catalog tpch
接続後、下記のスキーマがあるので、今回はローカルのテストなのでtinyを利用してチェックをしていきます。
presto> SHOW SCHEMAS FROM tpch;
Schema
--------------------
information_schema
sf1
sf100
sf1000
sf10000
sf100000
sf300
sf3000
sf30000
tiny
(10 rows)
Query 20191202_125600_00004_x4aa8, FINISHED, 4 nodes
Splits: 53 total, 53 done (100.00%)
0:08 [10 rows, 119B] [1 rows/s, 14B/s]
テスト
tinyなので、6万件ほどのデータしかない。
./presto-cli-326-executable.jar --server localhost:8080 --catalog tpch --execute "SELECT count(1) FROM tiny.lineitem" "60175"
SELECT GROUP BYの場合
./presto-cli-326-executable.jar --server localhost:8080 --catalog tpch --execute "SELECT orderkey, count(1) FROM sf1.lineitem GROUP BY 1"
- Execution Time 14.55s
- Peak Total Memory 180kB
SELECT GROUP BY ORDER BYの場合
Distributed Join = True
./presto-cli-326-executable.jar --server localhost:8080 --catalog tpch --execute "SET SESSION distributed_sort=true; SELECT orderkey, count(1) FROM tiny.lineitem GROUP BY 1 ORDER BY 1"
- Execution Time 5.06s
- Peak Total Memory 650.86kB
Distributed Join = false
./presto-cli-326-executable.jar --server localhost:8080 --catalog tpch --execute "SET SESSION distributed_sort=false; SELECT orderkey, count(1) FROM tiny.lineitem GROUP BY 1 ORDER BY 1"
- Execution Time 14.81s
- Peak Total Memory 489.67kB
というわけでデータ量が少ないとテストにならない。。。
しかしDistributed Joinの有無でのわかりやすい違いとしては、実行計画の違いがある。 下記のプランをみてもらうとわかるが、Distributed Joinの方が、各ワーカーでのソート処理がは入っている分、全体としてのステージ数が増えている。 基本的には、１ノードでのソート処理が非常に重い処理になるので、ステージが別れてしまうデータ転送などのデメリットがあったとしても、Distributed Jionの方がメモリ効率だけではなく、パフォーマンスも良いのではないかとは思う。
まとめ
ローカルでメモリ食う処理をやろうとするとけっこうすぐPrestoが死ぬ。 所詮8GBしか積んでなかった・・・
なので、テストにならないのでやめましょう・・・ とはいえ、ローカルでPrestoを１コマンドで試せて便利なので、どうぞ！
