DEIM Forum 2014 D1-3

列指向バッファ管理を用いた MapReduce の高速化
小沢 健史†

及川 一樹†

鬼塚真†

本庄利守†

† 日本電信電話株式会社 〒 180–8585 東京都武蔵野市緑町 3–9–11
E-mail: †{ozawa.tsuyoshi,oikawa.kazuki,onizuka.makoto,honjo.toshimori}@lab.ntt.co.jp
あらまし

蓄積された大量のデータを安く，実用的な速度で処理するために，安価な計算機クラスタを用いて効率的に分散計算を行うフレームワークに注目が集まっている．特に，MapReduce は最も普及している分散処理フレームワークの 1 つであり，Google, Facebook, Yahoo! をはじめとする多くの企業に利用されている．MapReduce では，処理の耐故障性を確保するため，チェックポイントとして中間出力をディスクに保存するが，既存の MapReduce 処
理系ではそのファイルレイアウトとして圧縮効率が悪いレイアウトを採用していた．本稿では，MapReduce における出力形式として圧縮効率の高い列指向バッファ管理を用いることで，MapReduce 上で動作するジョブを高速化できる可能性を示す．
キーワード

MapReduce, Hadoop, 列指向ストレージ

1. は じ め にHadoop [1] では， gzip，bzip2，LZO，LZ4，Snappy などがサポートされており，ワークロードに応じて有効な圧縮コーデッ

記憶装置・計算機・インターネットの発展により，産業界にクを選択することが可能である．2 つ目は，MapReduce アプ

おいて取り扱うデータの量が爆発的に増加しつつある．蓄積さリケーションレベルで Reducer への負荷を減らすよう工夫すれた大量のデータを安く，実用的な速度で処理するために，安る方法である [12]．これらの手法はデザインパターンとして命価な計算機クラスタを用いて効率的に分散計算を行うフレーム名されており，その代表例は以下の通りである:ワークに注目が集まっている．
特に，MapReduce は最も普及している分散処理フレームワー（ 1 ） 集約処理において Map 関数の中で事前集約する in-mapper combining パターン．

クの 1 つであり，Google, Facebook, Yahoo! をはじめとする（ 2 ） 行列計算において中間結果のレコード数を減らす

多くの企業に利用されている [2], [6], [16], [18]．MapReduce でstripes パターン．

は，並列処理をおこなう Map 関数と，Map 関数の結果を集約（ 3 ） 表の結合を行う際に，小さいテーブルをあらかじめ全する Reduce 関数を記述するだけで，計算機の故障や，計算機ての計算機に配布しておき，Map 関数のみで処理を完結させ間の同期処理を意識することなく，大規模な分散処理を行うこる memory-backed join パターン．とができる [6]．

いずれも，Map 関数を実行する MapTask のメモリを有効利

MapReduce 処理系の強みの一つとして，スキーマ定義のな用し，Map 関数の出力量を減らすことが共通点である．これ

い大量のデータに対して高いスループットで処理を行うことがらの方法を用いることにより．ナイーブな MapReduce プログできることが挙げられる．この特性により，従来の分散データラムよりも高い性能を示すことが報告されている [12]．

ベースでは処理しきれなかった量の生データを一箇所に集約しこれらの方法には，それぞれ欠点がある．まず，1 つ目の圧ておき，ETL を行った上で RDBMS などにデータをインポー

縮を用いる方法は，圧縮効率が限られている．これは，Hadoop

トするといった利用事例も出てきている ( [4], [13])．の中間出力ファイルの構造が，列指向形式と比較し圧縮と相性

MapReduce には，タスクを実行するために 2 種類のプロが悪い行指向形式を採用している [17] ことに起因する．行指向

セスがある．1 つは Mapper ，もう 1 つは Reducer である．

形式では，異なる型のデータが連続して配置されるため，圧縮Mapper は Map 関数を実行する MapTask を処理し，Reducer時のエントロピーが増大してしまい，圧縮効率が上がらない．

は Reduce 関数を実行する ReduceTask を処理する．MapRe-

一方で，列指向形式は同じ型のデータが連続して配置されるた

duce で行われる処理の中には，Reducer が 1 つであることをめ，圧縮時のエントロピーが抑えられ，圧縮効率が向上する傾強いる処理がある．例えば，WordCount を行った後に，頻度向にある．2 つ目の MapReduce アプリケーションレベルで負の多い単語順にソートする処理などである．このような処理の荷を減らす方法は，ユーザがメモリ管理を自前で行うという特場合，全ての MapTask の結果が 1 つの Reducer へ転送され性上，ユーザ自身がメモリ不足に注意を払う必要がある．また，るため，Reducer が行う処理が増大し，過負荷になってしまい

アプリケーションごとにユーザが適切な最適化を選択し，その処理に時間がかかるという問題がある．

都度実装を行う必要がある．

この問題の対策は 2 つある．1 つ目は，圧縮を用いて Map-そこで本稿では，MapReduce フレームワーク側でのバッファTask の出力を圧縮することで Reducer の IO 負荷を軽減する管理を工夫することである高速化を行う列指向バッファ管理モ方法である．MapReduce のオープンソース実装の 1 つである

ジュールを提案する．列指向バッファ管理モジュールは，中間MapTask
678

9:;<<=>

Map関数?>@;A>

バッファ管理モジュール
リングバッファ45#$%*+,-./
0123!"#$%

外部記憶装置&'()

図 1 MapReduce の概要

IFIle

IFIle

出力のバッファ管理方式として圧縮効率の良いデータ構造を用いることで，中間出力ファイルサイズの削減が可能である．

図 2 Hadoop におけるバッファ管理の概要

提案手法を実装して評価を行った結果，提案手法におけるバッファ管理モジュールが，既存の MapReduce 処理系におけるバッファ管理方式と比較し，同じ圧縮アルゴリズムを用いて中間ファイルの大きさを 25%削減できることを確認した．また，提案手法による中間ファイルの作成のオーバヘッドが小さく，MapReduce ジョブを高速化できる可能性を確認した．

IFile
Key1の長さ(1-5バイト)(Key1, Value1)(Key2, Value2)

Value1の長さ(1-5バイト)(Key2, Value3)

Value1 のバイト列

以下，2 章にて，MapReduce の基本的な挙動とその OSS 実
装である Hadoop で用いられているバッファ管理モジュールのKey1 のバイト列(Key3, Value4)...

問題点について述べる．3 章にて，既存のバッファ管理の問題点を解決した列指向バッファ管理モジュールを提案する．4 章
では，実験による圧縮率の評価を示す． 5 章で関連研究について述べ，6 章で本論文をまとめる．

2. 前 提 知 識

図 3 Hadoop における Shuﬄe フェイズの際に用いられる中間ファ
イル (IFile) のファイルレイアウト

2. 1 MapReduce

散ファイルシステム）が構築され，入力データが格納される．

図 1 を用いて，本研究が改善の対象とする MapReduce につ

MapReduce ジョブを開始すると，まずマスタが入力データをいて説明する．MapReduce ジョブは Map フェーズと Reduce

ユーザの指定したサイズで分割し複数の InputSplit を生成すフェーズの 2 フェーズから構成される．Map フェーズでは入る．次にマスタが InputSplit の数だけ MapTask を生成し，各力データ（D）を読み込んで Key/Value ペア (K1 , V1 ) を生成スレーブに割り当てる．MapTask が割り当てられたスレーブし，それを入力として各ペアに対してユーザが定義した Map

では Mapper が起動され，ユーザが定義した Map 関数が実関数を実行し，新たな Key/Value ペアリスト (K2 , V2 ) を中間行されて，Key/Value 形式の中間データが出力される．中間データとして出力する．Reduce フェーズでは，まず中間デー

データは同じ Key 値を持つものが一つのスレーブに集まるよタを Key 毎にグルーピング（Shuﬄe）する．そして各グルーうにネットワークを介して相互に移動（shuﬄe）される．マスプを入力としてユーザが定義した Reduce 関数を実行し，結果タはユーザが定義した数の ReduceTask を生成し，各スレーブとして Key/Value ペアリスト (K3 , V3 ) を出力する．この一連に割り当てる．このとき，中間データは Key 毎に分配される．の流れを式として表現すると，以下のようになる．

ReduceTask が割り当てられたスレーブでは Reducer が起動

D → map(K1 , V1 ) → {K2 , V2 }され，ユーザが定義した任意の Reduce 関数が実行されて，新shuf f le({K2 , V2 }) → {K2 , {V2 }}たに Key/Value 形式の処理結果が DFS に出力される．

reduce(K2 , {V2 }) → (K3 , V3 )2. 2 Hadoop における Shuﬄe 時のデータ構造

MapReduce 処理は複数のノード（計算機）をネットワーク

MapReduce の MapTask の出力に用いられているバッファで相互接続したクラスタ内で行われる．クラスタはリソースの管理の方式は，実装系依存である．そこで，まず MapReduce

割り当ておよびタスクの割り当てを行うマスタノードと計算を処理系の中でも主要なオープンソース実装である Hadoop を行うスレーブノードで構成される．クラスタ上には DFS（分対象にバッファ管理方式の解析を行った．その概要図を図 2 に示す．

Hadoop のバッファ管理モジュールに Map 関数の出力であMapTask

Map関数

る Key・Value のペアが渡されると，バイト列にシリアライズ(Key, Value)された Key オブジェクト，Value オブジェクトがリングバッ

バッファ管理モジュール
列指向バッファ

ファの先頭に書き込まれる，同時に，Key・Value それぞれのメタデータがリングバッファの後尾に書き込まれる．そして，ユーザにより指定されたペア数を超えるか，指定されたサイズを超えると，中間ファイルである IFile として書き出される．中間出力ファイルである IFile のレイアウトを図 3 に示す．IFileは，Key の順番にソートされた Key・Value のペアの配列で構

CIFileの一部の一部
Valueの一部
Keyの一部
Value型1用のバッファ型2用のバッファ型3用のバッファ型4用のバッファ外部記憶装置成されている．1 エントリは Key の長さ，Value の長さ，バイKey...

CIFileト列にシリアライズされた Key オブジェクト，バイト列にシリアライズされた Value オブジェクトで構成される．
このようなデータ構造は，MapReduce 処理系のように事前

図4

列指向バッファを用いたバッファ管理モジュールの構成に処理するデータの型が定義できないシステムでは実装が容

バイト
型の個数)
Type1:1バイト(含まれている型1)
Type2:1バイト(含まれている型2)
Type3:1バイト(含まれている型3)易であるという利点がある．また，IFile の Writer/Reader がnumTypes:1データの型を意識することなく書き込み・読み込みを行うこと(CIFileができるため，疎結合な設計を行うことが可能となる．

ヘッダ
型1用のバッファ型2用のバッファ型3用のバッファ...

一方で，この構造は全ての Key・Value のペアに対してデータサイズを確保するため，レコード数の多い MapReduce タス
クではオーバヘッドが大きくなるという欠点がある．また，タプルごとにデータを保存しておくアプローチは圧縮効率が低いという指摘がある [17]．

2. 3 中間出力書き出しの必要性
MapReduce では，ジョブを構成するプロセスの一部および計算機の一部が故障しても故障した Task のみをやり直すことでジョブを中断することなく実行することができる．
プロセスの故障には MapTask の故障と ReduceTask の故障

...

バイト
型1のバッファ長)
lenType2: 4バイト
(型2のバッファ長)
lenType2: 4バイト
(型3のバッファ長)
lenType1: 4(...の 2 通りがある．MapTask が故障した場合は，分散ファイルシステムから再度データをロードし直し，MapTask をやり直す．ReduceTask が故障した場合は，ディスクに書かれているMapTask の中間出力をコピーしなおして，再度 ReduceTask

図 5

列指向バッファ管理で利用する中間データ (CIFile) のレイア
ウトを実行する．計算機の故障が発生した場合，その計算機が保持している MapTask の中間出力結果は失われてしまう．そのたとめて保存する点である．このようにすることで，同じ型，似め，その計算機で実行された全ての MapTask/ReduceTask をた傾向のデータを一箇所に集めることができるため，圧縮効率やり直すことで，ジョブを続行する．

効率および複数のデータに対する演算の並列度を高めることがHadoop 処理系の中には，遅延を下げるために，Shuﬄe の可能となる [22]，際に中間出力を永続化しないものもある．このようなアプロー

そこで本研究では，特に列指向データベースの圧縮特性にチを取り，MapTask/ReduceTask いずれにおいて故障が発生着目し， 列指向データベースのデータを型ごとに保存する列した場合，全ての処理は MapTask からやり直しとなる [11]．

指向バッファ管理モジュールを提案する．その概要図を図 4

結果として，永続化をしない場合は，永続化をする場合と比較に示す．MapTask の中間データの型情報が残っている段階でし，故障からの復旧や再実行に長い時間を要する．そのため，各型専用のバッファにルーティングを行う．バッファが指定しHadoop は標準で全ての MapTask の出力した中間出力データた大きさを超えたら，その段階でバッファごとに圧縮をかけ，を永続化するようにしている．

CIFile(Columnar IFile) 形式でデータを書き出す．

3. 提 案 手 法

3. 2 CIFile
CIFile フォーマットのレイアウトを図 5 に示す．CIFile は，

3. 1 列指向バッファ管理モジュール

ヘッダとシリアライズされたオブジェクトのバイト列で構成さ

OLAP 向けの読み込み特化型データベースとして，C-Store，れている．

MonetDB を初めとする列指向データベースがある [14], [17]．

CIFile のヘッダはデータ型がいくつ入っているかを表す 1 バ

列指向データベースの特徴は，データの保存形式を同じ型でまイトの変数 numTypes，入っているデータの型を表す 1 バイト

変数の numTypes 個含む配列，各データ型の利用しているバッ

4. 2 CIFile 内における型ごとの圧縮効率の比較

ファサイズを表す 4 バイトの変数を numTypes 個含む配列で次に，CIFile 内における型ごとの圧縮効率の比較を行った．

構成される．CIFile のヘッダの後ろには，型ごとにシリアライ

実験で用いる入力データは，前節同様，PUMA Benchmark

ズされた Key・Value オブジェクトのバイト列が保存される．

Suite で公開されているデータセットのうち，Wikipedia 50GB

4. 評

価

提案手法である CIFile と既存技術である IFile の比較と特性解析を行うため，実験を行った．に含まれているのデータの一部 (file87) である．
表 2 に，CIFile を用いたときの型ごとの圧縮効率の違いを示す．圧縮率は MapTask から出力された型，およびデータの型によって異なるが，Raw の場合，共起頻度計算の場合の両方実験に用いるプログラムは読み込まれた Key・Value のペアにおいて，整数よりも文字列が高い圧縮率を示している．整数をそのまま書き出す MapTask と 単語の共起頻度計算を行う

配列に対しては，p4delta [23] のような整数圧縮専用の圧縮ア

MapTask [12] の 2 種類である．読み込まれた Key・Value ペルゴリズムを用いることにより，圧縮率を向上できる可能性がアをそのまま書き出す MapTask は，TextInputFromat を用ある．

いて改行文字ごとに渡された Key，Value の値を直接中間ファ

4. 3 書き込み速度の比較

イルに書き込むプログラムである．本稿では，このプログラム

次に，CIFile 構築のオーバヘッドを測定するために，Map-を Raw と呼ぶことにする．単語の共起頻度計算は，与えられTask を実行して IFile と CIFile を作成し，ディスクに書き込たテキストに連続して存在する単語のペアの回数を数え上げむ速度の比較を行った．中間データには共起頻度計算の中間デーるプログラムである．たとえば，”An apple is red” という内タを用い，圧縮アルゴリズムには Snappy 圧縮アルゴリズム

容を保持しているテキストがあった場合，”An apple”，”appleを用いた．本実験に用いる入力データは，PUMA Benchmarkis”，”is red” がそれぞれ 1 回ずつ数え上げられる．上記の例でSuite で公開されているデータセットのうち，Wikipedia 50GB

は，探索範囲を広くして ”An apple is”, ”apple is red” と数えに含まれているデータ全体である．なお，ディスクに書き出す上げることも可能であるが，今回は隣り合っている単語のペア際のバッファサイズは 8K バイトであり，ディスク IO と IFileの回数を数え上げることとする．共起頻度計算を行う Mapper

および CIFile 構築は同一スレッドで行っている．

は，入力テキストを空白文字ごとに区切り，Key として隣接す

表 3 に，実験結果を示す．CIFile ファイルサイズは，IFile とる単語を連結して作成した複合キー，Value として 1 を出力し，比較して 21%の削減ができている．これにより，ディスク IO

Reducer はその集計を行うが，今回の実験では中間データファ量が削減され，33%の書き込み速度向上が確認できた．提案手

イルの大きさのみを評価するため， Reducer は利用しない．

法は，バッファにデータを書き込む際に型情報を用いてバッファ中間ファイルの書き出しには，MapReduce の処理系 Hadoopを移動する方式であるため，コピー回数は同じであり，オーバの IFile.Writer クラスと，提案手法の実装である CIFile 用のヘッドが小さい．オーバヘッドと比較し，中間データ書き出し

DataOutput クラスを用いた．

サイズが小さくなることによるディスク IO 削減の恩恵が大き

なお，実験用の計算機には Amazon EC2 の m2.4xlarge を用いた．

4. 1 異なる圧縮アルゴリズムを用いた比較
IFile と CIFile に対し， 異なる圧縮アルゴリズムを用いて圧縮を行い，中間ファイルの大きさを比較した．中間ファイルのいため，書き込み速度の向上が確認できた．
なお，MapReduce ジョブがどの程度高速化できるかに関する評価は，今後の課題とする．

5. 関 連 研 究

書き出しの際には，非圧縮，bzip2 圧縮，Snappy 圧縮を利用

Spark [21] は ，MapReduce よ り も 柔 軟 な 非 循 環 グ ラ フするようにし，比較を行った．なお，bzip2 圧縮には Hadoop(DAG) を DSL で記述することで，機械学習の計算インテの Bzip2Codec クラスを，Snappy 圧縮には snappy-java [15]

ンシブな処理を効率良く実行可能な処理系である．Spark の採を利用した Brisk プロジェクト [5] の SnappyCodec による圧用している RDD [20] は，DSL から処理とメモリ内およびディ縮を用いた．実験で用いる入力データは，PUMA Benchmark

スク内に保存されている中間データの依存関係を把握すること

Suite [7] で公開されているデータセットのうち，Wikipediaで，復旧を効率良くにおこなうことができる仕組みである．ま50GB に含まれているのデータの一部 (file87) である．このた，Shark [19] は Spark 上で動作する SQL 処理系である．そデータは 8355840 バイト・約 28 万行のテキストファイルでの他の列指向 DB 同様，メモリ上に載るデータを増やすためにある．

様々な圧縮方式を用いる．Spark・Shark ともに，中間データ

表 1 に，IFile と提案手法である CIFile において比較した結のデータ構造に関する言及はなく，本研究とは補完的な関係に果を示す．IFile が Key・Value のペアごとに長さを付与するのある．つまり，提案手法を RDD のデータ構造に採用することに対し，CIFile は，バッファごとに長さを付与するため，提案で，Spark/Shark を高速化することも可能である．

手法の CIFile の方が圧縮前でも小さくなっている．また，共CIF [8] [10] は，列指向データベース同様，分散ファイルシス起頻度計算の中間ファイルで Snappy 圧縮を用いた場合，最大テム上のファイル形式と，データの配置を工夫することで計算で 25% のデータサイズ削減を確認できた．これは，同じ型ご機間のネットワーク IO やディスク IO を削減し，MapReduceとにバッファをまとめて圧縮をかけた効果であると考えられる．

ジョブを高速化する技術である．CIF は，列ごとにデータの表 1 IFile と CIFile における中間データサイズの比較
ファイルフォーマット

データ名 圧縮無しファイル (バイト) bzip2 圧縮ファイル (バイト) Snappy 圧縮ファイル (バイト)

IFile

Raw

9206558

13611113554898

CIFileRaw

9025188(2%削減)

1214726(11%削減)

3283301(8%削減)

IFile

共起頻度計算18806076

18190636095471

CIFile共起頻度計算17923310(5%削減)

1633830(11%削減)

4591160(25%削減)

表 2 CIFile における型ごとの圧縮効率の比較
データ名型

圧縮無しファイル (バイト) bzip2 圧縮ファイル (バイト) Snappy 圧縮ファイル (バイト)

Raw

整数

99917

32749((67%圧縮)

54374(45%圧縮)
2850701(65%圧縮)

Raw

文字列

8274496915891(88%圧縮)共起頻度計算整数

442138173477(60%圧縮)

296318(32%圧縮)共起頻度計算文字列13957109

1448904(89%圧縮)

4128572(70%圧縮)

表 3 IFile と CIFile における中間データサイズの比較と書き込み時間の比較
ファイルフォーマット

データ名

書き込み時間 (秒) Snappy 圧縮ファイル (バイト)IFile

共起頻度計算4307

36799837380

CIFile共起頻度計算2880(33%削減)

29365516361(21%削減)

ファイルを分割するだけでなく，データの co-location を行う．RCFile [9]，ORCFile [3] は MapReduce 処理系用のデータフォーマットである．列が大量に存在する場合にオブジェク
トのデシリアライゼーションを遅延することで MapReduce 処理系における入力を素早く行う手法である．
提案手法が対象としている MapReduce の中間ファイルは，blob 型の 2 列のスキーマと見なすことができるが，全ての列に対してに読み込み・書き込みを 1:1 の比率で行うため，CIF，RCFile，ORCFile で高速化可能なワークロードとは異なる．また，提案手法である CIFile は， CIF，RCFile，ORCFile とは補完的な関係にある．つまり，分散ファイルシステムからの読み込みの高速化を行う CIF，RCFile，ORCFile と，中間デー
タを削減することで Shuﬄe 処理の高速化を行う CIFile を組み合わせて利用することで，さらなる MapReduce ジョブの高速化を行うことができる．

6. ま と め
本稿では，MapReduce の Shuﬄe フェイズにおける中間出力を削減するため，列指向バッファ管理モジュールの一部である CIFile を提案し，そのファイル形式を利用することによる圧縮率の向上を確認した．このことから，提案手法を Hadoopに組み込むことにより，既存の MapReduce アプリケーションを未修正で高速化できる可能性がある．今後は，Hadoop への実装，評価を行う．また，Hive のように型が存在する処理系での実行効率の向上を検証する．

文

献[1] : Apache Hadoop, http://hadoop.apache.org/.

[2] : Apache Hadoop Wiki, http://wiki.apache.org/hadoop/
PoweredBy.
[3] : Create a new Optimized Row Columnar file format for Hive, https://issues.apache.org/jira/browse/HIVE-3874 (2013).
[4] : Treasure Data s Plazma: Columnar Cloud Storage，http://blog.treasure-data.com/post/53534943282/
treasure-datas-plazma-columnar-cloud-storage (2013).
[5] DataStax: Brisk, https://github.com/riptano/brisk.
[6] Dean, J. and Ghemawat, S.: MapReduce: simplified dataprocessing on large clusters, Proceedings of the 6th conference on Symposium on Opearting Systems Design & Implementation - Volume 6, OSDI’04, Berkeley, CA, USA,USENIX Association, pp. 10–10 (2004).
[7] Faraz Ahmad, Seyong Lee, M. T.: PUMA: Purdue MapReduce Benchmarks Suite.
[8] Floratou, A., Patel, J. M., Shekita, E. J. and Tata, S.:
Column-oriented Storage Techniques for MapReduce, Proc.
VLDB Endow., Vol. 4, No. 7, pp. 419–429 (2011).
[9] He, Y., Lee, R., Huai, Y., Shao, Z., Jain, N., Zhang, X. andXu, Z.: RCFile: A Fast and Space-eﬃcient Data Placement
Structure in MapReduce-based Warehouse Systems, Proceedings of the 2011 IEEE 27th International Conferenceon Data Engineering, ICDE ’11, Washington, DC, USA,IEEE Computer Society, pp. 1199–1208 (2011).
[10] Kaldewey, T., Shekita, E. J. and Tata, S.: Clydesdale:
Structured Data Processing on MapReduce, Proceedings ofthe 15th International Conference on Extending Database
Technology, EDBT ’12, New York, NY, USA, ACM, pp.
15–25 (2012).
[11] Li, B., Mazur, E., Diao, Y., McGregor, A. and Shenoy, P.: A
platform for scalable one-pass analytics using MapReduce,
SIGMOD ’11: Proceedings of the 2011 ACM SIGMOD International Conference on Management of data, New York,NY, USA, ACM, pp. 985–996 (2011).
[12] Lin, J. and Dyer, C.: Data-Intensive Text Processing with
MapReduce, Morgan and Claypool Publishers (2010).
[13] Ohta, K.: Hadoop meets Cloud with Multi-Tenancy, http://www.slideshare.net/treasure-data/hadoop-meets-cloud-with-multitena(2013).
[14] Peter Boncz, Marcin Zukowski, N. N.: MonetDB/X100:
Hyper-Pipelining Query Execution, Conference on Innovative Data Systems Research 2005 (2005).
[15] Saito, T. L.: snappy-java, https://code.google.com/p/

snappy-java/.
[16] Silberstein, A. E., Sears, R., Zhou, W. and Cooper, B. F.:
A batch of PNUTS: experiences connecting cloud batchand serving systems, Proceedings of the 2011 ACM SIGMOD International Conference on Management of data,
SIGMOD ’11, New York, NY, USA, ACM, pp. 1101–1112(2011).
[17] Stonebraker, M., Abadi, D. J., Batkin, A., Chen, X., Cherniack, M., Ferreira, M., Lau, E., Lin, A., Madden, S.,
O’Neil, E., O’Neil, P., Rasin, A., Tran, N. and Zdonik, S.:
C-store: A Column-oriented DBMS, Proceedings of the 31stInternational Conference on Very Large Data Bases, VLDB
’05, VLDB Endowment, pp. 553–564 (2005).
[18] Thusoo, A., Shao, Z., Anthony, S., Borthakur, D., Jain, N.,Sen Sarma, J., Murthy, R. and Liu, H.: Data warehousing
and analytics infrastructure at facebook, Proceedings of the2010 ACM SIGMOD International Conference on Management of data, SIGMOD ’10, New York, NY, USA, ACM, pp.
1013–1020 (2010).
[19] 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, pp. 13–24 (2013).
[20] 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 the9th USENIX Conference on Networked Systems Design andImplementation, NSDI’12, Berkeley, CA, USA, USENIX
Association, pp. 2–2 (2012).
[21] Zaharia, M., Chowdhury, M., Franklin, M. J., Shenker, S.
and Stoica, I.: Spark: Cluster Computing with Working
Sets, Proceedings of the 2Nd USENIX Conference on Hot
Topics in Cloud Computing, HotCloud’10, Berkeley, CA,USA, USENIX Association, pp. 10–10 (2010).
[22] Zukowski, M.: Balancing Vectorized Query Execution withBandwidth–Optimized Storage, PhD thesis, Universiteit
van Amsterdam (2009).
[23] Zukowski, M., Heman, S., Nes, N. and Boncz, P.: SuperScalar RAM-CPU Cache Compression, Proceedings of the22Nd International Conference on Data Engineering, ICDE
’06, Washington, DC, USA, IEEE Computer Society, pp.
59– (2006).