論文 / 著書情報
Article / Book Information
論題(和文)

低遅延ストリーム処理のための結合演算並列実行方式

Title(English)

Parallel Join Execution Methods for Low Delay Stream Processing

著者(和文)

渡辺陽介, 横田治夫

Authors(English)

Yousuke Watanabe, Haruo Yokota

出典(和文)

情報処理学会研究報告, Vol. 2012-DBS-154, No. 10, pp. 1-8

Citation(English)

IPSJ SIG Technical Report, Vol. 2012-DBS-154, No. 10, pp. 1-8

発行日 / Pub. date2012, 8

権利情報 / Copyright

ここに掲載した著作物の利用に関する注意: 本著作物の著作権は（社）情報処理学会に帰属します。本著作物は著作権者である情報処理学会の許可のもとに掲載するものです。ご利用に当たっては「著作権法」ならびに「情報処理学会倫理綱領」に従うことをお願いいたします。
The copyright of this material is retained by the Information Processing
Society of Japan (IPSJ). This material is published on this web site withthe agreement of the author (s) and the IPSJ. Please be complied with
Copyright Law of Japan and the Code of Ethics of the IPSJ if any users
wish to reproduce, make derivative work, distribute or make available tothe public any part or whole thereof.

Powered by T2R2 (Science Tokyo Research Repository)

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

低遅延ストリーム処理のための結合演算並列実行方式
渡辺 陽介1,a)

横田 治夫1,b)

概要：ストリームデータの増加に共に伴い，ストリームデータに対する問合せ処理が重要となってきている．情報源の種類も多様化し，複数の情報源から到着するストリームデータ同士の統合処理は特に需要が高くなっている．ストリームの統合を実現する演算はウインドウ結合と呼ばれており，時間幅や列数で定義されたスライディングウインドウにより処理対象を最新のデータに限定した結合処理を行う．結合処理を高速に実行するための既存のアプローチとして，分散環境を用いた処理の並列化が提案されており，ス
ループットの改善が行われている．しかし，処理遅延の改善に関しては，分散環境がネットワーク遅延の影響を受けやすいということもあって，これまであまり考慮されてこなかった．本研究では，分散環境の代わりにマルチコア環境を用いて，複数ストリームに対するウインドウ結合演算の処理遅延を改善するような並列化方式を提案する．現在，N 個のストリームに対する結合を実現する方式は，2 入力の結合演算をツリー状に多段接続するものと，N 入力の結合演算を用いるものの 2 種類に大別される．本研究は，両方の方式に対応する並列結合ツリー方式と並列 MJoin 方式の 2 つを提案する．並列結合ツリー方式では，
N 個の入力ストリームに対し，それぞれのストリームの出力を最速で出力することに特化したプランを N個導出し，それらを N 並列に走らせるというものである．並列 MJoin 方式は，既存の MJoin のアルゴリ
ズムがシーケンシャルに行っていたプローブ処理を N 並列化したものである．本研究では，提案手法 2 つとオリジナルの方式 2 つについて，評価実験を行いそれぞれの性能の比較を行う．

1. はじめにエンジンの扱うデータ処理演算のうち，複数の情報源のストリームデータ同士を統合する処理算はウインドウ結

近年，時間と共に次々と生み出されるストリーム型の合 (Window-Join)[8] と呼ばれている．ウインドウ結合演

データに対して注目が集まっている．カメラ・センサーか算は，リレーショナル代数演算の中の結合演算の拡張であらの計測データや，マイクロブログ [1] からのつぶやきなり，無限に到着し続けるストリームデータを扱うために，ど，ストリームデータの種類は豊富である．そのため，ス処理対象となるデータの範囲を制限するためのスライディ

トリームデータに対する選択・蓄積・統合・分析などのデー

ングウインドウの概念が導入されたものである．

タ処理の要望が高まっている．特に複数の情報源から到着するストリームデータ同士の統合は需要が高い．

ストリーム処理エンジンは主記憶上でのデータ処理を前提としているため，DBMS 等よりもパフォーマンスは高ストレージに格納された RDB データなどの静的な情報

かったが，ストリームデータのさらなる増加に伴い，よりとは異なり，次々と新しいデータが得られるストリームで少ない遅延でより大量のデータを処理できる，効率的な問は，従来型の DBMS によるデータ処理は十分なパフォー

合せ処理技術が求められるようになってきた．問合せ処マンスが得られない．そこで，ストリームデータに対する理効率化の一つの方向として，複数台のノードを用いた連続的な問合せ処理を実現するストリーム処理エンジンの分散処理があり，これまで多くの研究において，ストリー開発が行われてきている [2], [3], [4], [5], [6], [7]．ストリー

ムに対する問合せ処理を分散化する手法が提案されていム処理エンジンでは，問合せ処理要求として SQL ベース

る [9], [10], [11], [12], [13], [14]．

の問合せ文またはデータの流れを表すフロー等を登録す

これらの分散処理手法が重点を置いていることはスルー

ると，新規データが到着する度に対応する演算評価を繰

プットの向上，すなわちノードごとに異なる入力データをり返し行い，処理結果を逐次生成する．ストリーム処理

同時に処理させることで，システム全体として単位時間当たりに処理可能なデータ量を増加させることである．これ

1

a)
b)

東京工業大学Tokyo Institute of Technology
watanabe@de.cs.titech.ac.jp
yokota@cs.titech.ac.jp

c 2012 Information Processing Society of Japan⃝までの並列化手法はスループット向上については貢献してきたが，一方で，入力されたデータをより少ない遅延で出1

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

力することに関してはあまり考慮してきてはいない．処理

2.1 スライディングウインドウ

遅延は，データが演算を通過する際に発生してしまう不可

本稿では，リレーションのタプルをベースにした，タプ

避のものであるが，応答性の高さを求められるアプリケー

ルストリームを前提とする．各タプル t には情報源からシションにおいては非常に重要な要素である．しかし，分散

ステムに到着した時刻を表すタイムスタンプ t.T S がつけ

環境ではノード間のデータ転送のための通信オーバヘッド

られているものとする．本研究の目的は処理遅延の削減でが必要なために，処理遅延を削減する目的で並列化を行うあるので，新規にタプルが到着した場合，可能な限り短いことは難しかった．

時間でそのタプルから生成できる処理結果を出力へ送るも分散化とは別の方向で，近年のマルチコア CPU の登場のとする．データ処理に関しては，本稿はウインドウ結合

により，1 台のノードの中でもデータ処理を並列化するこの処理遅延の削減が目的のため，ここでは結合演算のみをとが容易になってきている．1 台に数十個のコアを有する

扱う．

ノードも珍しくなくなった．マルチコア環境は，一般的なウインドウ結合演算は，リレーショナル代数演算の中の分散環境よりも並列化にかかるオーバヘッドがはるかに小結合演算の拡張であり，無限に到着し続けるストリーム

さいため，従来の分散ストリーム処理とは異なる戦略が可

データを扱うために，処理対象となるデータの範囲を制限能である．例えば，複数個の CPU コアをスループット向するためのスライディングウインドウ (以下ウインドウ) の上だけでなく，処理遅延を削減する目的で用いることがで概念が導入されたものである．ウインドウは，時間幅で指きる．このようなマルチコアを前提とした並列ストリーム

定されるものと，列数で定義されるものの 2 種類がある．

データ処理については，まだ十分な検討が行われていない．

時間幅指定のウインドウでは，新規到着したタプルのタイ

本研究は，マルチコア環境における処理遅延の削減が可ムスタンプを基準に，w 単位時間 (e.g. w ミリ秒) 以内に届

能な並列ストリーム処理の実現を目標とし，特に本稿では，いた他の情報源のタプルを処理の対象とする．列数指定の

複数ストリーム上でのウインドウ結合演算を対象に，マルウインドウでは，あるタプルが到着した時点で，最新の k

チコアを活かした並列処理方式を検討する．現在，N 個の個のタプルが処理の対象となる．本稿では，時間幅のもの

ストリームに対する結合を実現する方式は，2 入力の結合のみを対象とし，情報源ごとに異なる時間幅のウインドウ

演算をツリー状に多段接続するものと，N 入力の結合演算を指定可能とする．

を用いるものの 2 種類に大別される．本稿では，両方の方

各タプルは，ウインドウの範囲に含まれている間，シス式に対応する並列結合ツリー方式と並列 MJoin 方式の 2

テム内のメモリ領域に保持されていなければならない．本つを提案する．並列結合ツリー方式では，N 個の入力スト

稿では，ストリーム S からの到着タプル t を一定期間保持

リームに対し，それぞれのストリームの出力を最速で出力するための領域をウインドウ領域 WS と表記する．

することに特化したプランを N 個導出し，それらを N 並列に走らせるというものである．並列 MJoin 方式は，既存の MJoin [15] のアルゴリズムがシーケンシャルに行っていたプローブ処理を N 並列化したものである．また，本稿
では，提案手法 2 つとオリジナルの方式 2 つについて，評
価実験を行いそれぞれの性能の比較を行う．
本稿の構成は以下の通りである．まず，2 節において本

2.2 2 入力のウインドウ結合
まず最初に 2 つのストリームを入力として扱うウインド
ウ結合の処理アルゴリズムについて紹介する [8]．
ここでは 2 つのストリーム S1 , S2 に対するウインドウ結合 S1 1θ,W1 ,W2 S2 を考える．ただし，θ は結合条件，W1 ,W2 はそれぞれ S1 , S2 のウインドウ領域とする．

研究の提案手法を理解するうえで必要なウインドウ結合に結合のアルゴリズムとしては，入れ子ループ結合 (Nested-

ついて紹介する．次に 3 節において，提案手法である並列

loop join) ベースのものとハッシュ結合 (Hash join) ベース

結合ツリー方式と並列 MJoin 方式を説明する．4 節では，のものがある．ハッシュ結合は θ の条件が等結合 (=) の場

本研究で行った評価実験について述べる．5 節で関連研究合のみしか使えないが，大量データの処理に適している．

について紹介したのち，6 節にてまとめと今後の課題を述

本稿で述べる並列化手法はどちらにもアルゴリズムの場合べる．

も適用可能であるが，本稿の実験ではハッシュ結合の方を

2. 前提知識: ウインドウ結合
ここでは，これまでに提案されてきている，ウインドウ
結合について紹介する．最初にスライディングウインドウについて説明し，2 入力の結合アルゴリズムについて述べ，

3 入力以上の結合方式について述べる．

実装した．

2.2.1 入れ子ループ結合
S1 1θ,W1 ,W2 S2 を入れ子ループ結合で処理する場合について説明する (図 1)．
入れ子ループ結合では，ストリーム S1 からの新規到着
タプル t1 が与えられた際に，S2 のウインドウ領域 W2 内に含まれる全てのタプル w2 ∈ W2 とのマッチングが行われる．もし，マッチング候補 (t1 , w2 ) が条件 θ を満たす場

c 2012 Information Processing Society of Japan⃝2

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report
㻠

㼗㻝

㼎㻝

㻝

㼗㻝

2.3 N 入力のウインドウ結合

㼍㼍

2.2 節は 2 つのストリームが入力として与えられた場合

Outputの話であるが，次に N 個のストリームが与えられた場合の

W1

㼀㻿

㻷㻝

㻭㻝

㻞

㼗㻝

㼍㼎

䝬䝑䝏䞁䜾 㼀㻿
䕿

㻝

㽢

⋈

㏣ຍ
㼀㻿

t1

㻷㻝
㼗㻝

㻠

㻷㻞

㻭㻞

㼗㻝

㼍㼍

㼗㻞

㻟

W2

㼏㼏

ものと，N 入力の結合演算として実現するものがある．

2.3.1 結合ツリー

S1.K1=S2.K2

この方式では，N 個の入力ストリーム S1 , . . . , SN が与え㻭㻝られた場合，N-1 個の 2 入力のウインドウ結合演算を多段㼎㻝S1

᪂つ฿╔䝍䝥䝹

図 1

㻠

S2

に組み合わせて目的の結合処理を実現する (図 3)．
ウインドウ結合の組み合わせ方は，結合ツリー (Join

入れ子ループ結合tree) と呼ばれる木構造で表現される．結合ツリーは，ク
㼍㻝

㼎㻝

㻝

㼗㻝

エリの出力を根，各入力ストリームを葉とした 2 分木であ㼍㼍り，枝分かれしている部分が 2.2 節で説明した 2 入力の結Output
W1

H1
t1

㼀㻿

㻷㻝

㻭㻝

㻞

㼗㻝

㼍㼎

䝡䝹䝗

䝬䝑䝏䞁䜾 㼀㻿
䕿

⋈
h(x)

㼀㻿

㻷㻝

㻭㻝

㻠

㼍㻝

㼎㻝

結合処理について述べる．主なアプローチは 2 つあり，2入力の結合演算をツリー状に多段に組み合わせて実現する

㻷㻞

㻭㻞

㻝

㼗㻝

㼍㼍

㻟

㼗㻞

㼏㼏

W2

合演算に対応する．
結合ツリーの構造として left-deep ツリー, bushy ツリー
等どの構造を選ぶか，また各ストリームをツリー内のどの位置に配置していくかについては，問合せ最適化技術 [16]S1.K1=S2.K2

H2

䝥䝻䞊䝤

の長年のテーマでもある．一般に，各ストリームの入力
レート (#tuples/second) と各結合条件の選択率 (結合結果のタプル数/直積結果のタプル数) が完全に既知で，しかS1᪂つ฿╔䝍䝥䝹

S2

図 2 ハッシュ結合

もそれらが時間変化しないという前提であれば，最適なツリーを決めることが可能である (近似解でない場合は入力ストリーム数 N に対して多項式時間では解けないが)．最合は，タプル (t1 × w2 ) が出力される．マッチング処理の適なツリーをどのように生成するかについては，本稿の範終わった t1 は後続のタプルの到着に備えて W1 へ追加さ

囲を超えるのでここでは触れない．

れ，ウインドウの幅指定の条件を満たす間システム内に留本研究の目的である，処理遅延の削減という観点からこまり続ける．S2 からの新規到着タプル t2 が与えられた場のアプローチをみたときに重要なのは，ストリーム Si の合も同様な処理が行われる．

タプル ti が届いた時に，Si がツリー内のどの位置にあるか

2.2.2 ハッシュ結合に応じて，出力までに通過しなければならない結合演算の次に S1 1θ,W1 ,W2 S2 をハッシュ結合で処理する場合に数が変わってくるという点である．例えば，図 3 のようなついて説明する (図 2)．入れ子ループ結合では，結合相手

left deep な結合ツリーの場合，根から最も深い位置になっのウインドウ領域内に含まれる全てのタプルとのマッチンたストリーム (e.g. S1 ) のタプルの処理結果は N-1 個の演グが必要だったが，ハッシュ結合ではウインドウ領域内の算を通過しなければ出力できないのに対して，根から最もタプルに対するハッシュ表 HS を用意することでマッチン

浅い位置になったストリーム (e.g. SN ) のタプルの処理結グのコストを削減する．

果は 1 個の演算を通過するだけで出力にたどり着ける．ツ

ハッシュ表は，結合条件 θ で結合キーに指定された属性

リー内の位置に応じて処理遅延が変わるということが本研に対し，ハッシュ関数 h(x) を適用することで構築される．

究が今回着目した性質であり，3 節の提案手法では，マルストリーム S1 からの新規到着タプル t1 が与えられた際チコアによる並列化によりこの部分を補っている．

に，ハッシュ値 h(t1 ) を計算し，S2 のハッシュ表 H2 から

もう 1 点，次の 2 つ目のアプローチとの対比で重要な特h(t1 ) = h(w2 ) となるタプル w2 ∈ W2 とのマッチングが行徴がある．それは，1 つの結合演算の出力が次の結合演算

われる．マッチング処理の終わった t1 は後続のタプルのの入力となり，次の結合演算のウインドウ領域にはその中到着に備えて W1 へ追加され，さらに S1 のハッシュ表 H1

間結果が保持されるという点である．これらの中間結果はには t1 の情報が追加される．

別のタプルが到着した際の処理に再利用できるという利点

以降では，統合相手のストリームのハッシュ表を検索しがあるが，中間結果のためのウインドウ領域やハッシュ表てマッチングを行う部分の処理をプローブ，自分のストの管理も必要になるという欠点も同時に存在する．

リームのハッシュ表に新規到着タプルの情報を反映する処

2.3.2 MJoin

理をビルドと呼ぶものとする．

c 2012 Information Processing Society of Japan⃝

N 入力を扱うことが可能な結合演算 (MJoin [15]) のアプ

3

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

OutputOutput
W12…(N-1)
W12

⋈ W

⋈ W
W ⋈ W
1

⋈

N

䝡䝹䝗 䝥䝻䞊䝤1 䝥䝻䞊䝤2

3

W1

W2

W3

WN

2

t1

S1

䝥䝻䞊䝤(N-1)

S2

S3

図 3

結合ツリー

SN

ローチについて説明する (図 4)．

MJoin では，N 個の入力ストリーム S1 , . . . , SN に対応するウインドウ領域 W1 , . . . , WN とハッシュ表 H1 , . . . , HNを保持している．あるストリーム Si から新規タプル ti が到着した場合，ti を Wi と Hi へ反映させるビルドの処理を行い，その後で他のストリーム S1 , . . . , Si−1 , Si+1 , . . . , SNの各ハッシュ表を用いたプローブ処理を行う．1 つのハッシュ表のプローブ処理が終わったら，それを中間結果として次のハッシュ表のプローブ処理を行うということを N-1
回繰り返し，最後まで残ったものを出力する．
図 4 では左側からプローブ処理が行われているが，一般的にはプローブ処理を行うストリームの順番は，結合条件の選択率等に応じて中間結果がより少なくなるようなものが選ばれる．また，このプローブの順番は各タプルの属するストリームごとに決めることができ，選択率が変化した場合は動的に順番を変更してもよい．本研究の目的である，処理遅延の削減という観点からこのアプローチをみた
ときに重要なのは，プローブ処理をシーケンシャルに行っているという点である．この N-1 回のプローブ処理を並列

᪂つ฿╔
䝍䝥䝹

S1

S2
図 4

S3

SN

MJoin

リーを用いるものと N 入力の結合演算を用いるものがある．ここではそれぞれについてマルチコア向けの並列化を行った方式を 1 つずつ提案する．

3.1 並列結合ツリー方式
2.3.1 で述べた通り，結合ツリーを用いて 2 入力のウイン
ドウ結合演算をつなぎ合わせた場合，どのようなツリーの構造になるにしろ，ツリーの根から浅いところのストリームと深いところのストリームでは処理遅延に差が生じてしまう．しかし，逆に言えば処理遅延を減らしたいストリー
ムをあえてツリーの浅いところに配置すれば，そのストリームからのタプルについては短い処理遅延で結合処理が可能になるということである．
ここでマルチコアの特性を生かして，複数の異なる結合
ツリーを並列に実行することを考える．このとき，N 個のストリームに対して N 個のツリーを用意し，i 番目のツ
リーではストリーム Si が根の最も浅い箇所で結合処理が行われるようにする．すなわち，以下のような N 個の結合ツリーができるようにする (図 5)．

化できれば処理遅延が減るのではないか，というところが

S1 1 (S2 1, . . . , 1 SN )3 節で述べる 2 つ目の提案手法のアイデアである．

Si 1 (S1 1, . . . , 1 Si−1 1 Si+1 1, . . . , 1 SN )また，前述の結合ツリーを使うアプローチとの対比で重要な特徴は，MJoin では中間結果が保持されないという(1)

SN 1 (S1 1, . . . , 1 SN −1 )ことである．必要なウインドウ領域は少なくて済む半面，

根に一番近い場所の結合演算に配置するストリームだけ上

新規データが届くたびに N-1 回のプローブ処理が必要に記方針で決定し，残りの部分は従来通りの問合せ最適化手なる．なお，MJoin の改良手法の一つとして中間結果を法を用いて最適な木の構造を決定する．このようにするこ

キャッシュするものも提案されている [17] が，本研究ではとで，i 番目の結合ツリーは各ストリーム Si の処理遅延のキャッシュについては考慮しない．

最小化に特化したものとなる．

3. 提案内容: 並列ウインドウ結合

また，N 個のツリーからの出力をすべて受け取ってしまうと処理結果に重複が発生してしまうので，ストリーム Si

本節では，本研究で提案する処理遅延を削減するためののタプル ti が到着した場合には，i 番目の結合ツリーから

並列結合演算処理方式 2 種類について述べる．2 節で述べの処理結果だけを受け取るようにする．i 番目以外の結合

た通り，N 入力のストリームを統合する方式は，結合ツツリーは結合処理はするが処理結果を最後の出力へは送ら

c 2012 Information Processing Society of Japan⃝4

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

OutputOutputSi

S1CPU䝁䜰1

CPU䝁䜰i⋈⋈

Selector

Selector

SelectorCPU䝁䜰N⋈

SN

CPU䝁䜰1

⋈

W1
⋈

⋈

⋈
S2 S3 S4

S1

図 5

䝡䝹䝗

Si

S1 S2 S3

SN

OutputSelector⋈

Selector

S1⋈⋈

Selector

S2

⋈

⋈

䝥䝻䞊䝤

copy

CPU䝁䜰N+1

t1
᪂つ฿╔
䝍䝥䝹

並列結合ツリー方式

⋈

⋈
⋈
⋈

⋈

S1

S2

S3

図 7

並列 MJoin 結合

SN

理遅延を大幅に減らすことができる．並列 MJoin 方式はSelector

S3

WN

W3
䝥䝻䞊䝤 䝥䝻䞊䝤⋈S 1 S2 S3CPU䝁䜰N+2
CPU䝁䜰N

W2

⋈

⋈

merge
CPU䝁䜰2S4⋈⋈

すべての結合要求に対して適用できるわけではないが，制限については 3.2.1 節で述べる．
並列 MJoin 方式の処理手順について述べる (図 7)．スト
リーム Si からタプル ti が新規到着した場合，まず ti を N
個のスレッドの処理キューへ追加する (Copy 処理)．各ス

S2 S3 S 4 S 1

図 6

S1 S 4 S 3 S 2

S 1 S 4 S2 S3S 2 S 3 S1 S 4

ඹ㏻㒊ศᮌ1

ඹ㏻㒊ศᮌ2並列結合ツリー方式における共通部分木の共有

レッドでは以下の処理を実行する．

• ストリーム Si のウインドウ領域とハッシュ表に対するビルド処理 (1 スレッド)．ビルドが終わったら，ti単体からなる集合 Mi を返す．

ずに破棄する．図 5 における Selector 演算がその役割を行• Si 以外のストリーム Sj (i ̸= j) のハッシュ表とウイう．このようにしておくことで，どのストリームのタプル

ンドウ領域に対するプローブ処理 (N-1 スレッド)．たが到着しても最も少ない処理を行うツリーから処理結果をだし，プローブ処理の結果として，通常の統合結果

受け取ることができるため，処理遅延を最小化できる．

(ti × wj ) ではなく，ti とマッチした wj の集合を返す．

3.1.1 メモリ使用量の削減について

すなわち半結合 (semi-join) として動作する．処理結

N 個の結合ツリーを完全に独立で実行することも可能で

果は Mj と表記する．

はあるが，実際にはウインドウ領域などに必要なメモリ使

N 並列の実行の結果，N 個のタプル集合 M1 , . . . , MN が得

用量などが N 倍になってしまい，都合が悪い場合も考えらられる．最後にこれらを統合 (M1 × . . . × MN ) して結合演れる．そのような場合は，N 個の結合ツリーの間での共通算の処理結果の生成を行う (Merge 処理)．

部分木の共有化を行う．最も浅い結合演算以外の部分は，並列 MJoin では，入力ストリーム数 N に対して，ビル

N 個のツリーの間で共通部分が存在するため，重複した中ド処理に 1 スレッド，プローブ処理に N-1 スレッド，Copy

間結果を作らずに済ますことが可能である．
図 6 は，4 つの入力ストリーム S1 ,S2 ,S3 ,S4 に対して，並処理に 1 スレッド，Merge 処理に 1 スレッドが必要なため，
合計 N+2 スレッドを同時実行できるだけのコア数が必要

列結合ツリー方式を適用した場合である．ここでは S2 1 S3となる．

と S1 1 S4 が共通部分木として 2 回ずつ出現しており，こ3.2.1 並列 MJoin 方式の制限事項

れらは共有化が可能である．

ここでは並列 MJoin 方式が適用可能なクエリについて述べる．

3.2 並列 MJoin 方式

並列 MJoin 手法が適用できるクエリの前提として，すべ

ここではマルチコア環境での並列実行により，MJoin のてのストリームの間に結合条件があり，かつ，それらは同遅延時間を削減することを考える．2.3.2 で述べた通り，

じ結合キーを用いていることが必要となる．すなわち以下

MJoin では N-1 回のプローブ処理を順番に行っていた．このような結合条件のときとなる．

れを N 個のコアを用いて 1 回のビルド処理と N-1 回プロー
ブ処理を並列に実行し，最後に一か所に処理結果を集めて

S1 .k1 = S2 .k2 = . . . = SN .kN(2)

統合するというアプローチを考える．理想的にはプローブ

ただし，クエリ内で明示的に記述されている結合条件だけ

処理 1 回分の時間ですべてのプローブ処理が完了でき，処でなく，推移規則を使って導出可能なものも含めたときに，

c 2012 Information Processing Society of Japan⃝5

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report
hasNext()
next()
get(int index)䜲䝔䝺䞊䝍 (䜹䞊䝋䝹) 䜲䞁䝍䞊䝣䜵䞊䝇

M1
p1

M2

㼀㻿

㻷㻝

㻭㻝

㻡

㼗㻝

㼍㼎

p2

MN

㼀㻿

㻷㻞

㻭㻞

㻝

㼗㻝

㼍㼍

㻟

㼗㻝

㼍㼎

pN

㼀㻿

㻷㻺

㻭㻺

㻞

㼗㻝

㼍㼚

㻠

㼗㻝

㼍㼎

CPU

表 1 実験環境
AMD Opteron 6174 (2.2GHz, 12 コア) x 4

RAM

32GB

OS

Linux Kernel 2.6.35

Java

JDK(Oracle) 1.7.0 5

ストリームに対する結合処理である．N は 5, 10, 15, 20 と
⤖ྜ⤖ᯝ
䝍䝥䝹㞟ྜ

図 8

M1㽢M2㽢…㽢MN

マージ結果タプル集合の実装

した．ウインドウの時間指定は全ストリームとも同じ長さとし，今回は 1〜5 秒の長さのウインドウを用いた．結合
条件は 3.2.1 節の制約を満たすものを用いた．

上記が成り立っていればよい．
それでも，ユーザから与えられる結合要求が，そのまま
上記の制約を満すことを期待するのは難しい．しかし，制

今回はストリーム処理エンジンを用いずに，結合処理だ
けを実行可能な独立のものを Java によって開発した．評
価対象は以下の 4 つである．

約を満たさないストリームを一部除いた部分集合において，• 結合ツリー方式: 2.3.1 節で述べた方式で，シングルス

上記制約を満たすことは十分にありえる．そのようなクエ

レッドで動作する．ツリーの生成は，クエリに記述さリを含む場合の一般的な処理手順は以下のようになる．

れたストリームの出現順序のまま left deep ツリーを生( 1 ) まず与えられた入力ストリームを分析し，上記制約を成する．今回はストリームごとに入力レートや選択率満たすストリームの集合とそれ以外のストリームの集の違いがないので，結合順序による差はほぼ生じない．

合に分ける．

( 2 ) 制約を満たすストリーム集合のプローブ処理だけを並
列 MJoin 方式によって並列実行する．

( 3 ) 並列 MJoin の結果を使って，残りのストリームへのプ
ローブ処理を従来の MJoin 同様に逐次実行する．

3.2.2 Merge 処理の効率化
最後の Merge 処理は，論理的には N 個のビルド・プロー
ブの結果のタプル集合の直積 M1 × . . . × MN を生成する作• 並列結合ツリー方式: 3.1 節で述べた提案方式で，ストリーム数 N に対して N 個の結合ツリーを作り，N ス
レッドで動作する．今回の実装では共通部分木の共有化は行っていない．

• MJoin 方式: 2.3.2 節で述べた方式で，シングルスレッ
ドで動作する．今回の実装では，プローブ処理の順番はクエリで与えられたときのストリームの順番のまま行うものとした．

業に相当するが，実装上はこれをマテリアライズするコス• 並列 MJoin 方式: 3.2 節で述べた提案方式で，N+2 ス

トは高い．そこで，実装上の工夫として，直積結果を実際

レッド (ストリーム数 N+copy+merge) で動作する．

には生成せずに，内部で N 個のタプル集合のまま保持し，

N 個のポインタを動かして，外側からは見かけ上直積結果があるように振る舞わせるようにした (図 8)．

4 つともハッシュ結合をベースに実装されている．
処理遅延の計測は，タプルが生成されてから結合処理を終えて出力結果として出てくるまでの時間とした．タプル

直積結果に対する操作は，イテレータを用いて間接的にを生成する専用のスレッドを用意し，タプルを生成した時行わせるようにしており，イテレータの next() メソッド

刻をそのタプルのタイムスタンプとして登録する．結合結で次のタプルのデータが要求された場合，内部のポインタ

果を受け取る側も別スレッドで動作し，受け取った時刻とを移動させて，次の直積結果のタプルがあるように見せか

タプル内のタイムスタンプを比べて，処理遅延を計算する．

ける．

現在時刻の取得には Java の System.nanoTime() メソッド

4. 評価実験を使用した．今回はデータを 20 秒間流し続けたときに，各処理結果生成時に発生した処理遅延の平均を測定した．

提案手法の有効性を検証するために評価実験を行った．
実験環境は表 1 の通りであり，48 コアを持つマシンを用いた．

4.1 実験結果
ウインドウサイズを 1 秒に固定し，入力ストリームの数

実験データは，ストリームデータ，クエリともに人工的を 5, 10, 15, 20 と変えた時の処理遅延の平均値を図 9 に示に生成したものを用いた．ストリームデータは 10 個の数

す．横軸は入力ストリーム数，縦軸は処理遅延 (ナノ秒) で

値属性からなるタプル列で，入力レートは各ストリームとある．並列結合ツリー方式は，入力ストリーム数 5 から 15もに毎秒 20 タプル (到着間隔 50ms) とした．結合条件のまでは最も遅延が少なくなった．並列 MJoin 方式は，今ためのキーとなる属性の値は，ストリームごとにカウンタ回の実験では残念ながらどの場合でも最も遅い結果となっを用いて 1 ずつカウントアップして値を生成し，それ以外

た．これらの原因については次節で考察する．

の属性の値は乱数を用いた．実験に用いたクエリは N 個の

c 2012 Information Processing Society of Japan⃝

次に，入力ストリームの数を 10 に固定し，ウインドウ

6

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

回ってくる周期が長くなってしまうため，処理遅延の増加ฎ⌮㐜ᘏ(ns)
100000につながってしまったと考えられる．この部分の実装改善

90000
80000
70000

⤖ྜ䝒䝸䞊䠄䝅䞁䜾䝹䝇䝺䝑䝗䠅

60000

୪ิ⤖ྜ䝒䝸䞊

50000
Mjoin㸦䝅䞁䜾䝹䝇䝺䝑䝗䠅

40000

が今後の課題であることが分かった．

5. 関連研究୪ิMjoin

30000

多入力のウインドウ結合についてはいくつかの改良手法

20000
10000

が提案されている．Adaptive Caching [17] は，MJoin を

0
5

10

図 9

15

20

ධຊ䝇䝖䝸䞊䝮ᩘ

入力ストリーム数に対する処理遅延改良するための手法の一つで，MJoin のプローブ処理の過程で生成された中間結果をキャッシュする．GrubJoin [18]も多入力を扱う結合演算の 1 つであり，特に高負荷時に必ฎ⌮㐜ᘏ(ns)

要性の低いデータの処理を間引く Load shedding のアイデ

35000

アが入っている．これらの改良された手法は，本研究の提

30000
25000

⤖ྜ䝒䝸䞊䠄䝅䞁䜾䝹䝇䝺䝑䝗䠅

20000

୪ิ⤖ྜ䝒䝸䞊

15000

Mjoin㸦䝅䞁䜾䝹䝇䝺䝑䝗䠅

10000

୪ิMjoin

案手法と対立するものではなく，組み合わせて使用することも可能である．
分散ストリーム環境の前提にした研究も数多くされてい5000

る．MJoin を分散環境で実行するものとして，PMJoin[19]01000ms2000ms

図 103000ms

4000ms

5000ms䜴䜲䞁䝗䜴䝃䜲䝈

ウインドウサイズに対する処理遅延が提案されている．PMJoin では，各ストリームのウイン
ドウとハッシュ表を別々のノードで管理しており，到着タ
プルがプローブ処理のために各ノードを巡回する順序を，

サイズを 1〜5 秒に変えた時の処理遅延の平均値を図 9 にタプルのハッシュ値ごとに別々に設定することができる．

示す．ただし，今回の実験ではウインドウサイズを変えた

この研究は，不要な中間結果の削減によるスループットのことによる影響はほぼ見られなかった．これはハッシュ結向上を目的としているが，分散環境ではノードを巡回する

合の場合，ウインドウに含まれるすべてのタプルとマッチ際にネットワーク通信コストが発生してしまうため，処理ングを行わず，ハッシュ値が一致するものとだけマッチン

遅延の削減の目的にはあまり適していない．

グが行われるため，今回のキー値の設定ではウインドウサ
イズ設定の影響を受けなかったための考えられる．

Chase execution[20] は，Active stanby 方式で冗長化された分散ストリーム環境において同じクエリを異なる配置
プランで複数実行し，より早く生成された方の処理結果を

4.2 考察採用するという実行方式である．本研究の並列結合ツリー

並列結合ツリー方式は，図 9 でストリーム数 5 から 15方式も Chase Execution と同じ発想に基づいているが，本までは最も遅延が少ないが，20 の時にシングルスレッド

研究はマルチコア環境を前提としており，処理の冗長化はの MJoin に追い抜かれ，シングルスレッドの結合ツリー方

考慮されない代わりに，複数プランの間での共有化などが式にも追いつかれている．これは並列結合ツリー方式の場

可能な点が異なっている．

合，同じクエリを N 個同時に処理することとほぼ同じリ

我々の既存研究では，マルチコア環境を想定した低遅延

ソースを消費するため，他の手法に比べて並列度に応じてストリーム処理方式 [21] を提案しているが，この中は結合動的にオブジェクトを生成・破棄する回数が増え，ガベー演算は並列実行の対象とはなっていない．

ジコレクションにかかるコストが増加したことが原因と考えられる．改善策の一つとして，3.1.1 節で述べた共通部分
木の共有機能を実装することが挙げられる．
並列 MJoin 方式は，今回の実験では残念ながら最も遅6. おわりに本研究では，分散環境の代わりにマルチコア環境を用いて，複数ストリームに対するウインドウ結合演算の処理遅い結果となった．原因としては，データの受け渡しに用い延を改善するような並列化方式を 2 つ提案した．評価実験ているキューの管理方式の問題点が考えられる．今回の実により，並列結合ツリー方式については処理遅延の削減を装では，copy 処理のスレッドから各プローブ処理のスレッ確認できたが，並列 MJoin 方式は逆に最も遅くなるというドへのデータの受け渡しと，各プローブ処理のスレッドか結果になった．ただし，両手法ともまだ実装の問題で理論

ら merge 処理のスレッドへのデータの受け渡しに 2N 個の上の性能を出し切れていない部分があり，それについてはキューを用いている．今回はこれらのキューに新しいデー

今後の課題となった．またより多様な状況下での評価実験

タが追加されていないかを監視するアルゴリズムにラウは必要であり，それについても今後の課題である．

ンドロビン方式を採用したが，しかしこの監視方法では，謝辞 本研究の一部は日本学術振興会科学研究費補助金

キューの数が増えるほどキューの中身を確認する順番が基盤研究 A（#22240005），若手研究 B（#24700087）の助

c 2012 Information Processing Society of Japan⃝7

Vol.2012-DBS-154 No.10
Vol.2012-IFAT-107 No.10
2012/8/2

情報処理学会研究報告
IPSJ SIG Technical Report

成により行われた．

[13]

参考文献
[1]
[2]

[3]

[4]

[5]

[6]

[7]

[8]

[9]

[10]

[11]

[12]

Twitter, http://twitter.com/.
Daniel J. Abadi, Yanif Ahmad, Magdalena Balazinska,
Ugur Çetintemel, Mitch Cherniack, Jeong-Hyon Hwang,
Wolfgang Lindner, Anurag Maskey, Alex Rasin, Esther
Ryvkina, Nesime Tatbul, Ying Xing, and Stanley B.
Zdonik. The design of the borealis stream processing engine. In CIDR, pages 277–289, 2005.
Sirish Chandrasekaran, Owen Cooper, Amol Deshpande,
Michael J. Franklin, Joseph M. Hellerstein, Wei Hong,
Sailesh Krishnamurthy, Samuel Madden, Vijayshankar
Raman, Frederick Reiss, and Mehul A. Shah. Telegraphcq: Continuous dataﬂow processing for an uncertain world. In CIDR, 2003.
Rajeev Motwani, Jennifer Widom, Arvind Arasu, Brian
Babcock, Shivnath Babu, Mayur Datar, Gurmeet Singh
Manku, Chris Olston, Justin Rosenstein, and Rohit
Varma. Query processing, approximation, and resource
management in a data stream management system. In
CIDR, 2003.
Mohamed H. Ali, Ciprian Gerea, Balan Sethu Raman,
Beysim Sezgin, Tiho Tarnavski, Tomer Verona, Ping
Wang, Peter Zabback, Anton Kirilov, Asvin Ananthanarayan, Ming Lu, Alex Raizman, Ramkumar Krishnan, Roman Schindlauer, Torsten Grabs, Sharon
Bjeletich, Badrish Chandramouli, Jonathan Goldstein,
Sudin Bhat, Ying Li, Vincenzo Di Nicola, Xianfang
Wang, David Maier, Ivo Santos, Olivier Nano, andStephan Grell. Microsoft cep server and online behavioral targeting. PVLDB, 2(2):1558–1561, 2009.
Leonardo Neumeyer, Bruce Robbins, Anish Nair, andAnand Kesari. S4: Distributed stream computing platform. In Wei Fan, Wynne Hsu, Geoﬀrey I. Webb, Bing
Liu, Chengqi Zhang, Dimitrios Gunopulos, and Xindong
Wu, editors, ICDM Workshops, pages 170–177. IEEE
Computer Society, 2010.
Bugra Gedik, Henrique Andrade, Kun-Lung Wu,
Philip S. Yu, and Myungcheol Doo. Spade: the system s
declarative stream processing engine. In Jason Tsong-Li
Wang, editor, SIGMOD Conference, pages 1123–1134.
ACM, 2008.
Jaewoo Kang, Jeﬀrey F. Naughton, and Stratis Viglas.
Evaluating window joins over unbounded streams. In
ICDE, pages 341–352, 2003.
Ying Xing, Jeong-Hyon Hwang, Ugur Çetintemel, andStanley B. Zdonik. Providing resiliency to load variations in distributed stream processing. In VLDB, pages
775–786, 2006.
Ying Xing, Stanley B. Zdonik, and Jeong-Hyon Hwang.
Dynamic load distribution in the borealis stream processor. In ICDE, pages 791–802. IEEE Computer Society,2005.
Mehul A. Shah, Joseph M. Hellerstein, Sirish Chandrasekaran, and Michael J. Franklin. Flux: An adaptive
partitioning operator for continuous query systems. In
Umeshwar Dayal, Krithi Ramamritham, and T. M. Vijayaraman, editors, ICDE, pages 25–36. IEEE Computer
Society, 2003.
Tyson Condie, Neil Conway, Peter Alvaro, Joseph M.
Hellerstein, Khaled Elmeleegy, and Russell Sears. Mapreduce online. In NSDI, pages 313–328. USENIX Association, 2010.

c 2012 Information Processing Society of Japan
⃝

[14]

[15]

[16]

[17]

[18]

[19]

[20]

[21]

Peter R. Pietzuch, Jonathan Ledlie, Jeﬀrey Shneidman, Mema Roussopoulos, Matt Welsh, and Margo I.
Seltzer. Network-aware operator placement for streamprocessing systems. In Ling Liu, Andreas Reuter, KyuYoung Whang, and Jianjun Zhang, editors, ICDE,
page 49. IEEE Computer Society, 2006.
Evangelia
Kalyvianaki,
Wolfram
Wiesemann,
Quang Hieu Vu, Daniel Kuhn, and Peter Pietzuch.
Sqpr: Stream query planning with reuse. In Serge Abiteboul, Klemens Böhm, Christoph Koch, and Kian-Lee
Tan, editors, ICDE, pages 840–851. IEEE Computer
Society, 2011.
Stratis Viglas, Jeﬀrey F. Naughton, and Josef Burger.
Maximizing the output rate of multi-way join queries
over streaming information sources. In VLDB, pages
285–296, 2003.
Ahmed Ayad and Jeﬀrey F. Naughton. Static optimization of conjunctive queries with sliding windows overinﬁnite streams. In Gerhard Weikum, Arnd ChristianKönig, and Stefan Deßloch, editors, SIGMOD Conference, pages 419–430. ACM, 2004.
Shivnath Babu, Kamesh Munagala, Jennifer Widom,
and Rajeev Motwani. Adaptive caching for continuous
queries. In ICDE, pages 118–129, 2005.
Bugra Gedik, Kun-Lung Wu, Philip S. Yu, and Ling Liu.
Grubjoin: An adaptive, multi-way, windowed stream
join with time correlation-aware cpu load shedding.
IEEE Trans. Knowl. Data Eng., 19(10):1363–1380,
2007.
Yongluan Zhou, Ying Yan, Feng Yu, and Aoying Zhou.
Pmjoin: Optimizing distributed multi-way stream joins
by stream partitioning. In DASFAA, pages 325–341,
2006.
上田高徳, 打田研二, 秋岡明香, and 山名早人.
データストリーム処理におけるレイテンシ削減と高可用性のためのオペレータ実行方法. 日本データベース学会論文誌, 10(3):1–6, 2012.
渡辺陽介 and 横田治夫.
マルチコア環境における可換演算群の並列評価による低遅延ストリーム処理方式. In WebDB
フォーラム, 2011.

8