日本ソフトウェア科学会第 32 回大会 (2015 年度) 講演論文集

Spark 上での TPC-H Benchmark のマルチレイヤ最
適化の評価
千葉 立寛

堀井 洋

小野寺 民也Apache Spark は Scala で実装されたインメモリ指向の分散処理フレームワークであり，Hadoop/MapReduce よりも高速に大規模データを処理可能なシステムとして注目されている．Spark は JVM 上で動作するため，JVM の設定により，GC やロックの挙動を制御することが可能である．また，Spark では多数のスレッドが並列してデータ処理を行うため，OS の設定によりスレッドがメニーコアや SMT を効率的に利用することが可能である．さらに，Spark ではアプリケーションの書き方によりメモリやディスクの利用法が変化するため，解析するデータや Sparkの特性を考慮してアプリケーションを修正することにより，メモリを有効に利用可能になる．このように，アプリ
ケーションをより高速に実行するためには，様々なレイヤを総合的にチューニングすることが重要である．本稿では，Hive クエリ，Spark，JVM，OS レイヤごとのチューニングの効果を TPC-H Benchmark を題材として定量的に示す．
Besides being an in-memory oriented computing framework, Spark runs on top of Java Virtual Machine(JVM), so JVM parameter tuning is important for improving Spark application performance. Misconﬁgured parameters and settings result in performance degradation, for example, using too large Java heaps
often causes long garbage collection pause time, which accounts for over 10-20% of application execution
time. Moreover recent modern computing node has many cores and it supports to run multiple threads simultaneously with SMT technology. Thus it is also important to optimize in full stack. Not only optimizing
JVM parameter, but also OS parameters, Spark conﬁguration and application code itself based on CPU
characteristics are needed to take full advantage of underlying computing resource. In this presentation, we
will use TPC-H benchmark as our optimization case study and we will introduce our optimization resultsin full stack. As a result, we will provide how they contribute to improve Spark application performance.

つである．このような背景の中で，Apache Spark†1

1 はじめには Hadoop/MapReduce よりも高速に処理可能なシ

Hadoop エコシステムを中心とした大規模データステムとして注目されているフレームワークの一つ

向け分散処理基盤の普及とともに，ビジネス・サイ

である．Spark は，できるだけ再利用可能な形式でエンスの分野を問わず，様々な分野でのビックデー

インメモリにデータを配置し，ディスク I/O を極力タ処理に対する期待が高まっている．とりわけ，文発生させないようにすることによって低レイテンシ

献 [2] で様々な機械学習アルゴリズムが MapReduce

な分析処理を目指したシステムであるが，Spark ラ

で記述可能であることが示されたことにより，機械

ンタイム上で動作する様々なコンポーネントがコミュ

学習やグラフ処理などの繰り返し処理を多く含むよ

ニティにより開発されている．例えば，機械学習向け

うなワークロードを効率よく処理することが近年のライブラリ (MLlib) や，リレーショナルクエリ処理

ビックデータ処理基盤に求められる重要な要素の一を実現する Spark SQL，ストリーム処理向け Spark∗ This is an unrefereed paper.

Copyrights belong to
the Author(s).
Tatsuhiro Chiba, Hiroshi Horii, Tamiya Onodera, 日
本アイ・ビー・エム株式会社 東京基礎研究所, IBM
Research - Tokyo.

Streaming，R や Python などデータ分析で広く使われる言語への API の提供など，インタラクティブな†1 http://spark.apache.org/

処理からストリーム処理まで，様々な大規模データを2 節で背景とそれぞれのレイヤでの最適化において処理するための共通基盤としての利用が急速に広まっ

必要なことを列挙し，それを踏まえて，3 節で SQLてきている．

クエリの書き換え，4 節で Spark オプションおよび

Hadoop/MapReduce では，Map・Reduce の各処

JVM 数の最適化，5 節で NUMA を用いた最適化に

理ごとにデータをディスクに書き込むため，ディスク

ついて論じる．6 節ではそれぞれのチューニングにがボトルネックになるのに対し，Spark においても入よって得られた性能向上について説明する．

力の一次データソースとしてディスクアクセスが発生するものの，Spark では多くの場合インメモリで処2 背景

理されることが多いため，CPU やメモリ，さらには

2. 1 Apache Spark

ネットワークがボトルネックになることが予想されSpark は，Scala で実装されたインメモリ指向の分散

る．これらのボトルネックをできるだけ回避しノード

処理フレームワークであり，大規模データ向け分散処での実行性能を最大化するためには，Spark ランタイ

理基盤として現在広く使われている Hadoop/MapRe-

ム自体のチューニングだけではなく，アプリケーショ

duce よりも高速に処理可能なシステムとして注目さンを記述するユーザーの側もこれまで以上に CPU やれている．HDFS などに保存された同じ大規模デー

メモリを意識したプログラムの記述や設定を行う必タに対して，従来の Hadoop では扱いにくい機械学

要がある．アプリケーションの実行性能を最大化す

習やグラフ解析のような繰り返し処理を多く含むようるための要素として，Spark アプリケーション自体のなワークロードにおいて，入力データや中間データ，

書き方，Spark の設定の変更，JVM のチューニング，

結果を再利用可能な形でインメモリ上に置き，出来る

ハードウェアを意識した OS のチューニングなどが挙だけディスク I/O を発生させないにすることで，低げられるが，どのような設定を行うことで性能が向上

レイテンシな分析処理の実現を目指している．Spark

するかについての知見は少なく，それらのチューニン

上で動作する様々なコンポーネントやライブラリの開グは相互に影響するため，複数のレイヤでのチューニ発も活発であり，機械学習向けの MLlib ライブラリ

ングを総合的に判断して最適化していく必要がある．

や，グラフ処理向けの GraphX，SQL クエリ処理向

本稿では，TPC-H Benchmark を題材として，Hive

けの Spark SQL などがある．

クエリ，Spark，JVM，OS レイヤごとのチューニン

Spark では，RDD (Risillient Distributed Datasets)

グによってアプリケーション全体の実行性能がどの[8] と呼ばれるイミュータブルな分散コレクションを程度改善できるかを定量的に示す．まず，何もチューベースに，RDD に対して様々なオペレーション（例

ニングを行わない設定でベースラインの実行性能をえば map, ﬁlter, count, reducebykey など）を適用さ測定し，クエリの書き方，Spark や JVM の設定のせて新たな RDD を生成することで処理を実現する

変更，アーキテクチャの NUMA 構成を考慮した OS

プログラミングモデルである．RDD の変換フローは設定のチューニングによって，どの程度性能が改善

DAG として表現され，それぞれの RDD に定義されするかを計測し比較していく．1 ノードの POWER8た依存関係，すなわち，オペレーションとデータの局

マシン上で測定した結果，クエリの書き換えにより所性を考慮した上で，ステージに分割されて並列分散

20%程度，NUMA の設定を加えることでほぼ全ての実行される．

クエリで 10%程度，Executor JVM 数を調整するこ

Spark におけるメインプログラムをドライバと呼とで 20%程度，JVM オプションのチューニングによび，RDD や DAG，ステージの分割などの情報が管理

り 20-30%程度性能が向上した．これらのチューニン

されている．多数のノードが使えるような大規模な分

グを全て合計したものとベースラインを比較すると，

散システムにおいて，ドライバはクラスタマネージャ

クエリによっては 20-40%程度性能が向上することをを介して，アプリケーションの実行に必要な計算資確認した．

源を確保する．Spark では，Built-in の Standalone，

YARN，Mesos の 3 つのクラスタマネージャが利用

Hadoop なシステムの性能評価にもしばしば用いられ可能であるが，本稿では Standalone モードでの実行

る [4]．QGEN では各種 DB 向けのクエリが生成可能を前提とする．RDD 内の分割されたデータに対するであるものの，HiveQL などの SQL ライクなクエリ

処理の単位はタスクと呼ばれ，クラスタマネージャ

言語には対応していないため，本稿では，Hive 上でによって確保された Executor JVM 上で実行される．

Executor JVM に用意された Worker Threads でタ実行可能なクエリとしてオンライン上に公開されている TPC-H on Hive†3 をベースラインの評価用クエ

スクは実行され，Executor JVM は予め設定されたリとして用いている．

上限に達するまで Worker Threads を並行に起動することができる．

2. 4 マルチレイヤでの最適化にむけて
Spark は，生成された RDD をインメモリに出来る

2. 2 Spark SQL

だけ保持したり明示的にキャッシュすることで，同一Spark SQL [1] は，リレーショナル処理を Spark 上のデータセットに対する繰り返し処理を行うようなで実現するための分散クエリエンジンであり，最もワークロードに対しては，Hadoop よりも高い実行性

開発が盛んなプロジェクトの一つである．Shark と呼能を達成できる．Hadoop においては，MapReduce

ばれていた SQL on Spark [7] を発展させ，クエリオのインプットデータや中間データの Read/Write がプティマイザ (Catalyst) や，スキーマ情報を持った大量に発生する I/O バウンドな実行モデルであるた

テーブルを用意する API を用いてリレーショナル処

め，I/O スループットを高めることが性能向上の最も理を従来の Spark アプリケーションの中に組み込む

重要な要素であった．一方 Spark においては，ディ

こむことが可能になる DataFrame API を用いるこスク I/O が Hadoop に比べて減ってインメモリでのとで，複雑なクエリ処理に対してもアグレッシブに最処理に近づいていくため，依然として I/O は重要な適化されたコードを生成する機能を有する．さらに，

要素ではあるものの，ジョブによっては CPU やメモ

Spark SQL は Hive と互換性を持っており，既存のリ，さらにはネットワークバウンドになることが考え

Hive クエリを変更することなく Spark 上で実行可能

られる．そのため，Spark アプリケーションの実行性である．

能を向上させるためには，アプリケーション自体の特Spark SQL 上で Hive クエリを実行するには，Spark

性や，オペレーティングシステム，計算機システムのにインテグレートされている Hive の Thrift Server を

ハードウェア構成などをより一層考慮して最適化して起動する．Thrift Server は，Spark アプリケーション

いく必要があり，考慮しない場合，期待通りの性能がとして動作し，クラスタマネージャを介して Executor

達成できない可能性がある．本節では，システムからJVM を確保する．その後，Hive クエリは Beeline コ

アプリケーションを含めた横断的な最適化をしていく

マンドを用いて Thirt Server にサブミットされる．

上で重要な要素を各レイヤごとに列挙していく．

Spark アプリケーション
ユーザーに並列分散プログラムの知識がなくとも，

2. 3 TPC-H Benchmark
TPC-H†2 は，主に RDBMS に対する性能を評価す

オペレータを連結させることで Spark のランタイム

るために用いられるベンチマークであり，8 つのテー側で自動的にタスクが分散処理され，クエリによってブルデータに対して，22 種類の異なるクエリが定義

は複数のオペレータを同時に実行するような最適化がされている．テーブルデータを生成する DBGEN と適用されるが，アプリケーションの動作と Spark のクエリを生成する QGEN が TPC から提供されておコア動作を正しく理解してアプリケーションを記述

り，Hive や Presto，Impala，Tajo といった SQL on

することで，さらなる性能向上が期待される．例え†2 http://www.tpc.org/tpch/†3 https://github.com/rxin/TPC-H-Hive

ば，SQL においては，Catalyst による最適なクエリめであると考えるが，一時テーブルを作成する場合，プランの生成，Hive に実装されているコストベース

ファイルへ書き戻さなければならず，さらにジョブがオプティマイザ，クエリ対象となるテーブルのカラム分割されてしまうので非効率である．これらのクエリやデータサイズなどを事前に解析して統計情報を生に対しては，サブクエリを作成して出来るだけ少ない成しておくなど，システム側で利用可能な情報から自ジョブ数になるように変換する．

動的な最適化を期待するだけではなく，SQL の記述そのものに無駄なものがないかを考慮すべきである．

JVM

3. 2 JOIN 順序の交換
複数のテーブル A, B の JOIN 操作を行う場合，Spark は，既存の JVM 上で動作するアプリケー

Spark では，A をロードするステージ 0，B をローション (ウェブアプリケーション，Hadoop など) とドするステージ 1，そしてロードしたデータをシャッ

同様，JVM 上のチューニングも重要な要素である．

フルして JOIN するステージ 2 に分割される．(図巨大なヒープを必要とし，大量のオブジェクトを生1：左) シャッフルでは二つのテーブルデータがネッ

成する Spark においては，従来型のアプリケーショ

トワークを介して複数の Executor JVM に送られるンと比べてより一層ガーベージコレクション (以下，ため，出来るだけシャッフル時のデータ量を減らすこ

GC) の性能によってアプリケーションの性能が大きくとは，Hadoop と同様，Spark でも有効である．異な左右される．さらには，ユーザーが指定可能な JVMるサイズのテーブルを多段に JOIN する場合，大きの起動オプションには様々な種類があり，どの設定がなテーブルから JOIN すると，大量のデータが何度Spark にとって適切なのかを判断することは容易でも Executor JVM 間をシャッフルされるため非常にはないため，それぞれのチューニングによってどの程非効率である．よって，出来るだけ小さなテーブルか度 Spark の実行性能に影響するかを知ることは重要

ら JOIN していき，大きなテーブルの JOIN 順序はである．

最後になるよう書き換えを行う．

システム
多数の Worker Threads が並列に動作する Spark

3. 3 LIMIT 句による JOIN の変換

においては，多数のコアやソケットを搭載するノー

上記で述べたとおり，JOIN 時のシャッフルデータ

ドでは，Task を処理する Worker Threads とアクセ量を減らすことは重要であるため，Spark においてはスするデータのメモリ局所性 が乖離した場合，実行テーブルのデータサイズに応じて，ShuﬄeHashJoin

性能に大きな影響を与えることが懸念される．また，と BroadcastHashJoin の二つの JOIN アルゴリズム

SMT により OS 側から認識される仮想コア数は非常が自動的に切り替わる．全データを交換する Shuf-に多くなっており，よりノードの CPU，メモリ構成

ﬂeHashJoin については前節で説明した通りであり，を考慮した効率的なスレッド配置をすることは性能向

通常はこちらが使われる．一方，片方のテーブル

上させる上で重要である．

がかなり小さいとき，例えば 1MB 程度であった場

その他にも考慮すべき点は多く存在するが，本稿で合，BroadcastHashJoin が使われる．小さいテーブ

はこれらのレイヤでの最適化，および最適なパラメー

ルデータを一度ドライバ側に保存した後，これを全タを次節にてそれぞれ説明する．

Executor JVM に配られるが，テーブル A はそれぞれの Worker Threads が個々のファイルシステムか

3 SQL クエリの最適化

らロードするためデータ転送をする必要がなくなり，

3. 1 一時テーブルの削除

大幅にシャッフルデータ量を削減することができる．

オリジナルの Hive クエリの中には，一時テーブル(図 1：右)を作成し，その結果に対して新たなクエリ操作をする

サブクエリの結果との結合については，自動的にものが存在した．これはサブクエリが使えなかったたは BroadcastHashJoin を用いた JOIN には変換されみでノード内の全コアを利用する設定で起動される．
正確には，アプリケーションサブミット時の引数でそのアプリケーションに必要なコア数や Executor

JVM のメモリサイズを動的に確保して JVM を起動することが可能であるが，今回，後述する NUMA
ノードの構成を考慮して Worker Threads をノード内

CPU コアにバインドさせるため，Standalone モードのノードマネージャに登録する Executor JVM 数
図1

ShuﬄeHashJoin と BroadcastHashJoin の処
理の流れなどを予め決定して設定しておく．
ノード内で複数の Executor JVM を生成することは，ノード内であっても Executor JVM 間でのデーないが，得られたクエリ結果に対して LIMIT 句をつ

タ通信 (シリアライズ・デシリアライズ) が発生するけることでジョブを分割することができ，明示的にため，デメリットしかないように思われるが，ノード

BroadcastHashJoin を用いることが可能となる．サ内で用いる Worker Threads とヒープメモリを固定

ブクエリ結果の行数とデータサイズが予め分かっていする場合，1 Executor JVM あたりのヒープサイズ

る，かつ，そのサイズが小さい場合この手法は有効でによって GC Pause Time も変化する．例えば，巨大あるが，例えばサブクエリ結果が 100MB 程度である

すぎるヒープを JVM に割り当てる場合，GC Pause

場合，むしろ遅くなってしまうため，LIMIT 句をつTime は一般的に大きくなる．さらに同じ Executor

けるサブクエリには注意を払う必要がある．

JVM 内の Worker Threads であっても，NUMA ノー

JOIN 順序の変更などは，基本的にはクエリオプ

ドをまたぐように配置されるとメモリアクセスコス

ティマイザによって自動的に最適化されるものであトによって性能が変わる．これらの影響を踏まえて，り，Spark SQL にも spark.sql.codegen オプションを

Executor Threads の数は 1 物理コアにつき 2 つに制有効にすることで最適なクエリプランを生成するオ

限し，異なる Executor JVM 数であっても合計で 48

プションが存在する．しかしながら，今回実験に用いWorker Threads となるように調整した上で，複数のたクエリにおいてこのオプションを有効にした場合，Executor JVM を用意した場合にどのような性能変実行時エラーとなってしまうものが多かった．現在開化が見られるかを調べていく．

発中の機能であるため，将来的にこれらの最適化の一部は自動的に適用される可能性はあるが，本稿ではこ4. 2 Executor JVM のオプション

れらのオプションを使用していない．

spark.executor.extraJavaOptions にセットすると，
任意の JVM オプションを Executor JVM に追加可能

4 Spark オプションの最適化である．Executor JVM では多数の Worker Threads

4. 1 Spark Executor JVM 数の調整が起動するほか，GC や JIT，各種モニタなど JVM

Standalone モ ー ド で は ，spark-env.sh 内 で い由来の Threads，Spark の内部動作に必要な Threads

く つ か の 環 境 変 数 を セット す る こ と が で き ，が存在するため，例えば，GC アルゴリズムやヒープ

SPARK WORKER INSTANCES で 1 ノードあたり

サイズの調整，GC Threads 数の調整，不要な Proﬁl-の Executor JVM 数を，SPARK WORKER CORES

iing Threads の停止など多岐に渡る．本稿では，GCで 1 Executor JVM あたりの Worker Threads 数を，や JIT のログを元にして System.gc() を止めるオプ

SPARK WORKER MEMORY で 1 Executor JVM

ションと GC Threads 数の調整などを行った．

あたりのヒープメモリサイズを設定することができ，何も設定しない場合，Executor JVM はただ１つの図2

クエリ書き換えによる性能比較

図3

NUMA 設定を行ったときの性能比較

Hive 0.13 (Spark 内部でサポート) を用いた．

5 NUMA を考慮した最適化

TPC-H で提供された DBGEN を用いて Scale Fac-

各 Executor JVM では複数の Worker Threads が

tor 100 (100GB) のデータセットを作成し，Hive Table

動作するが，これらの Worker Threads は，OS 側のは Parquet 形式で保存され snappy 形式で圧縮した

スレッドスケジューリングポリシーと，JVM のスレッ

後，Hive Table にロードする．HDFS のブロックサ

ドポリシー，そして Spark ランタイムでの Task 実行

イズは 128MB である．TPC-H クエリは，2.3 節でのローカリティポリシーなどが相互に関係する．それ述べた通り，github 上に公開されている Hive 用のクぞれが互いのステータスを全て把握した上で実行する

エリをベースとする．

コアを決定するわけはないため，複数の NUMA ノー以下の実験では，特にことわりがない場合，ノード

ドに JVM 内のスレッドが分散してしまう場合，メモ内で用いるトータルの Worker Threads は 48，ヒー

リアクセスの遅延が発生し性能低下を引き起こす可能プメモリは 192GB とし，Executor JVM 数は 8 とす性がある．Linux では numactl コマンドを用いるこる．このとき，個々の Executor JVM では 6 Workerとでプロセスに対する CPU アフィニティを設定する

Threads が動作し，ヒープサイズは 24GB である．

ことができ，Executor JVM を numactl コマンド経由で起動して，Executor JVM が使える CPU を制限

6. 2 クエリ書き換えによる性能

することで，Executor JVM 内で動作する Threads

まずはじめに，ベースラインからクエリを書き換が異なる NUMA ノードで実行されるのを防ぐことをえたときの性能を測定する．図 2 は書き換えた 11 個

行う．

のクエリの実行時間を比較したものであり，多くのクエリにおいて 20%程度性能が向上することを確認

6 評価

した．Query 11 も書き換えを行っているが，ベース

6. 1 実験環境

ラインのクエリが 100GB データセットでの実行に

3, 4, 5 節で述べたチューニング方法の効果を調べる

失敗するため，図からは除いている．Query 11 ではため，本稿では 3.3GHz の IBM POWER8 プロセッ

BroadcastHashJoin が使われていたが，本来 Broad-

サ (12 コア) を 2 ソケット (合計 24 コア) 搭載した

cast するのに適さない大きなテーブルを Broadcast

1 ノードで実験を行った．メモリは 1024GB，ディスしていたため，非常に長い時間がかかるクエリとなっクは 1TB，OS は Ubuntu 14.10 (kernel: 3.16.0-31-ていた．JOIN 順を適切に入れ替えることで，小さgeneric) であり，SMT8 モードで動作しているため，

いテーブルを Broadcast することができるようにな

OS 側から認識されるトータルのコア数は 192 である．

り，100GB データセットでも実行可能となった．別ソフトウェアは，Java は IBM J9 JVM (1.8.0, SR1

途 10GB のデータセットを用意し Query 11 の性能比

FP10)，Scala 2.10.4，Spark 1.4.1，Hadoop 2.6.0，

較を行ったところ，20 倍以上高速化されることを確図4

Executor JVM 数を変えたときの性能比較

図5

JVM Option を追加したときの性能比較認した．Query 9 では，JOIN の順序が変わることで

6. 4 Executor JVM 数による性能変化

シャッフルされる総データ量が 48.4GB から 36.3GB

次に Executor JVM の数を調整したときの性能をまで削減可能であり，その結果，21%程度性能が向上

測定する．1 ノードで用いる Execuor JVM 数を 1, 2,

した．その一方であまり変化のないクエリも確認で

4, 8, 12, 24 と変えていくとき，トータルの Worker

きる．Query 7 や 15 ではクエリ書き換えにより一時

Thread を 48，ヒープサイズを 192GB に固定して，テーブルを削除してサブクエリ内で処理するようにし個々の JVM に均等に割り振るように設定する．また，たが，サブクエリの結果がそもそも小さく，処理段数

個々の JVM に対して NUMA の設定を行い，1 物理が増えることのオーバーヘッドによりトータルで見たコアにつき 2 つの Worker Threads が割り当たるよう実行性能は書き換え前と同程度になっている．また，にする．図 4 は，Query 1, 5, 9, 13 に対し Executor

Qurry 21 では，クエリの書き換えによりシャッフル

JVM 数を変化させ，それぞれのクエリの 1 Executor

データサイズが減る一方，個々の Executor JVM で

JVM のときの結果を基準としたときの性能変化を示の GC に一時停止時間が多くなっており，結果とししている．全てのクエリにおいて，複数の JVM をてクエリ書き換えの効果が相対的に小さくなってし用いたときのほうが性能が良いという結果が得られまっている．しかしながら，これは JVM のオプショ

た．シャッフルデータ量が少ないクエリ (Query 1) にンを変更することで改善することができ，この結果に関してはあまり性能変化は見られないが，シャッフル

ついては 6.5 節にて言及する．

データ量が多いその他のクエリに関しては，だいたい20%程度性能が改善した．
6. 3 NUMA 設定による性能
次に 5 節で述べた NUMA の設定を行った場合の

6. 5 Executor JVM オプションによる性能変化

性能を測定する．Executor JVM が利用するコアを

これまでの結果を踏まえてシャッフルデータ量が多い

numactl を用いて個々の NUMA ノードにアライン

クエリの GC ログを分析したところ，実行時間の 15 -するよう設定し，メモリのインターリーブ設定は行30%程度が GC に費やされおり，また，Global GC がわず，2.1 節で行った書き換え後のクエリを用いて，数多く呼び出されていることを確認したため，Execu-

numactl でコアを指定をする場合としない場合の性

tor JVM に対してユーザー側から System.gc() が呼び能を比較したときの結果を図 3 に示す．全クエリで出されても Global GC を起こさないオプションを追加

numactl を用いた場合に 10% - 15%程度性能が向上した．さらに，数あるオプションの中から性能向上に有することを確認した．とりわけ実行時間の短いクエ

効であるものを探索し，結果として次のようなオプショ

リよりも長いクエリや，Executor JVM 間でのシャッンを Executor JVM に追加した．(-Xnocompactgc -フルが多く発生するようなクエリにおいて効果が見

Xnocompactexplicitgc -Xdisableexplicitgc -Xmn12g

られた．

-Xgcthreads12 -Xtrace:none -Xnoloa -Xgc :tlhMax-

7 関連研究
Spark の公式サイト†4 などに様々なチューニング
に関するガイドがいくつか掲載されている．文献 [5]では，Garbage Collection のチューニングにフォー
カスが当てられており，OpenJDK 上で G1GC アル
ゴリズムを用いた場合の最適化について述べられているが，我々の知る限り，Spark 上の Executor JVM
図6

全ての最適化をおこなった後の性能比較

オプションや JVM 数のチューニング，システムレベ
ルの最適化について論じている文献についてはまだimumSize=32768 -XlockReservation -Xdump :sys-

存在していない．

tem:none -Xdump:heap:none) 図 5 は，このときのSpark はクラスタ環境で動作することが想定され結果を示したものであり，シャッフルが多い Query 5,ているため，ネットワークやシャッフルについての最9, 13 においては，20 - 30%程度性能が向上すること

適化についての研究がいくつか存在する．文献 [3] でを確認した．

は，シャッフル時にファイル数が Mapper と Reducerの数の積のオーダーで増加していき通信のオーバー

6. 6 全オプション適用時の性能比較

ヘッドが増加するのを防ぐため，Reducer 数と同等と以上の結果を踏まえて，今回行った全ての最適化

なるようシャッフルデータと統合するためのオプショを適用した後の全クエリの実行性能を比較したもの

ンについて論じている．本稿でもこのオプションはを図 6 に示す．ベースラインはクエリ書き換えをせ常に有効化した．また文献 [6] では，Inﬁniband 上でずに，8 Executor JVMs，6 Worker Threads/JVM，RDMA を Spark の Shuﬄe 通信に用いることで，既24GB/JVM のヒープを割り当て，NUMA や JVM

存の NIO や Netty ベースの Shuﬄe 通信を IP over

オプションを追加しないときの結果である．ほぼ全

IB ネットワークで実行するよりも高速に実行できるてのクエリにおいて実行性能が 20 - 40%高速にクエことを示している．

リが実行できることを確認した．一方，実行時間がSpark SQL 上での TPC-DS の性能測定結果は，短いクエリ (Query 6, 14, 15) に関しては，ほぼ変

Databricks などの Developers Blog などで性能を測わらないか若干性能が悪くなる傾向が観測された．

る指標として用いられているが，TPC-H に関して言その他のクエリ向けに多数の Reduce タスクに細分及されている文献は少ない．文献 [4] では，Hive on

化してヒープメモリを圧迫しないようにするため

MR, Hive on Tez, Impala など SQL on Hadoop な

spark.sql.shuﬄe.partitions をデフォルトの 200 から

システム上で Parquet や ORC などのカラムナーベー

400 程度まで増加しているが，タスクの計算時間が少スのテーブルや，snappy の圧縮を適用した場合など

ない場合，計算時間よりもオーバーヘッドのほうが多の性能について TPC-H のクエリを題材に論じている

くなってしまうため，全体の実行時間が短いクエリにが，Spark に関しては行っていない．

おいてはこのオーバーヘッド影響により遅くなる結果となった．

8 まとめと今後の課題
本稿では，TPC-H Benchmark を題材として，ア
プリケーションレベルから Spark や JVM，OS レベ
ルのチューニングによって，TPC-H クエリの実行性†4 http://spark.apache.org/docs/latest/conﬁguration.html

能がどの程度改善できるかを定量的に示した．その結果，クエリの書き換えにより 20%程度，NUMA の設定を加えることでほぼ全てのクエリで 10%程度，

Executor JVM 数を調整することで 20%程度，JVM
オプションのチューニングにより 20-30%程度性能が向上した．全てのチューニングを適用させた場合の性能は，何もチューニングを行わない場合と比べて，最大で 40%程度性能が改善することを確認した．今後の課題としては，クエリの傾向ごとに最適なチュー
ニングは異なっているため，クエリごとの特徴を踏まえてより最適な設定を行うことや，これらの知見をSpark ランタイムやスケジューラに適用し，コミュニ
ティにフィードバックしていくことを考えている．

参 考 文 献[1] Armbrust, M., Xin, R. S., Lian, C., Huai, Y., Liu,D., Bradley, J. K., Meng, X., Kaftan, T., Franklin,
M. J., Ghodsi, A., and Zaharia, M.: Spark SQL:
Relational Data Processing in Spark, Proceedings ofthe 2015 ACM SIGMOD International Conference
on Management of Data, SIGMOD ’15, New York,NY, USA, ACM, 2015, pp. 1383–1394.
[2] Chu, C. T., Kim, S. K., Lin, Y. A., Yu, Y., Bradski, G. R., Ng, A. Y., and Olukotun, K.: MapReduce for Machine Learning on Multicore, NIPS,

Schölkopf, B., Platt, J. C., and Hoﬀman, T.(eds.),
MIT Press, 2006, pp. 281–288.
[3] Davidson, A. and Or, A.: Optimizing Shuﬄe
Performance in Spark, Technical report, Universityof California, Berkeley - Department of ElectricalEngineer- ing and Computer Sciences, Tec Rep.,
2013.
[4] Floratou, A., Minhas, U. F., and Özcan, F.:
SQL-on-Hadoop: Full Circle Back to Sharednothing Database Architectures, Proc. VLDB Endow., Vol. 7, No. 12(2014), pp. 1295–1306.
[5] Kaczmarek, E. and Yi, L.: Taming GC Pausesfor Humongous Java Heaps in Spark Graph Computing, Spark Summit 2015, 2015. https://sparksummit.org/2015/events/taming-gc-pauses-for-humongousjava-heaps-in-spark-graph-computing/.
[6] X. Lu, W. Rahman, N. I. D. S. D. P.: Accelerating Spark with RDMA for Big Data Processing:
Early Experiences, August 2014.
[7] Xin, R. S., Rosen, J., Zaharia, M., Franklin,
M. J., Shenker, S., and Stoica, I.: Shark: SQL
and Rich Analytics at Scale, Proceedings of the 2013
ACM SIGMOD International Conference on Management of Data, SIGMOD ’13, New York, NY,
USA, ACM, 2013, pp. 13–24.
[8] Zaharia, M., Chowdhury, M., Das, T., Dave, A.,
Ma, J., McCauley, M., Franklin, M. J., Shenker,
S., and Stoica, I.: Resilient Distributed Datasets:
A Fault-tolerant Abstraction for In-memory Cluster
Computing, Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation, NSDI’12, USENIX Association, 2012,
pp. 2–2.