DEIM Forum 2014 D3-1

ストリーム処理エンジンを用いたストリームデータの OLAP 処理
中挾

晃介†

北川

博之††

天笠 俊之††† 筑波大学システム情報工学研究科コンピュータサイエンス専攻 〒 305–8573 茨城県つくば市天王台 1-1-1†† 筑波大学システム情報系情報工学域 〒 305–8573 茨城県つくば市天王台 1-1-1
E-mail: †nakabasami@kde.cs.tsukuba.ac.jp, ††{kitagawa,amagasa}@cs.tsukuba.ac.jp
あらまし

センサデータやログデータやマイクロブログ等，連続的に生成，配信されるストリームデータの増加に伴い，その処理を行うストリーム処理エンジンが開発されている．ストリームデータに対する分析としては様々なものがあるが，その一つに多次元分析に基づく OLAP 処理がある．これまでストリームデータの OLAP 処理については

J. Han らの研究があるが，ストリーム処理エンジンの利用は想定されていない．このため，OLAP 処理のために個別のシステムを開発する必要があったり，他のストリームデータ応用との連携が十分にとれないなどの問題がある．本
研究では，ストリーム処理エンジンが持つ集約計算機能を用いてストリームデータに対する OLAP 処理を実現するための手法とその処理の効率化手法について検討する．
キーワード

ストリーム処理エンジン，OLAP 処理

1. は じ め にの対応する全ての問合せをストリーム処理エンジンに登録する
ことは，処理性能上難しい場合が多い．また，必ずしも全ての

近年，センサデータやログデータやマイクロブログ等，際限頂点に対応する集約値を常時取得し続けなくても，ユーザ要求

なく，連続的に生成され，配信されるようなストリームデータに応じて導出できれば良いという応用も多いと考えられる．そが増加してきている．このストリームデータを処理するスト

こで本研究では，lattice 内の一部の頂点のみを適切に選択し連リーム処理エンジンが多数開発されている．ストリーム処理エ

続的問合せとしてストリーム処理エンジンに登録することで，ンジンは，その専用問合せ言語で記述された問合せをエンジン

効率的な OLAP 処理を実現することを目的とする．また，登に登録することにより，ストリームデータに対して登録した問録する連続的問合せからなる効率的な処理パイプラインの構成

合せを連続的に実行できる．一方，ストリームデータに対して法についても検討する．

より高いレベルの分析 (データの傾向や外れ値など) を行いた
いというニーズが増加している．このような分析の代表例と2. 基 本 事 項

して，データを多次元のデータキューブとして捉えて分析を行

2. 1 OLAP

う多次元データ分析 (OLAP 処理) がある．ストリームデータ

OLAP [2] は，データウェアハウス内のデータに対して多次に対して OLAP 処理を行う研究として，Jaiwei Han らはスト

元的なデータ分析を行うために，データを多次元データモデル

リームデータに対する OLAP 処理を容易にすることを目的としとして扱い処理する．OLAP は，その前処理として，分析の軸た Stream Cube と呼ばれるアーキテクチャを提案している [1]．

となる複数の次元表と，その分析対象のデータが格納された事

ただし，ストリームデータを OLAP 処理することは可能となっ実表から，データキューブと呼ばれる多次元データモデルを構ているものの，ストリーム処理エンジンを生かした OLAP 処

築する．データキューブはメジャーと呼ばれる分析対象となる理については十分検討が行われていない．このため，OLAP 処

数値の集合と，それに属する複数の軸で構成される．このデー

理のために個別のシステムを一から開発する必要があったり，

タキューブに対して，ロールアップやドリルダウン等の操作を他のストリームデータ応用との連携が十分にとれないなどの問行うことができる．ここでは，データの例として TPC-D ベン

題がある．

チマーク [3] に準じたデータを用いる．今，商品テーブル，顧

そこで本研究では，ストリーム処理エンジンを用いた OLAP

客テーブル，供給者テーブルの三つのテーブルと，ある期間に処理を実現するための手法を検討する．具体的には，ストリー

顧客が発注した部品の情報 (売上情報) を表すデータキューブをム処理エンジン側に連続的問合せを登録しておき，これらの登構築することを考える．この例では，商品，顧客，供給者が次録しておいた連続的問合せから OLAP 処理の結果として必要元表となり，売上情報が事実表となる．また，売上情報の要素なデータを得る．OLAP 処理では，分析対象のデータの次元である売上額がメジャーとなる．次元表と事実表の関係は図 1(属性) や次元内の階層のすべての組合せを頂点とする latticeのようなスタースキーマと呼ばれる構造で表せる．これらの表を考える．lattice の各頂点は，OLAP 処理において集約するから，図 2 のようなデータキューブが構築される．また，図 2

対象の次元の組合せに対応する．最も単純には，lattice 内の全に対して顧客で Roll-up を行い，顧客の住所 (県) で集約を行っての頂点の次元の組合せの集約を実行すればよい．しかし一般

た例，および，供給者で Drill-down を行い，供給者の中のどのには，lattice の全頂点数は膨大な数になることが多く，これら部門かを示した例を図 3 に示す．

売上情報
商品ID
供給者ID
顧客ID
売上額

1

28

769 289

300

2

商品

514
46

740

242

23

680

3

134

顧客
顧客ID
顧客名
住所

856 343463
271 857
610
565
370

96
796

75

576

86

972 751

6 5 4

商品
商品ID
商品名
ジャンル
供給者
供給者ID
供給者名

430 431 747

573 226367

878 603 736 912

605

246 893593

954 624

佐藤 鈴木 高橋 田中 渡辺

顧客

図 1 スタースキーマ

図 2 データキューブ

856 343 463
96 610857
271
796 565370

品
商1311 589
1028 703

737 1723

茨城 埼玉

顧客

品
商

や，連続的問合せ間の依存関係に着目した，OLAP 処理におい

593

供給者で
drill-down

514 28 769 289 300
46 740 242 23 680
75 576 86 972 751
430 431 747 573 226
367 878 603 736 912
605 246 893 954 624

佐藤

鈴木

高橋

顧客

渡辺

田中

C_1
B_3
B_2

A_3

100
140

B_1

130
89

A_2

てコアとなる連続的問合せ群である．これらに関しては，以降の章において詳述する．
例えば，商品ごとの直近一時間の売上を顧客の住所毎にモニタしたい場合は，以下のような連続的問合せをストリーム処理
エンジンに登録する．
Select 商品 ID, 顧客住所, Sum(売上額)

100

A_1

6 5 4 3 2 1

6 5 4 3 2 1

顧客で
roll-up

6 5 4 3 2 1

134

100 150

品
商

From 売上情報 [Range 1 hour]Group By 商品 ID, 顧客住所

佐藤

鈴木

高橋

田中

渡辺 顧客

図 3 データキューブに Roll-up および Drill-down を行った例

3. 関 連 研 究
3. 1 Stream CubeJiawei Han らは，ストリームデータの多次元分析を容易に静的なデータに対する OLAP 処理は，以上で述べたような手法で行われる．一方で，静的なデータとストリームデータの間のデータ処理における違いは，後者は，大量に生成され，入力から出力に向かい動的に流れ，急速に変化するということである．そのため，ストリームデータに対する OLAP 処理は，時々刻々と到着するデータを用いた最新の分析結果を示す必要があり，また，その効率化が望まれる．

2. 2 ストリーム処理エンジン
ストリーム処理エンジンは，際限なく到達するストリーム
データに対して長期間連続的に問合せ処理が可能なように開発されたものである．このようなストリーム処理エンジンには，STREAM [4]，Storm [5]，S4 [6]，Borealis [7] などが挙げられる．
ストリーム処理エンジンでは，ストリームデータに対し CQL等の言語を用いて問合せを行う．このような言語で書かれた問合せ (連続的問合せ) をストリーム処理エンジンに登録することにより，ストリームデータに対して連続的に問合せを行うことができる．
本研究では，OLAP 処理の一部をストリーム処理エンジンで行うことにより，より効率的なストリームデータに対するOLAP 処理を行う．つまり，ストリーム処理エンジンの機能を用いて必要なデータの集約を行う．具体的には，データの次元，
あるいは階層に対して集約を行う問合せの一部を連続的問合せとしてストリーム処理エンジンに登録する．ここで，登録する
連続的問合せは，ユーザが常にその結果を欲しいような問合せする，Stream Cube と呼ばれるアーキテクチャを提案している [1]．この Stream Cube の計算のために三つの技術を考案している．
一つ目は，tilted time frame である．これは，ストリーム
データ内の時間次元に着目し，最近のデータはより細かい粒度(15 分毎等) で，時間が経ったデータは集約してデータの粒度を粗く (1 時間や 1 週間等) して保持することにより，総データ量を小さくする手法である．
二つ目は，two critical layers である．これは，データの次
元やその階層全ての組合せを実体化するのではなく，minimal

interesting layer と呼ばれる，分析者が興味のある最低限の次元の集約の組合せと，observation layer と呼ばれる，分析者がその次元の集約の組合せからドリルダウンするための層の二つの層のみを実体化することにより，空間的コストを削減する手法である．
三つめは，popular path approach である．これは，実体化した二層間の中間層を計算するための一つのパスを用意し，このパスに沿ったドリルダウンにより中間層を計算する手法である．このパスは，H-tree [8] と呼ばれるデータ構造を用いて格納する．
この StreamCube の手法は，ストリームデータに OLAP 処
理を適用する手法であるが，一方で時間次元に対する集約の方法が事前に決められているため，時間に対する柔軟な問合せを書くことができない．また，実体化した二層間のパスをユーザが決める必要があるため，その決定に知識を要するという欠点

入力tid: 5, cid: 3, sales: 100

window

tid: 4, cid: 2, sales: 250
tid: 3, cid: 1, sales: 300
tid: 2, cid: 2, sales: 150

シノプシス

tid: 1, cid: 1, sales: 100

集約

tid: 4, cid: 2, sales: 250
tid: 3, cid: 1, sales: 300

tid: g4, cid: 2, sales: 400

tid: 2, cid: 2, sales: 150

tid: g3, cid: 1, sales: 300+, tid: g4, cid: 2, sales: 400-, tid: g2, cid: 2, sales: 150+, tid: g3, cid: 1, sales: 300-, tid: g1, cid: 1, sales: 400

出力
図 4 ストリームデータ処理における集約処理がある．さらに，ストリーム処理エンジンの利用はまったく考

また，ストリームデータ処理では，演算の結果を中間データと慮されていない．

して保持しておき，新規の入力タプルが入ってきた時，差分計

3. 2 静的なデータに対するデータキューブの部分的実体化

算によって問合せ結果を得る．この際，window 内に新たに到OLAP 処理では，様々なレベルのデータ分析を行うため，集

着したタプルによって生成されたタプルにプラス，window か

約演算が多用される．この集約問合せの応答時間を小さくする

らはみ出したタプルに起因するタプルにマイナスの記号をつけ，手法として問合せ結果を実体化することがあげられる．しかし，出力とする．中間データを保持しておく領域をシノプシスと呼全問合せの実体化には非常に大きな空間的コストがかかる．まび，プラスが付与された出力をプラスタプル，マイナスが付与

た，問合せ間には，ある問合せの結果が，他の問合せの結果かされた出力をマイナスタプルと呼ぶ．

ら導出することができるというような依存関係が存在する．そここで，ストリームデータ処理においてどのように集約演算

こで，Venky Harinarayan らは，そのような依存関係を表すが実行されるかについて図 4 を用いて説明する．選択演算や結lattice フレームワークと，実体化する一部の問合せを適切に選合演算等の他の演算の処理については，A. Arasu らの CQL にぶためのアルゴリズムによって，問合せの応答時間を削減する

ついての論文 [10] を参照されたい．図 4 は，入力タプルが tid，手法を提案している [9]．

cid，sales の三つの属性を持ち，cid を集約属性として sales を

lattice フレームワークは，ある問合せが，他の問合せの結果合計する例である．なお，window 幅はタプル 3 つ分であり，から答えられるといった依存関係を示す．例えば，ある問合せ

今，window 内にある演算対象データが tid の 1 から 3 から，

a が，別の問合せ b の結果から導出できる場合，a から b に上

2 から 4 にスライドされた状態とする．この状態から，ストに向かってパスが存在する．これにより，全ての次元によってリームデータ処理における集約演算を説明する．まず，指定の集約される問合せを top とし，まったく集約されない問合せを

window 幅に含まれる入力ストリームに対して集約属性についbottom とする lattice が構築される．

て集約演算を行う．この時，シノプシスには，演算対象となっHarinarayan らのアルゴリズムは，構築された lattice を元た元の入力データと集約結果のデータの両方を保持しておく．

に，空間的コストを与えられた制約の範囲内とし，かつ，時間

これは，元の入力データを保持していなければ，次の window

的コストが小さくなるように，実体化する問合せの組合せを選演算において正しく集約演算ができないためである．図 4 のぶアルゴリズムである．

4. ストリームデータ処理における集約演算

例では，tid が 2 から 4 のタプルを cid について集約する．これにより，cid が 1 の集約結果が 300，cid が 2 の集約結果が400 と計算され，これらの tid として新たに g3，g4 が振られ従来のデータベース処理では，クエリ発行時に，DB に格納

る．ここで，先ほど tid が 1 から 3 のタプルにおいて計算したされたデータをバッチ処理などで一括処理を行うのに対し，ス

集約結果のタプルである g1，g2 が今回の集約演算の結果によ

トリームデータ処理は，データ到着時に事前に登録したクエリ

り置き換わる．よって，シノプシスに tid が 2 から 4 の元デーを実行する，差分処理を行う．そのため，ストリームデータ処

タのタプルおよび tid が g3 と g4 の集約結果のタプルが格納さ理では，連続的に到着する入力ストリームに対し，最も新しいれ，これら g3，g4 にプラス，直前の集約演算の結果である g1，

タプルから一定の件数のタプルを演算対象として処理を行う．

g2 がマイナス付きで出力される．このように，集約演算の場

この件数を window 幅と呼び，この処理を window 演算と呼ぶ．

合，window に新しく入ったタプルと追い出されたタプルのぞ生データ (1)
商品名 (0.5)

生データ (1)
顧客名 (0.7)（商品）ジャンル (0.1)（顧客）住所 (0.3)

none (0.001)（商品）(0.85)（商品名, 顧客名）

(0.73)(0.65)（ジャンル, 顧客名）

（商品名, 住所）

(0.7)
（顧客名）

(0.3)

none (0.001)（顧客）(0.37)（ジャンル, 住所）

(0.1)

（住所）

（ジャンル）(none)

図 5 商品および顧客の lattice
図6

(0.5)
（商品名）

(0.001)

商品および顧客を統合した lattice

れぞれの集約属性の集約結果の更新により，プラスタプルとマ

本手法において，処理にかかる全体のコストは lattice 内のイナスタプルの組が 2 つ，計 4 タプルが生成される．他にも，それぞれの集約問合せにおいてかかるコストの総和と考える．

window に新しく入ったタプルと追い出されたタプルぞれぞれ

図 6 のような lattice 内には，ストリーム処理エンジンに登録の集約属性が同じ場合は，プラスタプルとマイナスタプルの組する集約問合せと，ストリーム処理エンジンに直接集約問合せ

が 1 つの計 2 タプルが生成され，新しく入ったタプルの集約属を登録せずに，必要に応じてストリーム処理エンジンに登録す性の集約結果がシノプシスに無い場合と，タプルが追い出される集約問合せの結果から必要とする集約結果を導出する (以降，ることによってシノプシス内にそのデータの集約属性の集約結

オンデマンドに評価すると言う) 集約問合せの二種類の問合せ

果が無くなる場合は計 3 タプルが生成される．

が存在すると考える．ここで前者を Qr = {r1 , ..., rn } とし，後5. 提 案 手 法

者を Qd = {d1 , ..., dm } とする．これらを合わせた lattice 内の集約の全ての集合を Q とする．また，lattice 内のそれぞれの

5. 1 処理の効率化

集約問合せ結果が参照されるタイミングは問合せごとに異なる．

集約処理をストリーム処理エンジンで行う場合，データの次ストリーム処理エンジンに登録する集約問合せは，新しいタプ

元，階層の全ての集約の組合せを連続的問合せとしてエンジンルが到着する度に問合せが実行されるが，オンデマンドに評価に登録することが考えられる．しかし，一般にその組合せの数する集約問合せはユーザからその集約結果が求められたときのは膨大になり，それら全てをエンジンに登録することは処理性み実行されるので，両者の間には参照される頻度に相違がある．

能上難しい．また，ユーザが取得したい集約は，それらの全て

そこで，それぞれの集約が参照される頻度をコストに重み付けではなく一部であることが大部分である．そのため，エンジンすることを考える．ここでは，Q の一つ一つの集約結果が単位にどの集約問合せを登録するかを選択する．

時間あたりに参照される平均の頻度を P とし，エンジンに登録

また，各集約間には依存関係が存在する．そのため，選択さする集約問合せの平均参照頻度を P (ri )，オンデマンドに評価れた連続的問合せを独立に評価せずに，それらを依存関係に基する集約問合せの平均参照頻度を P (di ) として重み付けを行う．

づいて結びつけ，先に評価された集約結果を利用することで，

ここからまず，ストリーム処理エンジンに登録した問合せに処理を効率化できる．そこで，各集約の評価順を表す処理のパ対する集約処理を行う単位時間あたりのコストを考える．これ

イプラインを構築する．

を C1 とする．C1 は，単位時間あたりにその集約問合せに到達

以上の二つの操作を行うために，まず，集約処理をストリー
ム処理エンジンで行う際のコストモデルを定義する．

するタプル数と集約計算自体の処理に応じたコストとなる．生
データの単位時間あたりの入力レートを i とすると，それぞれ

5. 2 コストモデルの集約問合せには，i × w のコストがかかる．w について，4

本節では，次節以降において，ストリーム処理エンジンに登節で説明したように，演算対象の window がスライドして集約録する集約問合せを決めるアルゴリズムに用いるコストモデル

結果の更新が行われると，プラスタプルとマイナスタプルが合について説明する．
まず，以下のようなスキーマを持つデータ内の二つの次元から lattice を構成することを考える．
売上情報 (商品 ID, 商品名, ジャンル, 顧客 ID,

わせて 2 個，3 個，4 個のいずれかの個数分生成される．その顧客名, 住所, 売上額)
商品はその次元内の階層としてジャンルが存在し，顧客はその次元内の階層として住所が存在するため，商品と顧客の個々のlattice は，図 5 のようになる．上から下に向かって集約度は上がり，一番下の none は，たった一つの値に集約されることを意味している．この二つの lattice を一つの lattice に統合したものが図 6 である．なお，図 5，図 6 のそれぞれの集約の肩に乗る数字については後述する．

ため，w は 2，3，4 の値をとる．以降の例では，最悪のケースを考えて w = 4 としてコストを考える．集約計算自体の処理のコストを考えると，集約処理は，タプルが到着する度に集約
属性に応じてハッシュを用いてタプルを振り分け集約すると考えるため，O(1) だけしかかからないとする．以上から，C1 は，
単位時間あたりにその集約問合せに到達するタプル数に応じたコストと考える．次に，オンデマンドに評価する集約問合せの単位時間あたりの処理コストを考える．これを C2 とする．オ
ンデマンドに評価する集約問合せは，ユーザからその集約が求められたときに限り，lattice 上で自分自身の上流にある Qr 内の問合せの中で，そのシノプシスに保持する全タプル数が最も

0.5

850

（商品名, 顧客名）

0.1

（ジャンル, 顧客名）

0.6

（顧客名）

730

0.4

0.3

（ジャンル, 住所）0.8(none)

650（ジャンル, 顧客名）

（商品名, 住所）

（住所）

図7

（商品名, 顧客名）

（商品名, 住所）

700

0.9

（顧客名）

（商品名）

300

0.2

（ジャンル）

370（ジャンル, 住所）

100

（住所）

（ジャンル）(none)

0.7

500
（商品名）

1

図 8 それぞれの集約結果のタプル数

それぞれの集約問合せの平均参照頻度少ない問合せを選び，その結果から必要とする集約結果を導出

け，ある一定の値を S が超えたときに上述のアルゴリズムを終する．そのためコストは，その連続的問合せのシノプシス内の了する．このアルゴリズムが終了した時点で，エンジンに連続

タプル数に依存する．従って C2 は，その連続的問合せの集約的問合せとして登録する集約問合せの個数ごとの最小の C の結果タプル数に自分自身の問合せの平均参照頻度 (単位時間あ値を比較し，これらの値の中で最も C の値が小さい個数個またりに自分自身の問合せが参照される回数) をかけたものに応で，エンジンに連続的問合せとして登録する．

じたコストとなる．さらに，ストリーム処理エンジンに登録した連続的問合せがそのシノプシスに持つ全タプル数を空間コス

以上のアルゴリズムを擬似コードとして表すと以下のようになる．

トとして考える．これを S1 とする．
以上から，処理コスト C および空間コスト S を以下のよう

nl = number of local_min_c;に定義する．

C =n
X

C1 (ri ) +

i=1

S =

n
X

local_min_c = {};mX

sum_of_S = 0;C2 (di )P (di )

(1)

i=1

S1 (ri )Q_r = {top aggregation};
for i=1 to all node except for top do begin
if sum_of_S > S_max then exit for loop;(2)

for j = 1 to m do begin

i=1

select that aggregation d in Q_d such

これらの式を用いて，処理コストの計算を行う．

that C is minimized;

5. 3 連続的問合せ群の決定するためのアルゴリズム

local_min_c = local_min_c union {(d,C)};

本節では，ストリーム処理エンジンに登録する集約問合せを決定するアルゴリズムについて説明する．このアルゴリズムにnl++;

ついては，3. 2 節で紹介した Harinarayan らのアルゴリズムを

sum_of_S += S;
end;

応用する．
まず，lattice のトップは必ずエンジンに登録する連続的問合せとして選ばれる．これは，ストリーム処理エンジン登録しな

end;
define number n that is nth element
whose C is minimam in local_min_c;

い集約問合せは，必要に応じてストリーム処理エンジンに登録する集約問合せの結果から必要とする集約結果を導出するため，

for i=1 to n do begin
extract d in nth (d,C) in local_min_c;

仮に lattice のトップがエンジンに登録されなかった場合，集Q_r = Q_r union {d};約結果の導出元となる集約問合せが無いことから，集約問合せを実行できないためである．次に二つ目にエンジンに登録する

end;

集約問合せを選ぶ．この選び方は，先ほど選んだ lattice の top

resulting Q_r is the continuous queries;の集約を除いた集約問合せの中から，次にエンジンに登録する
集約問合せを一つ選び，その集約問合せを連続的問合せとしてエンジンに登録した場合の処理コスト C を計算する．この計算

5. 4 処理パイプラインの決定

を，lattice の top の集約を除いた全ての集約に対して行う．こ

本節では，前節のアルゴリズムによって決定した連続的問合の結果，C の値が最も小さかったものを，次にエンジンに登録

せ群をどの順番で評価を行うかを表すパイプラインを構築する．

する集約問合せとして選ぶ．この後，三つ目にエンジンに登録

まず，lattice の top をパイプラインの root とする．次に，スする集約問合せを選ぶ．この場合も同様に，lattice の top と先トリーム処理エンジンに連続的問合せとして登録した集約問合

ほど二つ目にエンジンに登録する集約問合せとして選ばれたもせ間にエッジを張ることを考える．これは，自分の上流に存在のを除いた集約問合せの中から，C が最も小さくなるものを選する連続的問合せに向けてエッジを張る．ここで，経路の選択

ぶ．以降，この操作を繰り返し行う．この繰り返しを終了する肢が複数あり，どちらか一方に決める必要が生じた場合は，そ条件としては，S の値を用いる．S の値にある上限 Smax を設

れらの連続的問合せのシノプシス内にあるタプル数を比較し，

（商品名, 顧客名）

（商品名, 顧客名）
（ジャンル, 顧客名）
（商品名, 住所）
（商品名, 顧客名）[1h]（ジャンル, 顧客名）

（顧客名）

（商品名, 住所）（ジャンル, 住所）（商品名）（ジャンル, 顧客名）[1h]（顧客名）（ジャンル, 住所）（商品名）（商品名, 住所）[1h]（顧客名）[1h]（ジャンル, 住所）[1h]
（住所）

（住所）

（ジャンル）（ジャンル）（住所）[1h](none)（商品名）（ジャンル）[1h](none)(none)[1h]図9

構築された処理木
図 10 時間次元の集約を考慮した処理木

その数が少ない方の経路を選択するようにする．なぜなら，ま+ (850 + 730) × 0.3 + (1000 + 850) × 0.9

ず連続的問合せの入力レートは同一なので，入力レートでは比+ (850 + 730) × 0.8 + (850 + 730) × 0.2

較できない．また，タプル数が少ない方が，より粗く集約され+ (850 + 730) × 0.7ているため属性数も少なく，集約における処理が少なく済むためである．ストリーム処理エンジンに登録した連続的問合せ間= 7513にエッジを張った後，次に，連続的問合せとしてストリーム処同様に他の全ての集約について行うとそのコスト C は，以下の理エンジンに登録されなかった集約問合せも含めてエッジを張表のようになる．

ることを考える．この場合も同様に，自分の上流に存在する連
続的問合せに向けてエッジを張る．ただし，経路の選択肢が複数あり，どちらか一方に決める必要が生じた場合は，その上流

表1(ジャンル，顧客名) 7513
(商品名，住所)6645にある連続的問合せのシノプシス内にある集約結果のタプル数

(顧客名)

6840

を比較し，その数が小さい方の経路を選択するようにする．以(ジャンル，住所)6774

上で，処理のパイプラインが構築される．これにより，連続的

(商品名)

6285

問合せとして登録された問合せは毎回の演算で結果を出力し，

(住所)

6430

連続的問合せとして選択されなかった集約問合せがユーザから(ジャンル)

7400(none)

7105

要求された場合は，その祖先にある連続的問合せの結果から導出される．

5. 5 提案手法のアルゴリズムを用いた具体例

表 1 から，二つ目にストリーム処理エンジンに登録する集
約問合せとしてもっともコストが小さいものは，(商品名) であ

ここで具体的な例を用いて前節のアルゴリズムを説明する．
前提として，まず，ストリームの入力レートを 200tuple/sとする．lattice 内のパイプラインの入力レートは 400tuple/s，600tuple/s，800tuple/s がとりうるが，ここでは最悪のケースを考えて 800tuple/s とする．また，図 5，図 6 共に，生デー
タを 1 としたときのそれぞれの集約の集約率がそれぞれの肩に乗る数字で表したようになると仮定する．さらに，図 7 に，それぞれの集約問合せの平均参照頻度を，それぞれの肩に乗る
数字になると仮定する．なお，それぞれの集約問合せにおける
window 幅は 1000 とする．すなわち例えば，商品名と住所で集約する問合せは以下のようになる．
Select 商品名, ジャンル, 住所, Sum(売上額)

客名)，(商品名，住所) を実際に連続的問合せとして選択する

From 売上情報 [Rows 1000]

こととし，それらを中心に処理パイプラインを構築することをGroup By 商品名, 住所考える．例えば，(ジャンル，顧客名) は，その上流にある連続

るとわかる．よって，(商品名) を二つ目にストリーム処理エンジンに登録する集約問合せとして選択する．このように三つ目
以降も選択を続け，結果的に，(住所)，(顧客名)，(商品名，住所)，(ジャンル，住所)，(none)，(ジャンル)，(ジャンル，顧客名) の順番でストリーム処理エンジンに登録されることとなる．
次に，これらをどの順番で評価を行うかを表すパイプラインを構築する．先に決まった集約問合せの中で，最初から 5 つを選んだところで S が Smax の値を超え，アルゴリズムが終了したとする．つまり，(商品名，顧客名)，(商品名)，(住所)，(顧図 6 で示した集約率と合わせて，それぞれの集約の結果タプル

的問合せである (商品名，顧客名) に向かってエッジを張る．ま数は図 8 のようになると仮定する．この条件下において，C がた，(住所) からその上流にある連続的問合せに向かってエッジ最小となるようなエンジンに登録する連続的問合せを選ぶ．

を張ることを考えると，(顧客名)，(ジャンル，住所) への二つ

まず，lattice のトップである，(商品名，顧客名) が選ばれる．

の選択肢が考えられるが，それらのシノプシス内の集約結果タ

次に二つ目にエンジンに登録する集約問合せを選ぶ．(ジャンプル数が，(ジャンル，住所) の方が少ないため，こちらにエッル，顧客名) を選んだとき，コスト C は，以下のように計算さジを張る．このように，前節で示した規則から，処理パイプラ

れる．

インが図 9 のように構築される．なお，図 9 中で円で囲まれたC = 200 + 800 + (1000 + 850) × 0.4 + (850 + 730) × 0.6

ノードは，選択された連続的問合せ群である．

①

12000

0.5

（商品名, 顧客名）

10000

⑨

0.1

（ジャンル, 顧客名）

8000
C

6000

S

4000

④

の積算値

0.6

（顧客名）

③

2000

1

2

3

4

5

図 11

6

7

8

9

評価結果

5. 6 時間に対する集約
ストリームデータに対して OLAP 処理をし，データの傾向などの高いレベルの分析を行う場合，時間に対して集約を行いたいというニーズがある．ストリームデータを時間に対して集約処理を行うために，それぞれの集約問合せに加えて，ある広
い一定の時間の window 幅で集約をする問合せも行うことを考える．すなわち，図 10 のように考える．図 10 の場合は，1000
タプルの window 幅が 1 時間の window 幅よりも大きいときであり，この場合は時間の window 幅による集約はその下流に置かれる．例えば，商品名と住所で集約する問合せでは以下のよ
うな集約も行う．
Select 商品名, ジャンル, 住所, Sum(売上)

図 12

0.4

（商品名, 住所）⑥0.3

（ジャンル, 住所）

0.8

⑧

⑦

0.7

（住所）

0

⑤

(none)

②

0.9

（商品名）

0.2

（ジャンル）

エンジンに登録される順序と参照頻度の関係

た，本稿で提案したコストモデルにより，どのように処理コストが効率化されているかをシミュレーションし確認した．今後
は，時間次元に対する集約を加えることにより，処理コストにどのような影響が出るかを再検討する．また，上述した手法を実際のストリーム処理エンジンに組込み，実験を行い，本手法の有用性を実際に評価する．

8. 謝

辞

本研究の一部は，(株)KDDI 研究所と筑波大学の共同研究による．(CPE25113)

From 売上情報 [Range 1 hour]Group By 商品名, 住所
これにより，例えば商品名と住所の集約問合せにおいて，直近

30 分の集約が要求された場合においても，上記の問合せから導出できる．

6. 予備的評価
本章では，5. 5 節における具体例において，本稿のアルゴリ
ズムによって，どのように処理が効率化されているかを調べる．
そのために，連続的問合せとして選択される個数を 1 から 9 の間で変えたときの処理コストの和 (実線)，および，登録した連
続的問合せの集約結果のタプル数 (空間コスト) の積算値 (点線)を求める．この結果は，図 11 のようになる．ここで，x 軸は集約問合せをストリーム処理エンジンに連続的問合せとして登録する個数，y 軸はタプル数である．図 11 から，5. 5 節における例の場合，ストリーム処理エンジンに連続的問合せとして 5 つ
登録したときに最もコストが小さくなることがわかる．
また，図 12 は，それぞれの集約問合せがエンジンに登録される順番と、それぞれの集約問合せの平均参照頻度を表している．
エンジンに登録される順番は丸付き数字で表しており，図 12から，平均参照頻度が 0.9 である（商品名）が二番目に登録されるものとして選ばれ，平均参照頻度が 0.8 である（住所）が三番目に登録されるものとして選ばれることから，概ね参照頻度が高い集約問合せからエンジンに登録されている．

7. まとめと今後の課題
本稿では，ストリーム処理エンジンを用いた OLAP 処理を実現の手法と，処理の効率化のためのアプローチを示した．ま文

献[1] J. Han et al. “Stream Cube: An Architecture for MultiDimensional Analysis of Data Streams”, Distributed andParallel Databases, 18(2):173-197, Sep. 20 2005.
[2] J. Han et al. “Data Mining: Concepts and Techniques”,
Morgan Kaufmann, Third Edition, Jul. 2011.
[3] TPC-D, http://www.tpc.org/tpcd/default.asp.
[4] A. Arasu et al. “STREAM: The Stanford Data Stream Management System”, Technical Report, Department of Computer Science, Stanford University,
http://ilpubs.stanford.edu:8090/641/, 2004.
[5] Storm, http://storm-project.net/.
[6] L. Neumeyer et al. “S4: Distributed Stream Computing
Platform”, In Data Mining Workshops (ICDMW), 2010
IEEE conference on, pages 170-177, 13 Dec. 2010.
[7] D. J. Abadi et al. “The Design of the Borealis Stream Processing Engine”, In Second Biennial Conference on Innovative Data Systems Research (CIDR 2005), Asilomar, CA,
Jan. 2005.
[8] J. Han et al. “Efficient Computation of Iceberg Cubes with
Complex Measures”, In Proc. 2001 ACM-SIGMOD Int.
Conf. Management of Data (SIGMOD ’01), pages 1-12,
Santa Barbara, CA, May 2001.
[9] V. Harinarayan et al. “Implementing Data Cubes Efficiently”, In Proc. 1996 ACM-SIGMOD Int. Conf. Management of Data (SIGMOD ’96), pages 205-216, ACM, Montreal, Canada, June 1996.
[10] A. Arasu et al. “The CQL continuous query language: semantic foundations and query execution”, The VLDB Journal, 15(2):121-142, 2006.