DEIM Forum 2017 A1-4

ストリーム処理とバッチ処理の統合と実行最適化
長

裕敏†

塩川

浩昭††

北川 博之††† 筑波大学院システム情報工学研究科 〒 305–8573 茨城県つくば市天王台 1-1-1†† 筑波大学計算科学研究センター 〒 305–8573 茨城県つくば市天王台 1-1-1
E-mail: †denam96@kde.cs.tsukuba.ac.jp, ††{shiokawa,kitagawa}@cs.tsukuba.ac.jp
あらまし センサーデバイスやソーシャルメディアの発展により, 様々な分野で大規模データの利活用に注目が集まっている. 大規模データ処理のアプローチとして, データを蓄えて一括で処理するバッチ処理方式とデータを継続的に処理するストリームデータ処理方式がある. 近年ではデータ利活用の高度化に伴い両処理方式を合わせて使用する機会は多いが, 各処理方式ではデータモデルが異なるため, データを処理する利用者の学習・実装コストは大きく, 適切な処理方式を選択するのは困難である. そこで本研究では, 両処理をデータフローで統一的に記述できる処理記述法と処理を適切な処理方式で自動実行する処理基盤を提供することで, 容易で効率的なデータ処理を実現する.
キーワード

ストリーム処理，バッチ処理，統合
利用形態が複雑化してきている．しかしながら，各処理方式は
基本的にデータモデルが異なるため，実際の運用の場面においては適切な処理方式の割り当ての問題と学習・実装コストの問題が存在する．
複数の処理方式を組み合わせる例として，Twitter から最近のニュースに関連があるツイートデータの取得を行う場合を考える．図 1 に処理内容を示す．この例では処理内容を 3 段階に分けて考える．処理 1 では，ニュースストリームからニュースに出現した単語群を収集する．ニュースストリームに含まれるテ

図 1 例：最近のニュースに関連するツイートの取得

キストを単語分割し，ニュースに出現した単語群を二次記憶領域に格納する．処理 2 では，ニュースに頻繁に出現した単語群

1. 序

論

を摘出する．処理 1 で保存したニュースの単語群を読み込んだ
後に，各単語に対して集計処理を行い，単語の出現数をカウン

近年ではセンサーデバイスやソーシャルメディアの発展によトする．その後，出現頻度が高いものをいくつか抽出して二次り，人々が扱えるデータは爆発的に増大している．これに伴い，記憶領域に格納する．最後に処理 3 において最新のニュースト

様々な分野で大量データの利活用に注目が集まっている．例え

ピックに対応するツイートデータを抽出する．ここでは，処理ば，ソーシャルメディアの情報を利用した製品・サービスのポ2 にて得られたニュースの頻出単語とツイートストリームの 2

ジネガ分析やセンサデータを活用したシステムの異常検知など

つを入力データとして読み込む．その後，2 つのデータを照会が典型的な活用例として挙げられる．

し，ニュースの頻出単語が含まれているツイートデータを抽出

巨大なデータを処理するために一般的に用いられるアプロー
チとして, データを蓄えて一括処理するバッチ処理方式と，デー
タを蓄えずに常時処理するストリーム処理方式がある．バッチし結果として出力する．
この例では先に挙げた通り，データ処理の利用者にとって 2
つの課題が存在する．

処理方式ではデータを一括処理するため，レイテンシは大きく

課題 1：学習・実装コストが大きい

かかってしまうがメモリ効率がよくスループットも高い．既存

処理 1 にストリーム処理方式，処理 2 にバッチ処理方式を適の処理基盤として Hadoop [1] や Spark [2] などが開発されてい用すると仮定する．この場合，利用者はそれぞれの処理方式のる．一方で，ストリーム処理方式ではデータを常時処理するた処理基盤に従ったプログラムを用意する必要がある．しかしなめ，バッチ処理方式と比較してメモリ効率が低下するものの，がら，ストリーム処理方式とバッチ処理方式では処理モデルがレイテンシを抑えて高速に処理を実行できる．既存の処理基盤

多くの場合に異なるため，利用者に高い学習コストと実装コスには，Storm [3]，Spark Streaming [4] が開発されている．

トを必要とする．したがって，処理のパフォーマンスを最適化

データ処理を行う利用者は，処理内容に応じて様々なデータ処理基盤から適当な処理方式を選択して処理を実行する必要がするためには各処理方式の実行モデルを理解した上でそれぞれに適した構築・管理する必要がある．

ある．特に近年ではデータ利用の高度化に伴い，両処理方式を

課題 2：適切な処理方式の選択が困難

組み合わせて処理要求を満たす機会も多く，データ処理基盤の効率的に処理を行うにはそれぞれの処理段階において適切な処理方式を選択する必要がある．処理 1 ではストリームデー

は MapReduce のデータフローを抽象化したものである．map

タ，処理 2 では蓄積データのみを対象とするので，それぞれにや filter，merge，join 等のデータに対する変換処理のメストリーム処理方式とバッチ処理方式を適用するのが合理的でソッドが多数定義されており，これらを組み合わせて入力デーある．しかし，処理 2 の処理の実行頻度が週に一度といった少タに対する処理のデータフローを記述する．Platform API は処ないケースの場合は，処理 1 の処理を常時実行するのは非効率理を実行する処理基盤である Hadoop や Storm のライブラリのであり，バッチ処理方式を用いた方がメモリを効率的に利用で両方を抽象化したものであり，Producer API を用いて記述した

きる可能性がある．また，処理 3 ではストリームデータと蓄積

データフローを指定された処理方式で実行する．そのため，利データの両方を入力とするため，どちらの処理方式も利用する

用者は Producer API に従った処理記述を一度用意することで処ことができる．そのため，処理を効率良く行うために試行錯誤

理を各処理方式で実行することができる．

を重ねて適当な処理方式の割り当てを見つける必要があり，利用者にとって大きな負担となる．

CloudDataflow [6] は，Google Cloud Platform で提供される大規模データ分析処理基盤とそのマネージドサービスである．

これらの課題を解決するために，近年では Suminngbird [7] やCloudDataflow ではバッチ処理とストリーミング処理を統合す

Cloud Dataflow [6] といった両処理方式を統合した処理基盤がいるプログラミングモデルを提供している．Cloud Dataflow の

くつか開発されている．しかしながら，これらの処理基盤は課

プログラミングモデルの中核となるコンセプトに Pipeline と

題 1 で挙げた学習・実装コストの問題のみに焦点を当てており，

PCollection がある．Pipeline は外部ソースから入力データを使用する処理方式については利用者自身が事前に選択する必要

受け取り，そのデータを変換し，出力データを生成するといがある．そのため，課題 2 に示した適切な処理方式の割り当て

う，一連の処理のデータフローを表す．Pipeline 中のデータはの問題については依然として解決されていない．

PCollection という，レコードを際限なく持てるデータ集合で表

そこで本稿では，簡潔なプログラミングモデルを使用してスされる．PCollection 内の各レコードにはタイムスタンプが関連

トリーム処理方式とバッチ処理方式を統合的に利用可能な新し付けられており，ストリームデータを扱う際に Pipeline が提供

い処理基盤 JsFlow を提案する．JsFlow では以下の 3 つの機能するウィンドウ処理を用いてタイムスタンプに従って有限サイを提供することで，大規模データ処理の容易実行を可能にする．

ズのデータ集合に変換すること可能としている．これにより，(1) 半構造データ形式として広く普及している JSON [5] を用い有限の蓄積データとストリームデータの両方に対して単一のて，ストリームデータと蓄積データを JSON レコード集合としAPI である Pipeline を用いて処理を記述できるようにしている．

て抽象化する．(2)Hadoop [1] の JSON データに対する照会言語

これらの処理基盤はストリーム処理方式とバッチ処理方式がである Jaql [8] を拡張し，JSON レコード集合に対する処理の統一されたプログラミングモデルを提供しているため，序論でデータフローを記述することで，ストリーム処理方式とバッチ挙げた学習・実装コストに対する課題の解決に大きく貢献して処理方式を統一的に扱えるプ処理記述法を提供する．(3) 提案いると考えられる．しかしながら，これらの処理基盤は記述しする処理記述法に基づいた処理のデータフローを各処理方式でた処理のデータフローをストリーム処理方式とバッチ処理方式

実行した場合の処理時間とメモリ使用量を見積り，これらの情のどちらで実行するかについては利用者が全て選択する必要が報から適切な処理方式を自動的に選択する．

あるため，適切な処理方式の割り当てに関する課題点は解決で本稿の構成は以下の通りである．まず 2 章で本研究の関連きていない．

研究について述べ，3 章にて予備知識として本稿で提案する

本稿で提案する JsFlow では，両処理方式を統一的に扱えるプ

JsFlow が利用している JSON [5] や Jaql [8]，Flink [9] について

ログラミングモデルを提供するとともに，処理のデータフロー

述べる．4 章にて本研究で提案する JsFlow について述べる．5

から処理時間とメモリ使用量を概算し，処理方式を自動的に選章では評価実験について述べる．最後に 6 章で本稿のまとめと択して実行するため，大規模データ処理を容易に実現すること

今後の課題について述べる．

を可能とする．

2. 関 連 研 究

3. 予 備 知 識

バッチ処理基盤とストリーム処理基盤を統合的に扱うための処

本章では提案する統合処理基盤 JsFlow で用いるデータ構造

理基盤がこれまでにいくつか研究開発されている [4], [6], [7], [9]．

である JSON [5], 提案する処理記述法のベースとなる Jaql [8],本稿では，その中でも代表的な処理基盤である Summingbird [7]JsFlow が用いる分散フレームワークの Flink [9] について述べる．

と CloudDataflow [6] について説明する．

3. 1 JSON

Summingbird [7] は，MapReduce [10] 処理をストリーム処理

JSON [5] は軽量の半構造データフォーマットである．JSON方式，及びバッチ処理方式で実行できる処理基盤である．Sum-

では，データ全体を配列，またはキーと値のペア (フィールド)mingbird では，自身が提供する API を利用した Scala 及び Javaを列挙したオブジェクトとして記述する．値として利用できる

言語で書かれたプログラムを MapReduce プログラミングモデ

データ型には数値型，文字列型，ブール型，null，配列，オブルに基づくバッチ処理基盤 (Hadoop) 及びストリーム処理基盤

ジェクトがある．配列は全体を角括弧 ([]) で囲み，値をカンマ(Storm) のジョブに変換して実行できる．SummingBird の中核と区切りで列挙していく．オブジェクトは全体を中括弧 ({}) で囲なる API に Producer API と Platform API がある．Producer API

み，キーと値をコロン（:）で区切ったペアをカンマ区切りで列図 2 Jaql プログラミングモデル

図 4 JsFlow 全体図

Dataset API が提供されており，これらを同一の基盤上で動作させることができる．Dataset API を用いて登録した処理はバッ
チ処理となるため，データフローに含まれる全ての演算子に対して処理が完了した場合に実行を終了する．これに対し，

DataStream API を用いて登録した処理はストリーム処理となる
図 3 Jaql 処理記述例

ため，データフローに含まれる全ての演算子は常に駆動し続け，継続的に処理を実行する．

挙していく．値として配列やオブジェクトを取ることができる

各 API では，各レコードに対して同一処理を行い一つあるいため，配列やオブジェクトを入れ子にして持つこともできる．

は複数のレコードを返す map，flatMap 演算や，複数レコー

そのため表形式に限らず，複雑な形のデータを保持できる．

ドに対して集約を行う max や min，sum といった集約演算な

3. 2 Jaql

どの演算子が共通して用意されている．入力データに対してそ

Jaql [8] は Hadoop 上の JSON 形式の蓄積データを処理するたれらの演算を適用してデータ操作を行い処理のデータフロー

めの関数型照会言語である．Jaql は処理をデータフロー形式でを構築する．また，DataStream API では際限のないストリーム

記述することで，JSON データに対してフィルタや結合，集約

データに対して集約演算や結合演算のような複数レコードを入などといった基本的な演算をシンプルで短いコード量で記述す力とする演算を適用するためにウィンドウ処理を提供している．

ることが可能である．Jaql で記述した処理は，システム側が処理内容を Hadoop の MapReduce ジョブに変換して実行するため，利用者は容易に並列処理を行うことができる．

4. 統合処理基盤：JsFlow
4. 1 概要Jaql のプログラミングモデルを図 2 に示す．Jaql ではまず入提案する処理基盤である JsFlow は，序論で述べた学習・実力データを Source から読み取る．Source はファイルといった装コストの削減と適切な処理方式の割り当ての解決をするためデータを読み取る処理を指す．次にデータは照会で指定されたに，以下の 2 つのアプローチを採用する．(1) JSON データに対演算子 (Operator) により操作する．照会が提供している演算にする処理の変換フローを記述することでストリーム処理方式とは filter によるフィルタリングや group による集約，join

バッチ処理方式を統合的に扱える処理記述法の導入．(2) 処理による結合など一通りの基本的な演算子が提供されているため，をバッチ処理基盤およびストリーム処理基盤で実行した際の処容易にデータ操作を行うことができる．最後にデータは Sink に

理時間とメモリ使用量を推定し，使用可能なメモリ量に基づい出力される．Sink はデータをディスクへ書き込む処理を指す．

た適切な処理方式の自動選択アルゴリズムの導入．

図 3 に Jaql の処理記述例を示す．ここでは入力データ中の本研究では 2 つのアプローチに加え，処理を実行する処理基

JSON オブジェクトの内，income が 25,000 以上の ID とユー盤として Apache Flink [9] を組み合わせた JsFlow のプロトタイ

ザ名を取得する処理記述となっている．例では，最初に Source

プシステムを構築した．図 4 に構築した JsFlow プロトタイプのにあたる read 演算によって HDFS に格納されている users

全体像を示す． 図 4 中の処理記述解析系においてユーザが記述

ファイルから JSON データを読込んでいる．次に Operetor に該

した JSON データの処理フローに使用されている演算情報やパ

当する filter 演算を用いて income が 25,000 以上のデー

ラメータを抽出し，続く処理方式選択機構にて抽出した情報をタに絞り込み，続く transform 演算で絞り込んだデータの用いて登録された処理フローの処理方式の自動選択を行う．そid と name を摘出している．最後に Sink にあたる write 演の後，選択された処理方式に合わせて処理フローを Flink が提算によって結果を HDFS 上に result というファイル名で保供する各処理方式の API を用いたプログラムコードを生成し，存している．

Flink 上に展開して動作させる．

3. 3 Flink

4. 2 バッチとストリーム処理を統合的に扱う処理記述法

Flink [9] はオープンソースの分散処理基盤である．ストリー

JsFlow が提供する処理記述法は Jaql を拡張することで実装

ムデータに対しては DataStream API，蓄積データに対してはする．3. 2 節で述べた通り，Jaql は JSON 形式で保存された蓄積データを処理するための照会言語であり，蓄積された JSON
レコードに対する変換処理を filter や join， group by，transform などの提供される演算子群を適用して処理の内容
を DAG として表現することで処理を記述する．

JsFlow では両処理方式を統合的に扱うために，蓄積された
JSON レコードだけでなくストリームデータに対しても処理を適用可能にする必要がある．本研究ではストリームデータを有限のレコード集合としてまとめるために Jaql 照会言語の Sourceにウィンドウ処理の演算を追加する．際限のないストリーム

図5

処理方式選択の流れ

データを一定時間，あるいは一定数の JSON レコード集合に区切ることで Source 以降の変換処理の演算子群を蓄積データに

4. 3 処理方式の自動選択手法

対する処理と同様に適用することができる．また，演算子群をJsFlow では 4. 2 節で述べた処理記述法を用いて記述した処理

適用してデータ変換したのちに再度データをストリームとしのデータフロー (Task) を受け取りストリーム処理方式，またはて配信可能とするために，tostream 演算を追加した．また，バッチ処理方式を自動選択し処理を実行する．まず，4. 3. 1 節

永続的に処理結果を保存するためにファイルに追記保存するにて各処理方式で実行した場合の違いについて述べ，4. 3. 2 節

append 演算を Sink の演算として追加した．

にて登録された Task の処理方式の選択手法について述べる．

また，ストリームデータに対するウィンドウ処理おいては

4. 3. 1 各処理方式の特徴

JSON レコード集合を生成するタイミングを設定する必要があ

バッチ処理方式では，各演算は起動した後に入力されるデー

り，蓄積データに対しては処理を定期的に行うかアドホックに

タ集合に対して処理を完了した後に実行を停止する． したがっ行うか指定する必要がある．したがって，window 演算では実て，各演算は以前の処理内容を保持していないため全てのレ

行間隔，read 演算では実行タイミングを設定する．

コードに対して再計算する必要がある．一方でストリーム処理
方式では，各演算は入力を待ち続け処理を永続的に継続する．

表 1 JsFlow が提供する演算子群
分類
入力演算

演算子名

処理内容概略

window

ストリームデータの取得と演算範囲の指定

read

二次記憶領域から蓄積データの取得

filter

条件を満たさないデータの除去

transform 射影によるフィールドの整形
操作演算

出力演算

top

入力の k 個以外のレコード除去

group by

グループ化し，集約関数を適用

そのため，入力データ集合に重複が発生する場合に各演算が前回分の演算対象の処理結果を保持し，差分があったデータに対してのみ計算することで計算量を削減して処理を高速化することができる．ただし，各演算子は処理結果の一部をオンメモリで保持しておく必要があるため，メモリ使用量と処理時間にト
レードオフが生じる．

4. 3. 2 コストモデルに基づく処理方式の選択
4. 3. 1 節で述べたように各処理方式には必要とするメモリ使

join

2 つ以上の入力を結合用量と処理時間にトレードオフの関係があるため，処理結果を

udf

ユーザが定義した処理の実行高速に得るためには使用できる計算資源に応じて適切な処理

tostream

結果をストリームデータで出力方式を選択する必要がある．そこで JsFlow では登録された各

write

結果を二次記憶領域に上書き保存

Task を各処理方式で実行した場合にかかるメモリ使用量と処理

append

結果を二次記憶領域に追記保存

時間を推定し，使用できるメモリ量において処理時間が短くなるように各 Task の処理方式を自動的に選択する手法を導入す

表 1 に JsFlow が提供する演算子を示す．演算は大きく別け

る．処理方式の選択は図 5 に示す流れで行う．

て 3 つに分類され，入力演算，操作演算，出力演算からなる．

メモリ使用量と処理時間の推定

それぞれ Jaql の Source，Operetor，Sink に該当する．入力演算

最初に登録された各 Task について各処理方式で実行した場

ではストリームデータあるいは蓄積データから設定した実行間合にかかるメモリ使用量と処理時間を推定する．Task にかか

隔毎，あるいは実行タイミング毎にデータを読み込み，JSONるメモリ使用量と処理時間の推定は Task 中の各演算に必要

レコード集合を生成する．次に操作演算では，直前の演算からなメモリ使用量と処理時間から算出する．バッチ処理方式で送られた JSON レコード集合に対して定義された処理を行い，は上流の演算からレコードを順次処理していくため，Task 中

データを操作する．表 1 に示す通り Jaql 同様に一通りの基本演の演算で最も必要なメモリ使用量が Task に必要なメモリ使算を提供しているので，容易にデータ操作を行うことができる．

用量となる．したがって，Task t におけるバッチ処理方式で最後に出力演算で，操作演算を介して得られた結果を出力する．

のメモリ使用量 BM (t) は， n を Task に使用している演算子

JsFlow では各演算間を->記号を用いて繋ぎ合わせ，入力演算，の数とし，Mi を Task 中の各演算での使用メモリ量とすると，

操作演算，出力演算の一連のデータフローを Task として登録

BM (t) = maxn
i=1 (Mi ) で計算する．一方でストリーム処理方を受け付ける．

式では各演算が常時駆動し，以前の処理結果を保持するため，

Task に必要なメモリ使用量は各演算のメモリ使用量の合計値と表 2 演算子メモリ使用量
演算子名

出力レコードサイズ

出力レコード数

メモリ使用量 M

window (Time)

RS

ストリーム：EI × (IR)バッチ：W S × IR

RS × W S × IR

window (Count)RS

ストリーム：EI
バッチ：W S

RS × W S

read

RS

ストリーム：N × (1 − OR)
バッチ：N

RS × Nfiltertransform
top
group
join

RS
RS × T R
RS
RS × T R(RSlef t + RSright ) ×
TR

N ×σ
N
K
N × AR(Nlef t + Nright ) × σ

udf
tostream
write
appendRS × T R
RS
RS
RS

N ×σ
N
N
N

表3

計算時間 C

RS × N × T R
RS × K
RS × N + RS × N × T R × AR
(RSlef t +Nlef t )+(RSright +Nright ) + (RSlef t × Nlef t +RSright + Nright ) × T R × σ
RS × N × T R × σ
ストリーム：RS × N
ストリーム：RS × N
ストリーム：RS × N

記号一覧

Of ilter (N )Otransf orm (N )Otop (N )DE(RS × N ÷ X)
Ogroup (N )DE(RS × K ÷ X)
Ojoin (Nlef t + DE((RSlef t × Nlef t +Nright )RSright + Nright ) ÷
X)
Oudf (N )Algorithm 1 処理方式選択アルゴリズム

記号

定義

RS

入力 JSON レコード 1 要素のデータサイズ

2: N 個の Task の登録を受け付ける

WS

ウィンドウサイズ

3: for t ∈ [1, 2, ..., |N |] do

EI

実行間隔

4:

N

入力レコード数

5:K

ユーザ指定の数

6:X

出力バッファサイズ

7:σ

データの選択率

8: end forIR

ストリームデータの入力レート

9: curM emU se=0

AR

データの集約率ORTR

データの重複率
データサイズの変換率

通信時間 D

1: 使用可能メモリ量 maxM em を設定する

BM (t) = maxn
i=1 (Mi )∑BP T (t) = n
i=1 (Ci + Di )∑SM (t) = n
i=1 Mi∑SP T (t) = n
i=1 (Ci + Di )10: for t ∈ [1, 2, ..., |N |] do
11:Task t をバッチ処理方式に設定

12:curM emU se = curM emU se + BM (t)13: end for
14: if maxM em <= curM emU se then

なる．したがって，Task t におけるストリーム処理方式でのメ∑
モリ使用量 SM (t) は，SM (t) = n
i=1 Mi で計算する．

Task の各処理方式での処理時間は Task 中の各操作演算での入力レコード全てに対して定義した処理を行う計算時間と，group by 演算のように同一のキーを持ったレコードを同一

15:実行を拒否し，使用可能メモリ量を増加させるか登録する Task
を減らすように警告

16: else
17:
18:

for t ∈ [1, 2, ..., |N |] do
Dif (t) = |BP T (t) − SP T (t)|19:

end for

ノードにまとめるネットワーク転送にかかる通信時間の総計

20:

Sort each t in the decending order of Dif (t)となる．したがって，Ci を Task 中の各操作演算にかかる計

21:

for t ∈ [1, 2, ..., |N |] do

算時間， Di を Task 中の各操作演算にかかる通信時間として，

22:

Task t におけるストリーム処理方式での処理時間 SP T (t) は，∑
SP T (t) = n
i=1 (Ci + Di )，同様にバッチ処理方式での処理時∑
間 BP T (t) は，BP T (t) = n
i=1 (Ci + Di ) で計算する．

23:

表 2 に各演算子にかかるメモリ使用量と計算時間を示し，表 2
中の各記号は表 3 に示す

Task の割り当て手法

if maxM em > curM emU se∧

Dif (t) > 0 thenif maxM em > curM emU se + SM (t) − BM (t) thenTask t をストリーム処理方式に設定

24:

curM emU se = curM emU se + SM (t) − BM (t)25:

end if26:
27:

end if

28:

end for

29: end if

図 5 の処理のアルゴリズムを Algorithm 1 に示す．最初に

JsFlow が使用可能なメモリ量を設定し，複数の Task の登録を受け付ける．Algorithm 1 の 3 行目から 8 行目が処理 1 に相当する．Task を各処理方式で実行した場合のメモリ使用量と処理時間を前述した方法で推定する．Algorithm 1 の 10 行目から∑< n
12 行目が処理 2 に相当する．maxn
i=1 (Mi ) =i=1 Mi なので，全ての Task をメモリ使用量が低いバッチ処理方式に設定する．

Algorithm 1 の 14 行目から 16 行目が check 処理に相当する．全ての Task のメモリ使用量の合計が使用可能メモリ量を上回っていた場合は実行を拒否し，使用可能メモリ量を増加させるか登録する Task を減らすように警告する．Algorithm 1 の 17 行
目から 28 行目が処理 3 に相当する．|SP T (t) − BP T (t)| から
処理時間の減少量を求め，使用可能メモリ量を超えるまで処理時間の減少量が大きい Task から処理方式を変更していくこと
で全ての処理方式を確定する．

表 5 Flink 設定

Flink Version
使用可能メモリ量
使用 CPU 数
並列度

図6

実験に用いる Task

表 4 各マシン性能

CPU
コア数
メモリ
HDD
OS

Xeon E5-2650 v3 2.3GHz
10
16.0 GB
240 GB
Ubuntu14.04

1.1.2
2.0 GB8
4

1 t = window("Source2", range 100 seconds ,10 seconds)
2 -> group by $.key into {$.id, $.key, sum: sum($.
num), $.text};
3 register task JsFlowSample {4window("Source1", range 100 seconds ,10seconds )5 -> filter $.num > 10
6 -> join t where t.key == $.key into {$.id, $.key,
t.sum, $.text}
7 -> tostream;
8 }Listing 2

5. 評 価 実 験
5. 1 実 験 概 要
4. 2 節で述べた，JsFlow が提供する処理記述法を用いること
で学習・実装コストを削減できているか JsFlow と Flink のコー
ドサイズを比較して評価を行う．また，4. 3. 2 節で述べた処理方式の自動選択手法を用いて，適切に処理方式を選択しているか処理にかかるレイテンシの観点から評価を行う．本実験では実行環境に Flink [9] を用いた分散処理環境を構築し，4. 2 節で述べた処理記述法を用いて複数の Task を登録する．本実験で用いた Task は図 6 のように 2 つの情報源と対応した入力演算子，filter 演算，group by 演算，join 演算，tostream
演算からなる DAG とした．各情報源にはストリームデータか分散配置されたファイルが割り当てられる．レコードサイズは80 Byte，ストリームデータの入力レートは 10 KB records/s とした．各操作演算でのデータサイズの変更はなく，filter 演
算，group by 演算，join 演算の選択率及び集約率をそれぞれ 0.9，0.01，0.01 とした．また，本実験では異なる Task において同様の処理を行う演算子の共有は考慮しないものとする．
処理方式の自動選択手法の有効性を評価するために，登録した各 Task を提案処理手法で割り当てを行った場合と，各 Task
を一律でストリーム処理方式．または，一律でバッチ処理方式
で行った場合での Task の実行タイミングから処理結果を返す
までのレイテンシの差で評価する． 実験ではそれぞれの処理方式に適した入力データ集合に重複がある設定が同一の Task を複数登録した場合と入力データ集合に重複がない設定が同一の

Task を複数登録した場合，及び Task ごとに設定が異なる複数
Task を登録した場合の 3 通り行う．
実験環境については 2 台のマシンで分散環境を構築した．各
マシンの性能と Flink の設定を表 4 と表 5 に示す．
1 {"id": Long, "key": Integer, "num": Integer, "text":
String,}1 public class DataStreamSample{
2 public static void main(String[] args) throws
Exception {3
StreamExecutionEnvironment env =StreamExecutionEnvironment.
getExecutionEnvironment();
4
env.setStreamTimeCharacteristic(
TimeCharacteristic.EventTime);
5
DataStream<Tuple4<Long, Integer, Integer,
String>> left = new SampleData.Source().
getSource(env, 10000).
assignTimestampsAndWatermarks(newTimeLagWatermarkGenerator()).filter(newmyFilter());
6
DataStream<Tuple4<Long, Integer, Integer,
String>> right = new SampleData.Source().
getSource(env, 10000).
assignTimestampsAndWatermarks(newTimeLagWatermarkGenerator()).keyBy(1).
window(SlidingEventTimeWindows.of(Time.
seconds(100), Time.seconds(10))).sum(2);7
DataStream<Tuple4<Long, Integer, Integer,
String>> mid = left.join(right)
8.where(new KeySelector<Tuple4<Long, Integer,
Integer, String>, Integer>() {9public Integer getKey(Tuple4<Long, Integer,
Integer, String> t) { return t.f1; }})10.equalTo(new KeySelector<Tuple4<Long, Integer,
Integer, String>, Integer>() {11
public Integer getKey(Tuple4<Long, Integer,
Integer, String> t) { return t.f1; }})12.window(SlidingEventTimeWindows.of(Time.
seconds(100), Time.seconds(10))).apply(newmyJoin());
13
mid.print();
14
env.execute();
15 }
16 }

Listing 3
5. 2 プログラムコード量の比較評価のそれぞれを用いて図 6 の Task を実装し，コード量を比較

Listing 1

した．処理対象とするレコード構造は Listing 1 のように設定する．図 6 の Task において各情報源にストリームデータを設定

図 7 実験 1 における各方式のレイテンシ比較

し，各入力演算にウィンドウ長を 100 秒・実行間隔を 10 秒とした window 演算を設定する．

JsFlow の処理記述法を用いた例を Listing 2 に示し，Flink の
DataStream API を用いた例の一部を Listing 3 に示す．

図 8 実験 2 における各方式のレイテンシ比較

以下のレイテンシで実行できていることが分かる．

5. 3. 2 実験 2：入力データ集合に重複がない Task 群の比較
本実験でも 5. 3. 1 節と同様に図 6 の Task において各情報源にストリームデータを設定し，各入力演算に window 演算を設

Listing 2 と Listing 3 でのコード量を比較すると，JsFlow は 8定する．ただじ，Task 中の各 window 演算のウィンドウ長を行で処理を記述できるのに対し，Flink の DataStream API を用

10 分・実行間隔を 10 分と設定して入力データ集合の重複を排いる場合には主要部分だけでも 16 行，全体で 77 行のコード量除している．設定が同一の Task を 1，3，5 個登録し，各 Task

が必要となる．そのため JsFlow を用いることで実装の容易化

を一律ストリーム処理方式で行った場合と一律バッチ処理方式が期待できる．また，Flink では行いたい処理方式に合わせてで行った場合，及び提案処理手法で処理方式の選択を行った場

DataSet API か DataStream API を用いる必要がある．JsFlow で合での Task のレイテンシの平均値を比較する．

は入力データを全て JSON レコード集合として抽象化すること

実験結果を図 8 に示す．図 8 ではウィンドウ長を 10 分・実行

で，入力データや処理方式を意識することなく，処理を JSON

間隔を 10 分に設定した場合での各方式のレイテンシを比較しレコード集合に対する変換のデータフローで記述可能としていている．入力データ集合の重複がないので JsFlow では全てのる．これらの理由から JsFlow は提供する処理記述法によってTask をメモリ使用量が少ないバッチ処理方式で実行を行ってい

学習・実装コストの削減に貢献できていると結論付ける．

る．図 8 の実験結果より，登録する Task が 1 つの場合はレイ

5. 3 実験 1：処理方式の自動選択手法の評価

テンシはほとんど同一であるが，登録する Task が 3 つ以上の5. 3. 1 入力データ集合に重複がある Task 群の比較

場合はストリーム処理方式を行うのに十分なメモリを確保でき

本実験では図 6 の Task において各情報源にストリームデー

ずにバッチ処理方式に比べてレイテンシが 1.5 倍程度大きくかタを設定し，各入力演算に window 演算を設定する．Task 中

かっている．JsFlow ではバッチ処理方式に Task を振り分けての各 window 演算のウィンドウ長を 10 分・実行間隔を 1 分とおり，適切な処理方式を選択できている．

設定して設定が同一の Task を 1，3，5 個登録した．本実験で5. 3. 3 実験 3：入力データ設定が異なる Task 群の比較

はこの状況下における，各 Task を一律ストリーム処理方式で本実験では図 6 の Task の入力演算子を表 6 ように設定した行った場合と一律バッチ処理方式で行った場合，および提案処

計 8 個の Task を登録する．情報源 1 にはストリームデータを理手法で処理方式の選択を行った場合での Task のレイテンシ

設定し，入力演算 1 には全て window 演算を用いている．ウィの平均値を比較する．

ンドウ長を 10 分あるいは 100 秒に設定し，入力演算 1 におけ実験結果を図 7 に示す．x 軸は登録した Task の数，y 軸はレる Task 1〜6 の実行間隔を 1 分あるいは 10 秒とし，Task 7,8 のイテンシを表す．この Task は 1 つの Task を登録するのにスト

実行間隔をその Task のウィンドウ長と同値とした．情報源 2 にリーム処理方式で 1.0 GB，バッチ処理方式で 0.48 GB のメモリ

は各マシンに分散配置したファイルかストリームデータを設定が必要であるため，JsFlow では Task の登録数が 3 以下の時はし，Task 1〜4 の入力演算 2 には read 演算， Task 5〜8 の入力

全てストリーム処理方式で実行し，登録数が 5 個の時は 3 つの演算 2 には window 演算を用いる．Task 1,2 で読み込むファイ

Task をストリーム処理方式，2 つの Task をバッチ処理方式で実ル中のレコード数を 100K，Task 3,4 で読み込むファイル中のレ行している．図 7 の実験結果より，登録する Task が 3 個以下

コード数を 6M で用意し，Task 5,6 でのウィンドウ長を 10 分，の場合はメモリ量が十分にあるため差分計算を用いるストリー

Task 7,8 でのウィンドウ長を 100 秒で設定した．入力演算 2 にム処理方式で行った方がバッチ処理方式よりレイテンシが 2〜

おける Task 1〜4 の実行タイミングは登録時とし，Task 5,6 で3 倍程度抑えられているが，登録する Task が 5 個の時は，ストの実行間隔を 1 分，Task 7,8 での実行間隔を 10 秒で設定した．

リーム処理方式ではメモリ量が足りずにバッチ処理方式よりレ

なお，一方の入力演算が実行条件を満たした場合には，もう一イテンシが 1.2 倍程大きくかかっている．JsFlow ではメモリ量方の入力演算も実行される．

が十分ある場合は全てストリーム処理方式で行い，不足した場

各 Task を一律ストリーム処理方式で行った場合と一律バッ

合はバッチ処理方式を用いてメモリ使用量を使用可能なメモリ

チ処理方式で行った場合，及び提案処理手法で処理方式の選択量に抑えているため，レイテンシが短い処理方式と同等かそれを行った場合での Task のレイテンシの平均値を比較する．

表6
Task 数
Task 1
Task 2
Task 3
Task 4
Task 5
Task 6
Task 7
Task 8

実験 3 の各 Task の入力演算子設定

入力演算子 1

ウィンドウ長 1

実行間隔 1

入力演算子 2

レコード数 2 / ウィンドウ長 2

実行間隔 2 / 実行タイミング

windowwindowwindowwindowwindowwindowwindowwindow

10 minutes
100 seconds
10 minutes
100 seconds
10 minutes
100 seconds
10 minutes
100 seconds

1 minute
10 seconds
1 minute
10 seconds
1 minute
10 seconds
10 minutes
100 seconds

read
read
read
read
windowwindowwindowwindow

100K records
100K records
6M records
6M records
10 minutes
10 minutes
100 seconds
100 seconds

once
once
once
once
1 minute
1 minute
10 seconds
10 seconds

ることは対象としていない．しかし，実際に処理の運用を行う
際には，処理を新規に追加する場合や処理の情報源となるスト
リームデータの入力レートが変わることで，割り当てられた処理方式を変更した方が効率良く処理を行える可能性がある．そのため，処理が追加されたタイミングや処理基盤の起動後からの経過時間に応じて処理方式を再度割り当てることに検討の余
地がある．
図9

実験 3 の Task 群の各方式でのレイテンシ比較

実験結果を図 9 に示す．JsFlow の処理方式の割り当てにおいて全てバッチ処理方式で行った場合には 3.0 GB のメモリを必要とし，全てストリーム処理方式で行う場合は 6.2 GB のメモ

7. 謝

辞

本研究の一部は，科研費・基盤研究 B(26280037)，及び国立研究開発法人情報通信研究機構の委託研究 (APD28151) の助成を受けたものである．

リを必要とする．そのため使用可能メモリ量である 4.0 GB になるまで推定処理時間の減少が大きい Task から処理方式を変更し，Task 1，2，5，6，8 はストリーム処理方式で実行し Task

3，4，7 はバッチ処理方式で実行した．図 9 の実験結果が示す
ように, JsFlow を用いて Task の割り当てを行った場合にレイテンシが一番抑えられている．これは， 全ての Task をバッチ処
理方式で行った場合は差分計算を用いないために入力データの重複部分の再計算によってストリーム処理方式より大きく遅れており，全ての Task をストリーム処理方式で行った場合は十分なメモリ量が確保できていないため，処理の途中でディスク

I/O による遅延が発生しているからだと考えられる．以上より，
JsFlow による処理の割り当ては有効であると考えられる．

6. ま と め
本研究では，ストリーム処理方式とバッチ処理方式を統合的に実行可能な処理基盤 JsFlow の開発を行った．本稿では，JSONを用いてストリームデータと蓄積データを JSON レコード集
合として抽象化し，処理を JSON レコード集合に対する変換のデータフローで記述可能にすることで，両処理方式を統一的に扱えるプログラミングモデルを提供した．また，各処理方式で実行するのに必要なメモリ使用量と計算量を推定し，使用可能なメモリ量に応じて処理時間が短くなるように処理記述の処理
方式を自動的に選択するアルゴリズムを提案した．これにより，大規模データ処理処理を簡潔なプログラミングモデルを用いて容易に実行することを可能にした．評価実験により，学習・実装コストの課題の削減と処理の振り分けの有効性を示した．
今後の課題として処理方式の動的変更が考えられる．本研究で提供する処理基盤が行っている処理方式の割り当ては登録時に一度だけ行い，一度割り当てられた処理方式を動的に変更す

文

献[1] K. Shvachko, H. Kuang, S. Radia, R Chansler. The Hadoop Distributed File System. In MSST, pages 1-10, 2010.
[2] M. Zaharia, M. Chowdhury, M. J. Franklin, S. Shenker, and I. Stoica. Spark: cluster computing with working sets. In HotCloud, pages
10-10, 2010.
[3] A. Toshniwal, S. Taneja, A. Shukla, K. Ramasamy, J. M. Patel, S.
Kulkarni, J. Jackson, K. Gade, M. Fu, J. Donham, N. Bhagat, S. Mittal, and D. Ryaboy. Storm at Twitter. In SIGMOD, 2014.
[4] M. Zaharia, T. Das, H. Li, S. Shenker, and I. Stoica. Discretizedstreams: an efficient and fault-tolerant model for stream processing
on large clusters. In USENIX, pages 10-10. 2012.
[5] F. Pezoa, J. L. Reutter, F. Suarez, M. Ugarte, and D. Vrgoc. Foundations of JSON Schema. In WWW, pages 263-273, 2016.
[6] T. Akidau, R. Bradshaw, C. Chambers, S. Chernyak, R. J. FernandezMoctezuma, R. Lax, S. McVeety, D. Mills, ´ F. Perry, E. Schmidt, etal. The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order
Data Processing. In PVLDB, Volume 8, No.12, 2015.
[7] Oscar Boykin, Sam Ritchie, Ian O Connell, and Jimmy Lin. 2014.
Summingbird: A Framework for Integrating Batch and Online
MapReduce Computations. In VLDB, Volume 7, No.13, 2014.
[8] K.S. Beyer, V. Ercegovac, R. Gemulla, A. Balmin, M. Eltabakh, C.C. Kanne, F. Ozcan, E.J. Shekita. Jaql: A Scripting Language forLarge Scale Semistructured Data Analysis. In VLDB, pages. 12721283, 2011.
[9] P. Carbone, A. Katsifodimos, S. Ewen, V. Markl, S. Haridi, and K.
Tzoumas, Apache flink: Stream and batch processing in a single engine, In Data Engineering, pages 28-38, 2015.
[10] K.-H. Lee, Y.-J. Lee, H. Choi, Y. D. Chung, and B. Moon. Parallel data processing with mapreduce: A survey. In SIGMOD, pages
11-20, 2011.
[11] T. Akidau, R. Bradshaw, C. Chambers, S. Chernyak, R. J. FernandezMoctezuma, R. Lax, S. McVeety, D. Mills, F. Perry, E. Schmidt, et al.
The dataflow model: a practical approach to balancing correctness,latency, and cost in massive-scale, unbounded, out-of-order data processing. In PVLDB, pages 1792-1803, 2015.