DEIM Forum 2013 F2-4

Map/Reduce におけるバケット再グループ化を用いた
ハイブリッドハッシュ結合アルゴリズム
廣瀬 繁雄†

大森匡†

新谷 隆彦†† 電気通信大学大学院情報システム学研究科 〒 182–8585 東京都調布市調布ヶ丘 1-5-1
E-mail: †shi1153020@hol.is.uec.ac.jp, ††{omori,shintani}@is.uec.ac.jp
あらましMap/Reduce における巨大データ処理においても，データ集合 R と S の結合演算は重要な処理の 1 つである．従来の研究では RepartitionJoin や DirectedJoin といった単純なハッシュ分割・マージソートで結合が行われている．しかし，n:m( 多対多 ) の結合演算といった制約の緩い等結合などでは，わずかなデータの偏りでも結合の CPU
コストが増加し，容易に負荷が生じる．本稿では Map/Reduce 上のハイブリッドハッシュ結合においてデータ偏りの検出と負荷分散を行なう方法を提案する．具体的には，ビルドフェイズの Map/Reduce 処理の中で同時にバケット内のデータ偏りの度数分布を調べ，そのバケットの再グループ化を行なうことで望ましいパーティション構成を求め，それによってプローブフェイズ時の reducer の負荷分散を行なう方法を述べる．
キーワード

MapReduce

Hybrid Hash Join Algorithm with Bucket Regrouping on Map/Reduce
Shigeo HIROSE† , Tadashi OHMORI† , and Takahiko SHINTANI†† The University of Electro-Communications,Graduate School of Information Systems
1–5–1 Chofugaoka, Chofu, Tokyo, 182–8585 Japan
E-mail: †shi1153020@hol.is.uec.ac.jp, ††{omori,shintani}@is.uec.ac.jp
Abstract Even in map/reduce applications, relational-style joins of big data sets under various constraints are
significant. Currently repartition join or directed join is a major algorithm for 1-to-many equality-join. However,in cases of many-to-many join such as equality-join having weak constraints or general theta-join, data skew easily
causes unbalance among reducers. To solve this problem, this paper describes a new algorithm of hybrid hash joinusing two map/reduce jobs. Intuitively, we embed a skew detection ability in reduce tasks of the build phase, andregroup resulting data-buckets into new partitions so as to decrease map/reduce overhead of the next probe phase.
Key words MapReduce

1. は じ め に
BigData と呼ばれるように，近年大量のデータが日々生成さが存在する．本稿で使用する Hadoop [1] とは，大規模分散処
理フレームワークである Map/Reduce [6] を参考にし，Apache
が JAVA によって実装したフレームワークである．

れている．BigData とは web のアクセスログやコンピュータ

この BigData の分析の中で，テーブルとテーブルとの結合による科学技術計算の結果，多様なコンテンツやリレーショナ演算は重要な処理の 1 つであるが，時間を多く使用する処理でルデータ等様々な種類のデータである．この大量なデータに対ある．例えば Facebook などでは，ユーザーの情報テーブルとして価値のあるデータ空間への変換を行うことで，傾向の特定ユーザーが訪問をしたクリックストリームのなどの分析を行う

や特異なデータの発見等を行っている．

ため，このテーブルの結合処理を絶えず数分や数十分といったこれらは元来並列データベースにおいても行われてきた処理

間隔で，数千台のノードで行っている．

だが，近年 Map/Reduce や NoSQL といったフレームワーク

Map/Reduce での結合演算は，元来並列データベースで行わやデータ管理システムが開発され，BigData に対して使用されれていた結合演算アルゴリズムのうち，ナイーブなアルゴリズ

始めている [2] [3]．これらはスキームを持たないような，不定ムである HashJoin が主に使われている [8]．Hadoop において長データを処理することを想定している場合や，並列分散性をも HashJoin は単純な MapReduce 処理によって実現されてい高め台数効果を容易に出すことを目的としているといった特徴

るが，並列データベースで行われてきた様々なアルゴリズムやskewhandling 等を鑑みると [11] [12] [13]，現状の Hadoop での結合演算では十分とは考えられない．

Hadoop 上での結合演算は主にマスターテーブルとログテー
ブルの結合といった 1:n の等結合が多いが，一方では，類似度

2. 2 DirectedJoin
また，DirectedJoin と呼ばれる 2 つのテーブルを同じハッシュ値で分割し HDFS と呼ばれる共有ファイルシステムに保存し，Map 処理のみで結合を行う結合演算が存在する．

結合やユーザー定義関数等を用いた結合など，データセット R

2 つのテーブルのパーティショニングを行っているということ S の要素間の多対多の関係性を計算する結合 (n:m 結合) 演とは，テーブルの偏りを調べることも可能なので，ユーザーが算も重要である．例えば R を論文のデータ集合，S を別の論

負荷分散機能を持たせることも可能である．また，Hadoop に

文集合，A を期間とすると，同じ期間の 2 つの論文集合の要素

おいて 1 つの Mapper の担当するデータ範囲は，予め決められ間の類似した組み合わせを求める条件式は，find (r, s) from Rている HDFS のチャンクサイズに依存する．このため，Map

x S where r.A = s.A and f(r,s) という式で書くことが出来る．

のみで結合を行う場合，チャンクサイズによって負荷分散が行

この場合 f(r,s) の類似度判定関数の内容によっては CPU コス

える場合も存在する．

トが非常にかかるものと予想できる．また，こうした n:m の結

しかしこの DirectedJoin では，両方のテーブルについてパー合を Hadoop のような分散処理基盤でナイーブな HashJoin をティショニングを行い，HDFS に保存を行わなければならな行うと，1:n の結合に比べてノード毎の処理量の偏りが更に増いため，巨大データの場合この HDFS への保存のコストが大大してしまう．

きい．

そこで本稿では巨大なデータに対する頑健で効率的な

2. 3 問題点の整理

Map/Reduce 上の n:m 結合演算のアルゴリズムの実装を行

既存のアルゴリズムの問題を以下にまとめる．

うことを目的とし，パーティションの再グループ化により負荷•分散を行う方法を提案する．また，結合条件が類似度や範囲制

ブルをパーティショニングする必要がある

約といった θ 条件が与えられている n:m 結合においても，本質•的には何らかの形で等結合の形で実行されることが多いことかある

ら [9] [14]，本稿では n:m の等結合演算を扱うとする．
以下 2. では Hadoop 上で一般的に行われている結合アルゴリズムの解説，3. ではハイブリッドハッシュ結合と改良方法の提案，4. では提案方法の詳細な説明，5. では実験結果を示す．

2. Hadoop 上での一般的な結合演算
2. 1 RepartitionJoin

負荷分散を行うにはサンプリングを行うか，両方のテー

Reducer のメモリ制約を満たさなければならない場合がそこで本稿ではこれら 2 点を解決するために，ハイブリッド
ハッシュ結合を用いることとした．

3. Hadoop 上でのハイブリッドハッシュ結合とその改良
3. 1 Hadoop 上での単純なハイブリッドハッシュ結合
Hadoop 上においての標準的なハイブリッドハッシュ結合ア

Hadoop で結合演算を行う場合一般的には RepartitionJoin

ルゴリズムの説明に移る．ハイブリッドハッシュ結合は片側のと呼ばれる，ハッシュ分割とマージソートを利用した結合方法

テーブルをパーティショニングするビルドフェイズと，もう片側が用いられる [4]．テーブル R,S といった 2 つのテーブルの結合のテーブルと結合するプローブフェイズを 2 回の Map/Reduceを行う場合，Map 処理ではテーブル R,S のそれぞれのレコー

ジョブを行うことで実現されている．DirectedJoin に比べて片ドを結合キーでハッシュ分割を行う．Reduce 処理では R,S の方のテーブルのパーティショニングだけでよいため HDFS へ

各レコードは同様のハッシュ値で振り分けられるので，結合をの保存のコストが少なく，更に片方のテーブルのデータの分布行うことが出来る．

状況を調べることが出来る．

片方のデータに偏りが多少あっても計算コストが低いような，

この時の処理の概要を図 1，2 を用いて説明する．

例えばマスターデータとログデータのような 1:n の等結合演算を行うだけの場合などでは RepartitionJoin で十分な速度となる [5]．しかし 1. で述べたような n:m の結合演算の場合，データに偏りがあると Reducer の CPU コストが非常に大きくなり，
データの偏った 1 つのノードで非常に時間がかかってしまう．
また，MultiJoin ではいくつかのテーブルのハッシュテーブル
を作る必要があるなど，結合条件やデータによっては Reducerのメモリ制約が存在する [7]．
このようにデータが偏ることで処理が重くなってしまう場合，

Hadoop では現在行っている処理を空いているノードに分担させ協調して動くような，動的に負荷分散を行う機能は存在しな
い．そのためサンプリングを行うことで予めに処理が偏るハッ
シュ値を求め，Map/Reduce ジョブの実行前に処理の分担させると定義しておくといった，静的に負荷分散を実現している．

図 1 HybridHashJoin ビルドフェイズ

とで，Reduce 時に更に分割しても結合キーがどのパーティショ
ンであるかということを明確にすることが出来る．
プローブフェイズではテーブル S に対してハッシュ値による分割を行う．得られたハッシュ値のパーティションが更に分割されていた場合，レンジ情報を元に対応するパーティションを求める．Reduce 処理では対応する R のパーティションと結合を行うこととした．
この時の処理の概要を図 3，4 を用いて説明する．
（ 1 ） ビルドフェイズの Map 処理ではテーブル R に対してハッシュ分割を行う．
（ 2 ） Reduce 処理では出力のハッシュテーブルがメモリに乗図 2 HybridHashJoin プローブフェイズ

るように分割し，Reducer がパーティションを複数に分ける必要があるのならば分割したレンジの位置を出力する．図 3 中で（ 1 ） ビルドフェイズの Map/Reduce ジョブでは，テーブル

R に対してハッシュ分割を行い，各 Reducer がそれぞれ HDFSに書きこむ．
（ 2 ） プローブフェイズの Map/Reduce ジョブの Map では
テーブル S を読み取り，そのまま key を結合キー，value を結合属性として出力し，テーブル R に使用したハッシュ値でハッシュ分割を行う．
（ 3 ） Reduce では，ハッシュ分割されたテーブル S のパーティションが存在する．このテーブル S に対応するテーブル Rをそれぞれ HDFS より読み取り，ハッシュテーブルを作成する．
（ 4 ） Reduce 処理で (3) で作成したハッシュテーブルを用は R1 が大きい場合で R1-1 と R1-2 の 2 つにレンジ分割した．
（ 3 ） プローブフェイズの Map 処理ではテーブル S を読み取り，key を結合キー，value を結合属性とし，テーブル R に使用したハッシュ値と，ハッシュ値に対応するパーティションが分割されている場合，レンジ情報も参考に分割を行う．図 4
では R1 が (2) で 2 つに分けられたため，S1 は R1 のレンジ情報により，R1-1 と R1-2 のどちらのレンジに属するか判断し分割を行っている．
（ 4 ） Reduce 処理では，分割情報を元に生成されたテーブ
ル S のパーティションと，それに合致するテーブル R のパーティションを読み出し結合処理を行う．

いて結合し出力する．

Fariha Atta らは，これら Hadoop 上でのハイブリッドハッ
シュ結合の，ハッシュ分割の部分をレンジ分割で実装した [15]．
負荷分散に対応する場合，RepartitionJoin と同様にビルドフェ
イズ前に R と S のデータのサンプリングを行い，各パーティ
ションの担当レンジを調整することで実現していた．しかし，RepartitionJoin と同様にサンプリングを行えないようなデー
タの場合では対応することができない．また，プローブフェイズにおいてテーブル R のパーティションのハッシュテーブルを作成しているが，ビルドフェイズでパーティションをメモリサ
イズ以下に作らなければ実行できなくなる恐れがある．

図 3 HSJ ビルドフェイズ

3. 2 HybridSkewJoin(HSJ)
まず，R のパーティションをある一定のサイズとすることでR に対応した負荷分散が行えると考えた．同時に頑健性を維持するため，R のパーティションがプローブフェイズ時にハッシュテーブルをメモリ上に作れる大きさとした．そこでビルドフェ
イズの Reduce の出力時に，プローブフェイズの Reduce 処理で作成するハッシュテーブルが，確実にメモリに乗ることが出来るような大きさとなるように更に分割を行なうこととし，このアルゴリズムを HybridSkewJoin(HSJ) と呼ぶ．例えばパーティションサイズの上限が 256MB の時に，ある Reducer が512MB のパーティションを振り分けられてしまった場合，これを 2 つの 256MB のパーティションに分割する．この 512MB
パーティションは結合キーでソートされているため，パーティ
ションの中でどこで区切ったかというレンジ情報を記録するこ図 4 HSJ プローブフェイズ

3. 3 HybridSkewJoin with adaptive probe(HSJ+AP) ならば複数の Reducer のうちどれか 1 つに出力する．図 6 で3. 3. 1 アルゴリズム

は S1 のハッシュ値のレコードは 2 つの Reducer のうちどちら

HSJ ではテーブル R がパーティションサイズの上限で分割かに出力を行う．

されることで負荷分散を行う．しかしこれでは不十分だと考え，（ 4 ） 同様に Reduce 処理では S のパーティションと合致すテーブル S に対応した負荷分散を行うとした．具体的にはテーるテーブル R のパーティションを読み出し結合処理を行う．

ブル S を 2 つに分割しプローブフェイズを 2 回に分けて実行

3. 3. 2 分散化を行う条件についてし，1 回目の結果を元に 2 回目の処理の分散を行うこととした．

3. 3. 1 で述べた 1 回目のプローブフェイズの結果を元に，処この方法を HybridSkewJoin with adaptive probe(HSJ+AP)

理の分散化を行う時の条件について説明を行う．1 回目のプローと呼ぶ．例えば 1 回目のプローブフェイズで，ある R のパー

ブフェイズでは pi 毎に結合条件に合致したレコード数を同時

ティション pi のデータが偏り処理が重くなってしまい，pi を担に計算し出力し，これを各パーティションのコスト ci とおく．

当する Reduce ノードが他の Reduce ノードと比べて時間がかあるノードが明らかに他のノードよりもコストが大きくなる場

かっていた場合，2 回目のプローブフェイズではパーティショ

合は，ci がノード毎の平均コストを上回る場合である．各ノーン p を担当する Reducer を増加させる．2 回目のプローブフェ

ドの平均コストは総コスト

イズでは，S のレコードが p の担当範囲ならば増加した複数の数 NR から求めることができ，式 (1) を満たす ci があるならReducer のうちどれかに振り分ける．このように予め Reducer

ば，そのパーティション pi は分散を行うべきだと判断する．

∑all

i=1 ciNRと同時実行できる Reducer∑allを増やすことで処理の分散化を行う．

ci >この時の処理の概要を図 5，6 を用いて説明する．

i=1 ci(1)NR

式 (1) を満たす場合，ノード毎の平均総コストを下回るように

Pi の担当 Reducer を出来る限り小さい値で増加させる．つまり式 (2) を満たす最小の α を増加後の Reducer 数とする．

(
arg min
α∈N

ci <
α =

∑all

i=1 ci)NR(2)

このように分散化を行うことで，最も終了時間の早いノード∑allと最も終了時間の遅いノードの実行時間の差が最悪でもi=1 ciNRとすることができる．
図 5 HSJ+AP 1 回目のプローブフェイズ

4. HybridSkewJoin with
Bucket Regrouping(HSJ+BR)
4. 1 問 題 定 義
HSJ では，ハッシュテーブルが作れる大きさにパーティショ
ンを区切ると定義した．しかし，3. 3 のように処理が偏り 1 つのパーティションに対し複数の Reducer が計算を行う場合，プ
ローブフェイズでのデータ移動のコストが増加してしまう場合が存在する．例を挙げると，パーティションの大きさを 256MBとし，この内の 20%が同一な結合キーのレコードとなり，この20%のレコード集合の処理が重く負荷分散をを行わなければな
らない場合である．複数の Reducer が 20%のレコード集合の処理のためにパーティション全体を読み込まなくてはならない．
図 6 HSJ+AP 2 回目のプローブフェイズ

また，ある結合キーのレコードが両方のテーブルに大量に存在する場合，この結合キーを担当する Reducer は非常に遅くな（ 1 ） ビルドフェイズと 1 回目のプローブフェイズは 3. 2 と同様である．
（ 2 ） 処理の重い Reducer があるか判断を行う．式 (1) を満たす Reducer があるならば，対応する Reducer 数を α 個に増加させる．図 5 では R1 と S1 の結果が偏っていると判断し，

R1 と S1 の Reducer 数を 2 とした．
（ 3 ） 2 回目のプローブフェイズでの Map 処理では，Keyが 1 回目のプローブフェイズで時間のかかったパーティション

る．しかし片方のテーブルのみある結合キーのレコードが非常に多いが，もう片方はそれほど存在しない場合では，処理の内容とデータの量によるが極端に遅くはならない．このため，片方のテーブルのみでデータの偏りの大小を判定することは有用であると考えられる．

4. 2 アルゴリズム
そこで我々は，ビルドフェイズでパーティションよりも小さ
いサイズのバケットで分け，更にバケットを収集し再グループ化を行うこととした．バケットを細かく分けすぎるとプローブフェイズの Reducer が大量に増加してしまうため，処理が軽（ 2 ） Reduce 処理で一定のファイルサイズでバケットを出いと考えられるバケットはパーティションサイズに再グループ

力し，このとき同時にバケット毎の度数分布も出力する．

化することで，このグループを 1 つのパーティションとして扱（ 3 ） ビルドフェイズが終わったら再グループ化をマスター

う．逆に処理が重いと考えられるバケットはグループ化せず，ノードが行う．度数分布を元にバケットを纏めパーティション

単独のバケットをパーティションとして扱う戦略をとった．こを作成する．このときバケットのハッシュ値と結合キーのレンのため，ビルドフェイズでは R の各バケットの度数分布を生ジ情報から新たなパーティションの分割情報を記録する．

成することとした．この方法を HybridSkewJoin with Bucket（ 4 ） プローブフェイズの Map 処理ではテーブル S のみ読Regrouping(HSJ+BR) と呼ぶ．

み取り，(3) で作られた分割情報を元に出力する．

この時の処理の概要を図 7，8，9 を用いて説明する．

（ 5 ） Reduce 処理では各 Reducer が対応するパーティショ
ンを読み取り結合を行う．

4. 3 バケットの再グループ化と処理の分散化について
4. 3. 1 度数分布の生成
HSJ+BR ではビルドフェイズのバケットのグループ化時に，各バケットの度数分布を元に決定している．このバケットの度数分布の求め方は以下の 2 つの場合が考えられた．
（ 1 ） 独立な結合キー毎にレコード数をカウントする（ 2 ） 結合キーに対して適当な長さのハッシュ関数を用いて，ハッシュテーブルの値で度数分布を得る
ここでは (1) の方法を使用した．バケットの分布状況を得る
図 7 HSJ+BR ビルドフェイズ

とき (1) を使う場合，バケット内の結合キーが全て異なる場合
等，分布状況によっては作成と判定のコストが多くかかってしまう場合が存在する．しかしテーブル R 側のある独立な結合
キーのレコード数が少なければ，それと同一なテーブル S 側の結合キーのレコード数は極端に多くならなければそれほど時間はかからないと考えられた．そこで独立な結合キーのレコード数に閾値を定めることで，閾値以上のレコード数を持つ独立な結合キー kn と，そのレコード数 brn を得ることが出来るように度数分布を出力するとした．

4. 3. 2 バケットの再グループ化の判定方法
4. 3. 1 により，ある一定以上のレコード数を持つ結合キーとそのレコード数によりバケット内の分布状況を求める方法を示した．この情報を元にバケットの処理の分散化と再グループ化の判定を行う．
ここで，あるバケット bi のコスト ci を，ある閾値以上のレ
図 8 HSJ+BR 再グループ化

コード数を持つ独立な結合キー kn のレコード数 brn の総和とする．このコスト ci が小さいようならば他の小さいバケットと再グループ化を行い，このグループをパーティションとして扱う．逆にコストが大きければ，バケット単体をパーティショ
ンとして扱う．また，分布状況によっては HSJ+AP のように
パーティションの Reducer 数を増加させる処理も行うことで，処理の分散化を行う．
・再グループ化の判定
バケットの再グループ化によるパーティションの作成は，グループを行うバケットのコストの和がある閾値 t よりも小さい場合に行う．この閾値 t は式 (1) と同様に，全体のコスト

図 9 HSJ+BR プローブフェイズとノード毎の平均予想コストから求めることとした．よって t∑alli=1 ci ，クラスタが同時実行できるは全バケットのコストをReducer 数を NR とおいた場合，式 (3) となる．
（ 1 ） ビルドフェイズでは Map 処理でテーブル S に対してハッシュ分割を行う．

1600

1000
800

400

コストが大きいと判断されるバケット bi は，コスト ci が式

200(4) を満たす場合分散化を行う．
∑all
ci >

i=1 ci0

アルゴリズム名(4)

NR

この時バケット i の担当 Reducer 数は，式 (5) を満たす α とHSJ+BR

600

・バケットの分散化(3) で求められる閾値 t よりも大きい場合と定めた．つまり式

Job1
Job2
Job3

HSJ+APの和が t よりも大きくならない様に再グループ化を行う．

1200
時間(秒)

テーブルを作れる大きさを上限とし，複数のバケットのコスト

1400

HSJ

i=1 ci(3)
NR
グループ内のバケットの最大個数はパーティションがハッシュ

Repartition

∑all

t=

図 10 両方のテーブルに同様な偏りの分布がある場合 (R5 万 S5 万)することでコストの大きいバケットの処理の分散化を行う．

(
arg min
α∈N

ci <
α =

∑all

i=1 ciHSJ+BR におけるプローブフェイズの各ノードの Reduce 処理

)

NR

(5)

の所要時間をそれぞれ図 11 から 15 に示す．図中の Task1 とは各ノードが一番初めに行なった Reducer の時間であり，Task2

また，再グループ化を行なったパーティションのコストがそとは Task1 の Reducer の後に実行した Reducer の時間である．

れぞれ平均化されるように再グループ化を行う等，再グループ

図 14，15 では Reducer が同時実行数以上存在したのでいくつ化を行うバケットの組み合わせの戦略も考えられるが，本稿でかのノードは Reducer を複数回実行することとなった．

は単純にファイル名でソートされた先頭のバケットから順番に
上記の条件を満たすように再グループ化を行なった．

5. 実

験

表 1 には HSJ によるパーティションと，HSJ+BR のバケットによる再グループ化と分散化を行なったパーティションの実
行時間の比較を示す．実験では全てのパーティションは 4 つのバケットに分割された．Partition1 から 4 ではそのバケット

5. 1 実 行 環 境が再グループ化により HSJ と同様の範囲のパーティションと実行環境は 1 台のマスターノードと 5 台のスレーブノードかなった．Partition5 では 3 つのバケットが 1 つのパーティショ

らなり，スレーブ 1 台が同時実行できる Map タスク，Reduceンとなった ( 表 1 の 14 秒 )．残りの 1 つのバケットは 4 つのタスク共に 1 とした．Hadoop のバージョンは 0.20.2 を使用し，Reducer に分散して実行された．この 4 つに分散されたバケッ

OS は VineLinux6.0(64bit) とした．

トは {} でまとめた時間となった．

使用したデータはテーブル R，S 共に 1 レコード 100byte，

再グループ化を行う場合，HSJ でのパーティションを跨いで結合キーを 50byte とし，共に 1GB のデータとした．偏りとしグループ化を行うため，HSJ のパーティションと HSJ+BR のてキー 1 が 50,000 レコード，キー 2 が 25,000 レコード，キー

パーティションは一致しない場合が存在するが，本稿のデータ

3 が 12,500 レコード... といった傾き 1/2 となるような偏りのでは HSJ のパーティションを跨いでグループ化を行うことはあるデータとし，R，S 共に同様な偏りの分布を持つデータとなかったため，表 1 のように示す．

した．
結合条件は R.key=S.key を条件とした n:m の等結合演算と表 1 HSJ と HSJ+BR のパーティションの速度比較 (秒)HSJ

HSJ+BR

し，計算部分として HDFS への出力を行わないが，結合結果

Partition11816をメモリ上に作成するとした．

Partition2

17

19

Partition3

77

76

Partition4

389

412

5. 2 結果RepartitionJoin と HSJ，HSJ+AP，HSJ+BR の結果を図10 に示す．

Partition5 1322 14,{367,306,302,404}

パーティションのハッシュテーブルを作成できる最大のサイズ
は 256MB とした．プローブフェイズを 2 回に分ける HSJ+AP
では，S は 20%と 80%に分けるとした．HSJ+BR での 1 バケッ
トのサイズは 64MB とし，独立な結合キーのレコード数の閾値
は 10 とした．

6. お わ り に
本稿では Map/Reduce 上における n:m の結合演算において，データが偏り結果として全体の処理時間が低速になる場合

図 10 中の Job1 は RepartitionJoin では全体の実行時間，に有効なアルゴリズムである Hybrid Skew Join with Bucket

HSJ，HSJ+AP，HSJ+BR ではビルドフェイズの実行時間とな

Regrouping (HSJ+BR) を提案した．このアルゴリズムでは細る．Job2 以降はプローブフェイズの実行時間であり，HSJ+AP

かいバケットに分割し，その度数分布からプローブ処理の負荷では Job3 が 2 回目のプローブフェイズの実行時間である．

を推定し，バケットの再グループ化と処理の分散化の決定を行ま た ，こ の と き の RepartitionJoin と HSJ，HSJ+AP，

うという機構を用いた．結果として両方のテーブルに同じ分布

1200

1400

Task1

1000
時間(秒)

800
600
400

800
600

ノード名

ノード毎の Reducer の所要時間

(HSJ)
450

Task1
Task2

400
350

250

150

ノード名

図 14

(HSJ+AP Job3)の偏りがある場合では，提案方法の有効性を示せた．
しかし，R と S が分布の違う偏りの場合や，R は偏っている[9]が S が一様分布な場合などで追加実験を行う必要がある．
また，各バケットの度数分布では単純に独立な結合キー毎に[10]

レコード数をカウントし，閾値を設けることでマスターノード
における再グループ化と分散化の処理量を減らしている．しか[11]

し汎用性の観点からある程度の長さのハッシュ値を用いた分布判定を行なうような，Fine-grain なハッシュテーブルを用いたほうが普遍性があると考えられる．

[12]

最後に，本稿では n:m の等結合演算を扱ったが，θ-Join や

SetSimilarityJoin，BandJoin といった様々な条件の結合にお[13]

いては，ネットワークコストや負荷分散の最適化モデルが更に重要になってくると考えられる．このような様々な条件の結合[14]演算の高速化に対応することも今後の課題である．
辞本研究の一部は，科研費 (課題番号：24500109) の助成によるものである．
文
[1]
[2]
[3]
[4]
[5]

[6]
[7]
[8]

献

Hadoop. http://hadoop.apache.org/.
MongoDB. http://www.mongodb.org/.
Cassandra. http://cassandra.apache.org/.
J. Lin, C. Dyer. ”Data-Intensive Text Processing With
MapReduce”, Morgan and Claypool Publishers.
S. Blanas, J. M. Patel, V. Ercegovac, J. Rao, E. J. Shekita,
Y. Tian ”A Comparison of Join Algorithms for Log Processing in MapReduce”, ACM SIGMOD, pp. 975-986, 2010.
J. Dean and S. Ghemawat. ”MapReduce: Simpliﬁed Data
Processing on Large Clusters”, OSDI , pages 10-10, 2004.
F. N. Afrati , J. D. Ullman. ”Optimizing joins in a mapreduce environment”, EDBT, pp. 99-110, 2010.
A. Pavlo, E. Paulson, A. Rasin, D. J. Abadi, D. J. DeWitt, S. Madden, and M. Stonebraker. ”A comparison of

ノード5

ノード名

ノード毎の Reducer の所要時間(HSJ+AP Job2)

謝

0

ノード5

ノード4

ノード3

50

ノード名

ノード毎の Reducer の所要時間

ノード5

100

ノード2

ノード5

ノード4

ノード3

0

ノード2

0

ノード1

50

200
150

100

50

250

ノード4

100

300

200

ノード3

150

時間(秒)

時間(秒)

200

Task1
Task2
Task3

ノード2

300

図 13

ノード4

ノード毎の Reducer の所要時間

350

250

時間(秒)

図 12

(RepartitionJoin)
Task1

ノード1

300

ノード名

ノード1

図 11

ノード3

ノード5

ノード4

ノード3

ノード2

0

ノード1

200

0

ノード2

400

200

ノード1

時間(秒)

Task1

1200

1000

[15]

図 15

ノード毎の Reducer の所要時間(HSJ+BR)approaches to large-scale data analysis”, SIGMOD, pp. 165178, 2009.
A. Okcan, M. Riedewald ”Processing theta-joins using
mapreduce”, ACM SIGMOD, pp. 949-960, 2011.
R. Chaiken, B. Jenkins, P. Larson, B. Ramsey, D. Shakib, S.
Weaver, J. Zhou. ”SCOPE: Easy and Eﬃcient Parallel Processing of Massive Data Sets”, VLDB, pp.1265-1276, 2008.
D. J. DeWitt, R. H. Gerber, G. Graefe, M. L. Heytens, K. B.
Kumar, and M. Muralikrishna. ”GAMMA - A High Performance Dataﬂow Database Machine.”, VLDB, pp. 228-237,
1986.
D. J. DeWitt, J. F. Naughton, D. A. Scheneider, S.Seshadri,”Practical Skew Handling in Parallel Joins”, VLDB, pp. 2740, 1992.
M. Nakayama, M. Kitsuregawa,M.Takagi, ”Hash-Partitioned
Join Method Using Dynamic Destaging Strategy”, VLDB,
pp. 468-478, 1988.
S. Dhaudhuri, V. Ganti, R. Kaushik ”A Primitive Operator for Similarity Joins in Data Cleaning”, ICDE, pp. 5-16,
2006.
Fariha Atta, ”Implementation and Analysis of Join Algorithms to handle skew for the Hadoop Map/Reduce Framework”, Master’s thesis, University of Edinburgh, 2010.