DEIM Forum 2019 D4-4

データストリーム管理システムに関する再考
杉浦健人†

石川 佳治†† 名古屋大学大学院情報学研究科 〒 464–8603 愛知県名古屋市千種区不老町
E-mail: sugiura@db.is.i.nagoya-u.ac.jp, ishikawa@i.nagoya-u.ac.jp
あらまし

リアルタイムに生成されるデータを活用したいという要求は多く，近年は高スループットな処理能力を持つストリーム処理システムが商用及びオープンソースソフトウェアで活発に開発されている．一方，そうしたストリーム処理システムは一貫性などの ACID 特性を保証をしていないものが多く，それらの保証はアプリケーション開発者の実装に委ねられる．しかし，データベース分野でトランザクション処理が発展し続けているように，ACID 特性を満たすデータ処理への要求は根強い．そこで，本稿ではデータストリーム管理システムに再度着目し，データモデルやトランザクション処理について調査，検討する．特に，イベントタイムに基づく処理順を考慮した処理について，既存研究のアプローチを基に議論する．
キーワード データストリーム管理システム，データストリーム処理，トランザクション処理しないことを意味する．したがって，トランザクションをどの

1 はじめにデータの大規模化に伴い，大量のデータをリアルタイムに処理するためのストリーム処理システムが商用ソフトウェア・OSS（open source software）を問わず活発に開発されている．イン
ターネットを始めとする情報技術の発展によりビッグデータ時代とも呼ばれる時代が訪れ，データの大量生成と大量消費が行われている．2000 年代には MapReduce [9] に基づく Hadoop [3]や Spark [4] など，主に静的な分散処理システムによってそうしたビッグデータを処理していた．しかし，それらのシステムはバッチ処理に基づいており，日々加速する意思決定のリアルタイムな支援には不足があった．そうしたリアルタイムなデータ処理の要求に応えるために，最近 10 年では Flink [2] や SparkStreaming [6]，Samza [5] を始めとする分散ストリーム処理システムが活発に研究，開発されており，そのユーザ数を増やしている．
一方，ストリーム処理の多くは連続的 OLAP（online analysisprocessing）のみを対象としており，ACID 特性 [13] の保証は暗黙的に行われている．通常，古典的なストリーム処理は以下の3 つの仮定を置いている．
（ 1 ） 全てのストリームは append-only である．つまり，あるタプルに対する更新をタイムスタンプにより区別（元のキーとタイムスタンプの組を主キーとみなす）し，更新履歴を全て保持したテーブルをストリームとして扱う．
（ 2 ） 各処理はタイムスタンプの情報を用いることで，自身の処理で必要なデータを読める．
（ 3 ） 全ての連続的問合せ処理のデータフローは非循環である．つまり，問合せ処理の流れを DAG（directed acyclic graph）で表せる．
このような前提を置く場合，ACID 特性のうち isolation は暗黙的に満たされる．全てのストリームが append-only であるということは，あるデータ x に対する書き込み w(x) は一度しか発生ように同時実行しても書き込みに対して競合は発生せず，あるトランザクションの処理に必要なデータがすでに書き込まれていることさえ保証すればよい．その他，ある DAG で表される連続的問合せを一つのトランザクションとみなす場合，atomicity
及び durability の保証は exactly-once の保証とおおよそ等しく，ストリーム処理システムの多くはこれらの要素から ACID 特性の保証を暗黙的に行っているとみなせる．
しかし，近年ではより複雑なストリーム処理を記述したいという要求から，stateful な処理，つまり処理の内部状態を考慮した処理を実現しようとする動きがある．これはプログラム中で更新される変数の値をデータベースに格納し，時間的に後の処理もしくは DAG 上で後段に存在する処理でデータベースに格納した値を参照することだと言い換えられる．つまり，上述した「3) 全てのデータフローが非循環である」という仮定が成り立たないことを意味する．
データフローが循環を持つ場合，トランザクションの処理順，言い換えればトランザクションの linearizability の保証が重要となる．簡単な例として，図のような 3 つのトランザクションからなるストリーム処理を考える．なおこれ以降，トランザクション Tx が Ty の後に実行されることを Tx ≺ Ty で表す．図において，トランザクション Ta 及び Tb とこれらの書き出した値を読む Tc の間には順序関係 Ta ≺ Tc 及び Tb ≺ Tc が存在する．
一方で，トランザクション Tb 及び Tc は内部状態をデータベース中のテーブルを用いて共有しており，これらの間にも順序関係が存在する．つまり，トランザクション Tb が読む値は，何らかのタイミングで Tc によってテーブルに書き込まれた値であり，その値がいつ書き込まれたかによってこれらの順序関係が Tb ≺ Tc なのか Tc ≺ Tb なのか定まる．このように連続的問合せとして事前にトランザクションが登録してある場合，各トランザクション間が満たすべき read-from の関係から，一部のトランザクション間に事前に順序関係が定まる．

クション間の順序関係については後述する Meehan らの研究の𝑇𝑎

Stream 1

ものがより汎化した定義であるため省略する．
𝑇𝑐𝑇𝑏

Stream 3

Stream 2

Meehan らの研究 [11] では in-memory DB である H-Store [10]上でストリーム処理を実装し，トランザクション処理を考慮したデータストリーム処理システムである S-Store を提案した．

S-Store では，ストリーム処理の要件として，1 章で仮定としてTable

述べた DAG で表されるようなデータフローに基づく処理を挙げている．つまり，トランザクションの順序付けられた実行で

図 1 内部状態を考慮したストリーム処理のデータフローある．これを表すために，まず S-Store ではあるトランザクショ
ン Ti に入力されるストリーム Si が以下のようにマイクロバッ

そこで本研究では，ストリーム処理におけるトランザクション処理を実現するために，処理の linearizability の保証方法について議論，検討する．つまり，トランザクション処理におけチ [14] に分割されることを前提とした．

Si = ⟨si .b1, si .b2, ..., si .br , ...⟩(1)

る ACID 特性の consistency として， linearizability を考慮した

更に，マイクロバッチ Si = ⟨si .b1, si .b2, ...⟩ は時刻順に入力されデータストリーム管理システム（data stream management system:るものとし，ストリーム処理のために満たすべき要件としてマDSMS）を提案する．ストリーム処理におけるトランザクショイクロバッチの入力順に合わせたトランザクションの実行を定ン処理を提案する既存研究は存在する [7, 11] ものの，ユーザが義した．つまり，あるトランザクション Ti に対するマイクロ

記述する DAG に基づきトランザクションの順序関係を定義しバッチ si .b j の処理を Ti, j で表すとすると，以下の式を満たすており，本来満たされるべき順序関係を適切に記述できていなことを要件とした．

い．それに対し，本研究ではトランザクション間の read-from関係に基づき順序関係を定める．更に，トランザクションの生Ti,1 ≺ Ti,2 ≺ ... ≺ Ti,r ≺ ...

(2)

存時間を考慮することで，同時刻に行われる処理の順序関係と更に，データフローにおけるトランザクションの順序を表すた

異なる時刻に行われる処理の順序関係を分離し，適切にトランめ，データフローの DAG をトポロジカルソートすることで得ザクション間の順序関係を記述する．

られる以下の式も満たすべき要件とした．なお，T1 から Tn が本稿の構成は以下の通りである．まず，2 章では関連する研究とその問題点を紹介する．

2 関連研究同じデータフロー上にあるものとする．

T1,r ≺ T2,r ≺ ... ≺ Tn,r(3)

このような処理を実現するために，S-Store はトリガーによる処データストリーム処理システムにおけるトランザクション処理順の制御を実装した．つまり，データストリームをテーブル理について検討した既存研究はいくつかある [7, 8, 11, 12]．しかで表しそれにトリガーを付与することで，上述の順序関係が守し，Chen らの研究 [8] と Mozafari らの研究 [12] は理論的な面られるようシリアルヒストリーを生成している．

について深く言及していないため，以下では Botan らの研究 [7]と Meehan らの研究 [11] についてのみ紹介する．

しかし，これらの研究では図 1 のような循環のあるデータフ
ローで表される処理の順序関係を適切に表現できない．まず，Botan ら [7] は，データベース管理システム（database man-既存研究ではそもそもデータフローが非循環であることを前提agement system: DBMS）におけるトランザクション処理を踏襲としている．S-Store ではデータベース中のテーブルへのアクしつつ，データストリーム処理のためのトランザクション処理セスも考慮しているが，テーブルがデータフローに与える影響を提案した．Botan らが着目したのは，データストリーム処理は議論しておらず，あるテーブルへの読み書きは一つのトランシステムはデータの入力によって処理が開始する連続的問合せザクション内で完結していると考えられる．S-Store では階層を扱うのに対し，既存の DBMS では基本的に命令（read/write）

的なトランザクションも使用可能であるため，あるテーブルにそのものが入力される one-time 問合せを扱うという点である．

アクセスするトランザクションが複数存在する場合，ユーザにつまり，既存の DBMS にはトランザクション間の順序を決定す

よって一階層上のトランザクションとして一つにまとめられるる仕組みが存在しないことを問題として挙げた．そこで，Botanことが前提となっていると考えられる．つまり，図 1 であればらはまずデータストリームを append-only かつ順序付け可能なTb と Tc を包含したトランザクションの存在が前提となる．

タプルの集合として，連続的問合せを one-time 問合せの系列として考え，DBMS と同様のトランザクション処理が行える

3 トランザクションの順序関係ものとした．更に，上述のような処理の順序を厳密に定めるた

3. 1 データベース管理システムにおける順序関係めに，CSR（conflict serializability）を拡張した CSAO（conflictserializability with arrival ordering）を提案し，データの到着順に合致した処理が行えるシステムを実装した．なお，トランザ

まず，DBMS におけるトランザクション間の順序がどのように決まるかを説明し，DSMS における順序関係を議論するためのアイデアを紹介する．

前提として， DBMS において CSR を考えるとき，全てのトランザクション T 上の順序関係は前順序で表される．つまり，User

SystemEvent time

Ingestion time

同じデータ x ∈ D に対して競合の発生するトランザクション

Ti, Tj ∈ T の間には下記のように両方向に順序関係が成り立つ．
Ti ≺ Tj , T j ≺ Ti

Processing time
Commit time(4)

このうち，いずれかの順序関係を破棄することで T 上での半図 2 時刻のイメージ図順序関係を構築するプロセスが CSR におけるシリアルヒストリーの選択だとみなせる．
つまり，あるシリアルヒストリーが選択されたとき，トランザクションの順序関係は半順序で表せる．例えば，以下のような 5 つのトランザクションの集合 T を考える．

3. 2. 1 ユーザ時刻とシステム時刻
本研究では，以下の 4 つの時刻を考える．なお，各名称はFlink で用いられているものを参考とした [1]．
（ 1 ） Event time：ユーザが指定する任意の時刻．

Ta = r(x)w(y),Tb = r(y)w(x),T−∞ = w(x)w(y)w(z),Tc = w(z),T∞ = r(x)r(y)r(z)（ 2 ） Ingestion time：トランザクションがシステムに入力された時刻．
（ 3 ） Processing time：トランザクション内の read/write 命令

なお，T−∞ 及び T∞ は初期状態及び最終状態を確定させるためがシステム内で処理された時刻．のトランザクションである．このとき，{Ta, Tb, T−∞, T∞ } は x（ 4 ） Commit time：トランザクションがコミットした時刻．と y について，{Tc, T−∞, T∞ } は z について競合が発生しておある時間軸における各時刻のイメージとしての位置を図 2 に示り，式 (4) のように T 上には前順序の関係が成り立つ．次に，す．なお，点線の矢印はユーザ・システム間の通信を表し，点以下のシリアルヒストリー hs を考える．

線の矩形はシステム内におけるトランザクションの処理の保留を，実線の矩形が実際に処理が行われている時間を表す．

hs = T−∞Ta Tb Tc T∞

更に，これらの時刻は，event/ingestion time をユーザ時刻，このとき，T 上の順序関係は以下の半順序で表せる．

T−∞ ≺ Ta,Ta ≺ Tb,

T−∞ ≺ Tc,

Ta ≺ T∞,T−∞ ≺ T∞,

Tb ≺ T∞,

Tc ≺ T∞

これらの順序関係は，任意のデータ x ∈ D に対する処理の競合から生まれる順序である．つまり，シリアルヒストリーにおいて各データに対する read/write 命令の順序が全順序となるために最低限必要な順序関係である．このとき，順序関係の存在しないトランザクションの位置を入れ替えたシリアルヒストリー

hs′ は，元のシリアルヒストリー hs と等価である．例えば，上記において Tc は Ta とも Tb とも順序関係を持たない．したがって，T−∞ ≺ Tc ≺ T∞ さえ満たしていれば，Tc の位置を変更しても元のシリアルヒストリー hs と等価な結果が保証される．
このように，DBMS においてトランザクションの順序関係は前順序から始まり，何らかのシリアルヒストリーを選ぶことで半順序へと定まる．しかし，既存研究 [7, 11] で指摘されているとおり，ストリーム処理における連続的問合せでは一部のトランザクション間の順序関係が始めから半順序で表される．次節では，この半順序の関係がどのように定まるか述べる．

3. 2 データストリーム管理システムにおける順序関係

processing/commit time をシステム時刻として 2 つに分けられる．ユーザ時刻は，言い換えればシステムによって操作できない時刻を表す．Event time はユーザが任意に指定できる時刻であるため，当然システムでは変更できない．Ingestion time
も，トランザクションがいつシステムに入力されるかはユーザ次第であり，システム側では操作できない．一方で，システム時刻はシステムが自由に設定できる時刻である．つまり，processing/commit time はそのトランザクションないし命令をいつ処理/コミットするかであり，これらはシステム側で任意に操作できる．

3. 2. 2 時間を考慮したトランザクション
データストリームを扱う上で重要となるのは，全てのデータが本質的に時系列データとなり，連続的問合せにおいて時間に関する条件が本質的に存在する点である．例えば，時間窓を用いた問合せについて考える．窓幅 w のスライディングウィンド
ウでデータを集約するような連続的問合せを考えたとき，ある
時間窓 [t, t + w] の問合せは，暗黙的に時区間 [t, t + w] のデータが全て揃っていることを要求する．つまり，時区間 [t, t + w] に対する read 命令を行う前に，時区間 [t, t + w] への write 命令が処理されていることを前提としている．
これは， CSR において競合の発生するトランザクション間

本節では，ストリーム処理においてトランザクションの順序の順序関係のうち，一部はタイムスタンプによって確定するこ関係がどのように定まるかを述べる．以下では，まず本研究でとを意味する．例えば，時区間 [t, t + w] のデータ x に対して扱う時刻の概念について説明し，その後時間に関して拡張した

write を行うトランザクション Ti と時区間 [t, t + w] のデータ x

トランザクションを紹介する．最後に，トランザクション間のを read するトランザクション T j が存在するとき，ストリーム順序関係の決定方法を述べる．

処理においてその順序関係は Ti ≺ T j であることが暗黙的に要求される．既存研究 [7, 11] ではこの順序関係をユーザが定義する処理の DAG やストリームの到着順で表そうとしているが，の行われる時間が論理的に決まっていないこと，つまり実行ヒ

本質的には上述したようなタイムスタンプに基づく read/write

ストリーに配置される際にシステム時刻を用いて時刻が割り当命令の順序関係である．

てられることを示す．

なお，データの時系列化はユーザによって明示的に指定されread/write 命令を定めたことで，命令の集合が満たすべき順ない場合にも行われる．データを読み書きする際に event time

序関係を以下のように定められる．に基づいて時刻が指定されている場合，当然それらのデータは［定義 3］ 命令の集合 op が与えられたとき，以下の式から導か時系列データとなる．一方で，データを event time 以外の時刻れる op 上の順序関係 ≺ をタイムスタンプに基づく順序関係とに基づいて処理する場合もデータは時系列に基づいて処理されする．ただし，Wti nt (key) = {w(t, key)|w(t, key) ∈ op ∧ t ∈ tint }る．例えば， DBMS は read 命令を processing time で， writeとする．

命令を processing/commit time で行うシステムである．つまり，全ての read はその命令を行う時点（ processing time ）で最新の値を読み，全ての write はその命令を行った時刻を基に書き込∀r(tint , key) ∈ op, Wti nt (key) , ∅→ arg maxw(t,key)∈Wti nt (key)t ≺ r(tint , key)(10)■

みかつコミットと同時（ commit time ）に各データの最終状態を確定させる．

この定義は，read 命令で読む時刻を明示的に指定するとき，以上を基に，本研究におけるストリーム及びトランザクショ

read/write に関する競合による順序関係がヒストリ生成前に定ンを定義する．まず，全データとストリームの定義を示す．

まることを示す．ただし，順序関係が必要な write 命令は式 (9)［定義 1］ データストリーム管理システムで管理する全てのに基づいて定まり，読まれない過去の write とは直接順序関係

データの集合を D で表す．ストリーム S はデータ集合 D を構が定まらない点に注意する．

成する要素であり，以下の性質を持つ．

以上の定義に基づき，命令集合の linearizability を定める．

S⊆D(5)∀Si, S j ⊆ D, Si ∩ S j = ∅(6)

［定義 4］ 命令の集合 op 及び op 上の順序関係 ≺ が与えられたとする．≺ が read-from に基づく順序関係を全て含みかつ半順

また，ストリーム S 中の各データは以下の性質を持つ．ただし，ストリーム中の各データ d ∈ S はタイムスタンプ，主キー，値の 3 つの組 d = (t, key, value) で表し，それぞれを d.t のような形式で参照する．

∀d ∈ S, d.t ∈ N■

序であるとき，op は linearizable である．

つまり，命令集合 op を事前に与えられた read-from の順序関係を壊すことなく何らかの実行ヒストリー（全順序な命令列）で処理できるとき，その命令集合は linearizable である．
最後に，ストリーム処理におけるトランザクションを定める．
［定義 5］ トランザクション T = (op, ≺) は有限な命令集合 opと op 上の linearizable な順序関係 ≺ の組である．

(7)∀d, d ′ ∈ S, d.key = d ′ .key ∧ d.t = d ′ .t → d.value = d ′ .value■

3. 2. 3 トランザクション間の順序関係トランザクション間の順序関係を，各トランザクション間の(8)■以下，議論に必要ない限りあるストリーム S 中のデータ d ∈ S
も単に d ∈ D で表す．つまり，全てのデータの値をタイムスタンプ及びキーによって一意に指定できるものとして議論する．
次に，データに対する読み書きの命令を以下のように定義する．

read-from 関係を用いて定める．
［定義 6］ トランザクションの集合 T が与えられたとする．ト
∪ランザクション集合中の全ての命令の集合 OP = T ∈T T .op に対し，read-from に基づく OP 上の順序関係を ≺ で表す．このとき，各トランザクション間の read-from に基づく順序関係を以下の式に基づき与える．

［定義 2］ 読み書きの命令を r(tint , key) 及び w(t, key) で表す．

∀w(t, key) ∈ T .op, r(tint , key) ∈ T ′ .op,

ただし，tint はタイムスタンプに関する実区間である．より厳w(t, key) ≺ r(tint , key) → T ≺ T ′(11)密には，r(tint , key) は以下の式に基づきあるデータ d ∈ D を読み出す．

つまり，あるトランザクションで書いた値を別のトランザクションで読むことがわかっているとき，それらの間には事前に

r(tint , key) := arg maxd ∈ {d |d ∈D∧d.key=key∧d.t ∈ti nt } d.t (9)（実行ヒストリーに配置する前に）順序関係が定まるとする

また，w(t, key) はデータ d = (t, key, value) を D に追加する命令である．

■■

更に，定義 6 から，以下の定理が導ける．
［定理 1］ Read-from に基づくトランザクション集合 T 上の順

全てのデータアクセスをタイムスタンプに基づいて処理する

序関係 ≺ が半順序であるとき，全トランザクションの命令集合

ことで，ユーザ時刻に基づく命令を表す．なお，システム時OP は linearizable である．更に，OP から生成できる実行ヒス

刻に基づく命令は ∞ を用いて表現する．つまり，読み込みはトリーには，少なくとも一つ serializable なものが存在する． ■r([−∞, ∞), key)，書き込みは w(∞, key) として表す．−∞ は過去

つまり，ある時点におけるトランザクションの集合 T が与えらに行われた全ての書き込みが対象となることを表し，∞ は処理れたとき，各トランザクション間の順序関係が半順序で表せる

か確認することで，何らかの実行ヒストリーが作成可能であるか確認できる．加えて，トランザクション間の順序関係が半順序であるため，少なくとも一つのシリアルヒストリーが存在する．以降，T 上の read-from に基づく順序関係が半順序である
ことを，T が linearizable であると呼ぶ．

T が linearizable であるかは，多項式時間で判定できる．全命
令集合 OP 上の read-from に基づく順序関係は，任意の read/write
命令の組を確認することで作成できる．その後，命令集合 OP上の順序関係からトランザクション間の順序関係を導き，それが半順序であるか確認する．半順序であるかはトポロジカルソートなどで循環の有無を確認すればよいため，トランザクションの数とその間の順序関係の数に対して線形で判定できる．
つまり，全体としての計算時間は多項式時間に収まる．

4 おわりに本項では，データストリーム処理におけるトランザクション処理に着目し，ストリーム処理においてトランザクションが満たすべき順序関係について議論した．今後は，検討した内容について議論を進めアイデアを具体化すると共に，理論的な保証やシステムとしての実装を行う予定である．

謝

辞

本研究は，JSPS 科研費（16H01722, 18H06461）の助成，および，国立研究開発法人新エネルギー・産業技術総合開発機構（NEDO）の委託業務による．

文

献[1] Apache Flink 1.7 Documentation: Event Time. https://ci.
apache.org/projects/flink/flink-docs-release-1.7/dev/
event_time.html. Accessed: Feburary 24, 2019.
[2] Apache Flink: Stateful computations over data streams. https://flink.apache.org/. Accessed: January 6, 2019.
[3] Apache Hadoop. https://hadoop.apache.org/. Accessed: January 6, 2019.
[4] Apache Spark — unified analytics engine for big data. https://spark.apache.org/. Accessed: January 6, 2019.
[5] Samza. http://samza.apache.org/. Accessed: January 6, 2019.
[6] Spark Streaming | Apache Spark. https://spark.apache.org/streaming/. Accessed: January 6, 2019.
[7] I. Botan, P. M. Fischer, D. Kossmann, and N. Tatbul. Transactional
stream processing. In Proc. EDBT, pages 204–215, 2012.
[8] H. Chen and M. Migliavacca. StreamDB: A unified data management
system for service-based cloud application. In Proc. IEEE SCC, pages
169–176, 2018.
[9] J. Dean and S. Ghemawat. MapReduce: Simplified data processing
on large clusters. In Proc. OSDI, pages 137–150, 2004.
[10] R. Kallman, Y. Zhang, J. Hugg, D. J. Abadi, H. Kimura, J. Natkins,A. Pavlo, A. Rasin, S. Zdonik, E. P. C. Jones, S. Madden, andM. Stonebraker. H-store: A High-Performance, Distributed Main
Memory Transaction Processing System. PVLDB, 1(2):1496–1499,2008.
[11] J. Meehan, A. Pavlo, M. Stonebraker, K. Tufte, H. Wang, N. Tatbul,S. Zdonik, C. Aslantas, U. Cetintemel, J. Du, T. Kraska, S. Madden, and D. Maier. S-Store: Streaming meets transaction processing.
PVLDB, 8(13):2134–2145, 2015.
[12] B. Mozafari, J. Ramnarayan, S. Menon, Y. Mahajan, S. Chakraborty,H. Bhanawat, and K. Bachhav. SnappyData: A unified cluster forstreaming, transactions, and interactive analytics. In Proc. CIDR,2017.
[13] G. Weikum and G. Vossen. Transactional Information Systems: Theory, Algorithms, and the Practice of Concurrency Control and Recovery. Morgan Kaufmann Publishers Inc., 1st edition, 2001.
[14] M. Zaharia, T. Das, H. Li, T. Hunter, S. Shenker, and I. Stoica. Discretized streams: Fault-tolerant streaming computation at scale. In
Proc. SOSP, pages 423–438, 2013.