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

Apache Spark におけるコード生成の最適化
石崎一明

Abstract: Apache Spark は分散メモリデータ構造を用いてインメモリで並列計算を行う，並列分散処理フレームワークである. 用意された並列処理可能な操作を組み合わせることでプログラムを記述し実行することで，大規模なデータ処理を高速に行うことができる．Apache Spark では，ユーザが Scala で記述したプログラムに対して，最適化を適用し Java コードを生成して実行する機構が存在する．本稿では，配列を扱う Scala プログラムから生成された Java コードにおける，冗長なデータ変換，非効率なデータ表現，による問題を指摘し，その原因を特定する．さらに，原因の解決方法を述べ，改善後の生成された Java コードを示す．

生成された Java コードで発生する不必要なボックス

1 はじめに化・ボックス化解除による性能上の問題を指摘する．

Apache Spark [1] [5] は分散メモリデータ構造を用

数値計算・機械学習のプログラムでは配列が使用されいてインメモリで並列計算を行う，Scala 言語で記述るため，この性能問題は重要である．さらに，この問された並列分散処理フレームワークである，Apache題を引き起こす原因である，冗長なデータ表現形式のSpark が用意した並列処理可能な操作を組み合わせる変換，生成されたコードで使用される非効率なデータことで記述したプログラムを実行することで，大規模表現，を特定する．これらの原因の解決方法を述べ，なデータ処理を容易にかつ高速に行うことができる．

改善された生成 Java コードを示す．

Apache Spark は，SQL・機械学習・グラフ処理など

本稿では，下記の貢献を示す．のライブラリが用意しており，プログラマはライブ

1. 並列処理可能な操作の中で配列を扱うプログラ

ラリが提供する関数を組み合わせて呼び出すことで，ムをコンパイルしたコードで発生する性能上のデータ分析のプログラムを簡単にを記述可能である．

問題点である，不必要なボックス化・ボックス化

Apache Spark には，プログラマが Scala を用いて解除の指摘並列処理可能な操作を組み合わせて記述したプログラ

2. 問題の原因である，冗長なデータ表現形式の変ムに対して，実行時に最適化を適用し Java コードを換，非効率なデータ表現，の特定と解決手法の再生成する機構が存在する．Java 仮想マシン上では，提案再生成された Java コードから変換された Java 仮想機械バイトコードが実行される．この機構を用いて，

2 関連研究ある種のプログラムでは，高い性能が得られることが与えられた SQL クエリから，実行可能なコードを示されている [4]．

再生成する研究が近年行われている [2] [3]．この機構

本稿では，並列処理可能な操作の中で配列を扱うプを用いることによって，再生成されたコードは既存のログラムに対してこのコード生成機構を適用した際，コンパイラによるプログラム最適化・対象ハードウェ

Kazuaki Ishizaki, 日本アイ・ビー・エム株式会社 東京
基礎研究所, IBM Research - Tokyo.

ア向けの最適化，が適用され，高速化可能なことが示されている．しかし，SQL クエリーで用いられることの少ない，配列を扱うクエリーのコード生成については議論していない．

3 研究の動機
モリデータを表現する構造体が存在する.DataFrame
に対して並列処理可能な SQL 操作（例えば filter,join な ど ）を 適 用 し て 新 た な

DataFrame を生成することで一連の処理を記述す
る．Apache Spark では Java コードを再生成可能な操作が２つ以上の連続することを検出した場合, 新たな DataFrame の生成を行わない，最適化された Javaコードを生成する．図 1 のプログラムは，このコード生成機構により，図 2 のプログラムに変換される．
図 2 のプログラムはコメントに示すように，下記の７つのステップから構成される．図 2 のプログラムでは，DataFrame

df2 を生成することなくスカラ値input value を用いた，最適化された Java コードが生成される．

1. 行の読み込み

df3.show

図 11

Apache Spark では，DataFrame と呼ばれる分散メaggregation,4

DataFrame を用いた Scala プログラム例

class GeneratedIterator {2Iterator inputIterator = ...;

3

Row projectRow = ...;

4RowWriter rowWriter = ...projectRow...;

5

protected void processNext() {6while (inputIterator.hasNext()) {78// 1. 行の読み込み

9Row inputRow = (Row) inputIterator.next(); // for df1

10// 2. 各列の要素の読み込み

11

int input_value = inputRow.getInt(0);12// 3. 4 が扱うデータ形式への変換

13// 4. 生成されたコードによる計算処理

14

if (!(input_value > 0)) continue;

15// 3. 4 が扱うデータ形式への変換

16// 4. 生成されたコードによる計算処理

17

int project_value = input_value + 1;

18// 5. 6 が扱うデータ形式への変換

19// 6. 各列の要素の書き込み

20

rowWriter.write(0, project_value);

21// 7. 行の書き込み
append(projectRow); // For df322}

23}

2425}

2. 行内の，各列の要素の読み込み

図 2

図 1 から生成された Java コード例

3. 各要素について，4. の処理で扱うことが可能なデータ形式への変換

4. 生成されたコードによる，計算本体の処理
5. 6. の処理で扱うことが可能なデータ形式への変換

6. 行内の，各列の要素の書き込み
7. 行の書き込み
図 1 の書き込み先を配列に変更した図 3 のプログ
ラムは，コード生成機構によって，図 4 のプログラムに変換される．元のコードは，スカラ値を配列に書き込む処理を行う単純なプログラムであるが，Lines1

val df1 = sparkContext.parallelize((-N to N), 1).toDF("x")

2val df2 = df1.filter("x > 0")3val df3 = df2.selectExpr("Array(x + 1)")4df3.show

図 31DataFrame で配列を扱う Scala プログラム例

class GeneratedIterator {2Iterator inputIterator = ...;

3

Row projectRow = ...;

4

Row projectRow = ...;

5

RowWriter rowWriter = ...projectRow...;

6

ArrayWriter arrayWriter = ...projectWriter...;7
89

protected void processNext() {while (inputIterator.hasNext()) {19-24, 30 で，3 章で述べるように非効率な処理が行10// 1. 行の読み込みわれていることが分かった．この非効率な処理によっ

11Row inputRow = (Row) inputIterator.next(); // for df1

12// 2. 各列の要素の読み込み

13

int input_value = inputRow.getInt(0);14// 3. 4 が扱うデータ形式への変換

15// 4. 生成されたコードによる計算処理

16

if (!(input_value > 0)) continue;

17// 3. 4 が扱うデータ形式への変換

18// 4. 生成されたコードによる計算処理

19

Object[] tempArray = new Object[1];20

if (false) { tempArray[0] = null; }て，性能が低下するであろうことは明らかである．この非効率な処理を取り除いた Java コードを生成することが，本研究の目標である．
1val df1 = sparkContext.parallelize((-N to N), 1).toDF("x")

2val df2 = df1.filter("x > 0")3val df3 = df2.selectExpr("x + 1")

else { tempArray[0] = input_value + 1; }21
22// 5. 6 が扱うデータ形式への変換

23ArrayData project_value =new GenericArrayData(tempArray);24
25// 6. 各列の要素の書き込み

26

int n = project_value.numElements();

27

for (int i = 0; i < n; i++) {if (project_value.isNullAt(i)) {28

arrayWriter.setNullInt(i);29} else {30
31

int element = project_value.getInt(i);32arrayWriter.write(i, element);

本章では，前章で述べた問題点の解決方法を示す．
さらに，その結果生成されるコードを示す．
１つ目の問題の解決のために，Line 19-21 において int 型のスカラ値から Object[] 型への代入が起きている理由を調査した．まず，Object[] への代入が必要となる理由は，配列が SQL の null 値を保存する必要があるためであった．生成されたコードで}

33

5 提案手法34}

35...

36// 7. 行の書き込みを，Object 型要素に null かボックス化された値で37append(projectRow); // For df3保持してる．一方，コード生成機構内部では，行の各}38列が格納する値の型を保持している．現在着目して

}

3940

は，配列の各要素が null 値または通常の値を持つか}いる列の型を調べると，「int 配列で，各要素は null
図 4

図 3 から生成された Java コード例

値を取らない」，となっていた．つまり，型情報は正確に保持していたにも関わらず，代入文のコード生成

4 問題点

時に null 値を保存可能な保守的コードが行われてい本章では，配列を扱う DataFrame プログラムから，た．この結果，ボックス化を発生するコードが生成さ

Apache Spark のコード生成機構を用いて Java コーれていた．この場合は，型情報に従って配列の各要ドを再生成した際に Java コード内で生じる，ボック素は null 値を取らないことが保証されているので，ス化・ボックス化解除の発生，の問題について述べる．

int[] 型へ代入するコードを生成する最適化によっ

図 3 の Lines 19-21 で は ，ス カ ラ 値 に 1 を 足て，ボックス化を除去可能である．

し た 値 を Java の Object 配 列 に 代 入 し て い る ．

GenericArrayData ク ラ ス は ，各 列 の 要 素 の 書Line 21 の 代 入 文 で は ，右 辺 は int 型 で ，左 辺

き 込 み の 際 に 使 用 さ れ る デ ー タ 構 造 で あ る ．現は Object[] 型であるため，型変換が必要になる．

在 の 実 装 で は ，保 持 す る 配 列 の 型 に か か わ ら ずJava bytecode に 変 換 す る 際 に ，tempArray[0] =GenericArrayData クラスが使用される．内部では，Integer.valueOf(input value + 1) というコード全ての Java の値が格納可能な Scala の Any 型の配列に変換して，型変換を行っている．この valueOf メが用いられている．コード生成機構内部では，行の各ソッドの中では Integer オブジェクトが生成され，列が格納する値の型を保持しているため，保持する配ボックス化が発生している．

列の型に基づいて特殊化されたクラスを使用する最GenericArrayData クラスは，各列の要素の書き適化が可能である．この場合は，int[] 型配列がコン込みの際に使用されるデータ構造であり，Scala でストラクタに渡されるので，int[] を内部で保持する記述されている．コンストラクタに渡された配列は特殊化された GenericIntArrayData クラスを用意Array[Any] に格納する実装となっている．図 3 のする．その結果，配列要素を取り出す際のボックス化

Line 23 では，GenericArrayData クラスのコンス解除を除去可能である．

トラクタには，Integer[] 型のオブジェクトが渡さ

この２つの提案手法により，図 3 のプログラムかられている．図 3 の Line 30 で，配列要素を取り出す生成された Java コードを図 5 に示す．図 5 のプログ

getInt() を呼び出す際に，Integer 型から int 型へ

ラムでは，Line 14 でボックス化の発生がなく，Lineのボックス化解除が発生している．

16 では int[] 型の配列として内部で保持され，Line22 ではボックス化解除が発生しない，最適化された

Java コードとなっている．
1class GeneratedIterator {2Iterator inputIterator = ...;

3Row projectRow = ...;

4Row projectRow = ...;

5RowWriter rowWriter = ...projectRow...;

6

ArrayWriter arrayWriter = ...projectWriter...;7

protected void processNext() {8while (inputIterator.hasNext()) {9されたコードで使用される非効率なデータ表現，を特定した．これらの原因の解決方法を述べ，改善された生成 Java コードを示した．数値計算・機械学習のプログラムでは配列を扱うため，この性能問題は重要であると考えている．この問題解決方法は，すでにPull request として github リポジトリに投稿済みである．今後は，より実用的なプログラムにこれらの問

10

Row inputRow = (Row) inputIterator.next();

題の影響を調査するとともに，Dataframe 以外の構

11

int input_value = inputRow.getInt(0);造体を用いたプログラムについても同様の問題を調12if (!(input_value > 0)) continue;

13

int[] tempArray = new int[1];14tempArray[0] = input_value + 1;

15

ArrayData project_value =16

new GenericIntArrayData(tempArray);17

int n = project_value.numElements();

18

for (int i = 0; i < n; i++) {if (project_value.isNullAt(i)) {19arrayWriter.setNullInt(i);20} else {21int element = project_value.getInt(i);22arrayWriter.write(i, element);

23}

2425}

26...
append(projectRow); // For df327}

28}

2930}

図 5

提案手法によって生成された Java コード例

6 まとめ
本稿では，並列処理可能な操作の中で配列を扱うプログラムに対して，Apache Spark が用意するコード生成機構を適用した際，生成された Java コードで発生する不必要なボックス化・ボックス化解除による性能上の問題を指摘した．さらに，この問題を引き起こす原因である，冗長なデータ表現形式の変換，生成査したい．

参 考 文 献[1] Apache Software Foundation: Apache Spark,
http://spark.apache.org/, 2016. Accessed: 201608-12.
[2] Diaconu, C., Freedman, C., Ismert, E., Larson,P.-A., Mittal, P., Stonecipher, R., Verma, N., andZwilling, M.: Hekaton: SQL Server’s Memoryoptimized OLTP Engine, Proceedings of the 2013
ACM SIGMOD International Conference on Management of Data, SIGMOD ’13, New York, NY,USA, ACM, 2013, pp. 1243–1254.
[3] Neumann, T.: EﬃcientlyCompiling Eﬃcient
Query Plans
forModern Hardware, Proc.
VLDB Endow.
[4] Sameer Agarwal, D. L. and Xin, R.: Apache
Spark as a Compiler: Joining a Billion Rows
per Second on a Laptop, https://databricks.com/
blog/2016/05/23/apache-spark-as-a-compilerjoining-a-billion-rows-per-second-on-a-laptop.
html, 2016. Accessed: 2016-08-12.
[5] 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, Berkeley, CA, USA, USENIX
Association, 2012, pp. 2–2.