データストリーム処理システム実装に向けて

　数年前からサーバーサイド (Web アプリケーションサーバーやサーブレットエンジン、EJB サーバー, WSO2 Carbon Platformなど）や Eclipse などで OSGi フレームワークを使って実装されていることが多くなってきている。OSGi ベースで実装することにより、システムのコンポーネント化、モジュール化の基盤として使えるだろう。また、ネットワーク上からコンポーネントの追加・停止をJVM (Java Virtual Machine) を止めることなくできるので、DSMS (Data Stream Management System) において、実行時に動的にオペレータのデータフローを最適化していくような仕組みもこれを利用すれば、比較的容易に実現できるかもしれない。OSGiの実装としては、Apache Felix が代表的なので、少しずつ調べていきましょう。

Borealis Project

2008年版のソフトウェアリリースがありソースが見れますので、 System S/SPADE と比較してみてください。GSIC 佐藤君が少し論文を読んでくれましたが、また、続きをやりましょう。
http://www.cs.brown.edu/research/borealis/public/#software

評価に用いるアプリケーション及びデータ

今まで公開されている DSMS 関連の論文が弱いところは、リアルアプリケーションを使った評価が少なく、用いているデータが人工的であること。我々は、実データはさすがに使えなくても、現実的なケースに基づくデータを用いて評価していきたいものである。以下が考えられるデータの候補。

• 通信携帯会社の通話記録データ CDR (Call Data Record)：　誰が誰に何分通話したか？を記録したデータ。インドの携帯契約者は増加の一途を辿っており、その処理が追いつかないと言われる。このデータに関してはある程度ランダムに生成できるであろう
• ライフログデータ：　携帯電話の GPS データなど。各ユーザーの緯度、経度情報は何らかのユーザーの動きのモデルを作らないといけないが、サーバーの性能だけ考えるとあまりそれらの情報は重要ではなく、ランダムで良い
• Web データ：　楽天などのオンラインショッピングにログデータ。これは老木君が創作演習で作っている
• テキストデータ： ニュースの RSS フィードなど巷に沢山ある
• 環境に関するデータ：　ハーバード大学のデータは実際に使える
  

Offline + Online の Sharing

USENIX主催の Middleware 2008 で発表された論文

Enabling Resource Sharing between Transactional and Batch Workloads Using Dynamic Application Placement
David Carrera (Technical University of Catalonia - Barcelona Supercomputing Center, Spain)
Malgorzata Steinder (IBM Research, United States)

SEDA の論文（続き）

このエントリのさらに続き

- Matt Welsh による SEDA の過負荷時の制御手法について論じた論文. DSMS における Load Shedding に関しても大いに関係してくる。
Matt Welsh and David Culler, "Adaptive Overload Control for Busy Internet Servers"

- SEDA の肝は 2.2 章に書かれているが、ソフトウェアをステージに分割し、開発者に明示的にキューのコントロールを許すことによって、アプリケーションに特化したパフォーマンスコントロールが可能になること。System S 自身はキュー自身は見えていないので、その部分は UDOP (User-Defined Operator) で実装する必要がある。また、ステージ化することのその他の利点は、パフォーマンスのボトルネックを発見するのが容易になること。ステージごとのキューの状態（常に満杯になっているか？）を監視することで、容易にボトルネックがわかるし、そこのキューだけを制御しさえすれば良くなる。

- ニュースサイトなどではある突発的な事故などが起きると、システムの想定するリクエスト数の百倍、千倍のリクエストが到着することがある (これを Slashdot Effect とかつては呼んでいた）。しかし、これらを想定して Capacity Planning していては Over Spec なシステム構築となる。よって、ソフトウェア的にそれらを回避する必要がある。

- 回避方法としては、コネクション数やスレッド数の上限を設定しておき、リクエストを受け付けない方法と、もう一つは、結果のクオリティを下げる方法がある。後者は例えば、HTML を通常版と軽量版 (Ethernet パケットに収まるぐらい。例えば、画像も解像度を低くするなど）の２つを用意しておき、過負荷時には軽量版の HTML ファイルをクライアントに返す方法などがある。

- この従来の手法は、コネクション数とスレッド数をサーバー管理者が Web サーバーの設定ファイルなどに設定する必要があるが、これは経験的に決められたものであり、必ずしも最適値にならない。よって、低い値にしまいすぎて、スループットが出なかったり、大きくしすぎると、常にシステムが満杯状態で、すべてのリクエストに対してレスポンスタイムが悪くなるという事態に陥ってしまう。

- このようなケースは、Web サイトの運営者側からすると宜しくない状況で、レスポンス時間が長いと、顧客が離れていってしまう恐れが高まる。運営者からすると、平均的にすべてのユーザーに悪いレスポンスタイムを返すよりも、一部のユーザー（例えば１０％）には、きっぱりと、「今は混んでるから後から来てね」と断り、他の90％に対して、それなりに acceptable なレスポンスを返してしまった方が良いだろう。

- この論文では、上記のような問題に対して、パフォーマンスコントロールを柔軟に可能にするフレームワークを SEDA 上に構築。Web Mail システム Arashi を構築して評価している。

- SEDA のキューに入ってくるメッセージの処理時間（レスポンスタイム）をモニタリングし、その傾向によって性能制御する仕組みを設計。例えば、レスポンスタイムが悪くなってきたら、ある一定数の確率でリクエストを Drop したり、他のサーバーに割り振ったりする。

- キューを監視するコントローラを設定し、リクエストがキューに入った時間と、処理を完了し、クライアントに結果を返すまでの end-to-end の時間を測定。その TAT (Turn around time) がある閾値以上になったときには、Load Shedding アルゴリズムを発動させ、一部のリクエストを優先して処理し、一部はあきらめることによって、全体のレスポンスタイムの中央値を良くする手法を提案。

DSMS のベンチマーク

Linear Road: A Stream Data Management Benchmark, VLDB 2004
http://www.cs.brandeis.edu/~linearroad/

Aurora (Brandeis 大学、ブラウン大学、MIT の共同DSMSプロジェクト）と STREAM (スタンフォードのDSMSプロジェクト）のストリーム処理システムのベンチマークシステム。

渋滞度や事故に応じた課金体系（混んでいるときに通行すれば高くなる）を採用する高速道路課金システムをシミュレート。RDBMS (System X と製品名を隠している）と Aurora とを比較し、Aurora が5倍程度、性能を発揮。(STREAMのデータはなし）

チャレンジは以下のとおり。

(1) 入力は？
ただ単にランダムではいけないこと。車などの動きをシミュレーション
--> MITSIM (MIT Traffic Simulator) という交通シミュレーターを使用
(2) 性能比較のメトリックスをどうするか？
クエリが Continuous であるため Completion time ではなく、Response Time と スループット（処理したクエリ数）を用いる
--> L-Rating というスコア（レスポンスタイムとスループット）を導入
(3) Validation をどうするか？　
ベンチマークとして、同じクエリに対して、その時々の状況（混雑度）によって出力となる料金が異なる。よって、ベンチマーク自体、
　　 ひとつのクエリに対して、複数の料金を返す
(4) 現在、Continuous Query は存在しない
--> 特定のクエリ言語を作るのではなく、フォーマルに predicate calculus を使用

各車両からは位置情報を30秒に1回、 1％の確率でクエリをシステムに発行。（その中で50％は account balance, 10% は 1日の合計料金、40%
は目的地までの時間）

DSMS はそれら常に到着するデータを受信し、以下の処理を実行
(1) あるセグメント上に車が何台存在するか、平均速度はどのくらいかを1分ごとに計測
(2) 事故も見地リアルタイムに渋滞度、事故度を判定
(3) クエリ（セグメントの統計及び事故の状況を見て料金や目的地に到着するまでの時間）を処理し、各車両に伝える

Linear Road はあくまでベンチマークの仕様。論文では、Aurora と System X (RDBMS の製品）を用いて Linear Road を実装。System X では、300行のクエリとコードからなる Stored Procedure として実装。

Telegraph CQ

UC Berkeley の Michael Franklin らの研究室のDSMS システム. System S の SPADE のように データフロー型の言語ではなく、DBMS の世界では標準の SQL を拡張した Continuous Query 用の言語を提唱。

http://telegraph.cs.berkeley.edu/にはソフトウェアのソースコードや論文が公開されている。

産総研

　本日は、産総研の秋葉原オフィスにて Ninf の１５周年記念ワークショップというものに参加してきました。Ninf はグリッドのような広域ネットワーク環境上（高遅延かつ低セキュリティ）で RPC (Remote Procedure Call) を実現するミドルウェアであり、我が国の重要なグリッド研究成果の一つとも言えます。

　久しぶりに、産総研の研究員の方々とお話しする機会があり、ストリームコンピューティングの議論もしてきました。参考のために、そのフィードバックをしておきます。

- ストリームコンピューティングにとって、汎用的なプログラミングモデルは本当に必要なのか？極端に言えば、すべてUDOP（コールバック関数とも言える）として実装することになるのでは？

- StreamDNA は、画像認識だけではなく、２次解析（Smith-Waterman 法のGPU版）のストリーム化までやって初めて研究的な価値が出るのでは？また、StreamDNA のリアルタイム性はシビアではないのでは？

- Load Shedding にこそストリームコンピューティングの本質があるのでは？世の中のアプリケーションとして、どんなタイプのアプリが Load Shedding を許容するかを分類すべき

-ストリームアルゴリズムなどOne-path のアルゴリズムのみを対象としているのか？必ずしも、One-path だけではなく、ある程度のバッファリングは許容する

- awk はストリームエディタと言える。

AT&T

米国の AT&T も積極的に データストリーム処理の研究を進めています。
少し古いですが、VLDB 2003 にこんなチュートリアルもあります。
http://www.vldb.org/conf/2003/papers/S38P03.pdf

DSMS のプログラミングモデル

System S の現在のプログラミングモデルは、データストリーム処理に特化した言語である。データストリーム処理なので直感的に記述することができる。しかし、専用言語ではなく、通常のプログラミング言語から呼び出せないのかという質問を投げかける人もいるのも事実だ。

Ruby, PHP, Python などのプログラミング言語からシームレスにデータストリーム処理できるようにするのも一つの研究テーマになるであろう。

データストリーム処理の投機処理

とりあえず、思いついたので、メモ。

データストリーム処理への QoS の指定

データストリーム処理上のアプリケーション開発者は、QoS を指定する。スループット、ストレージ容量の上限、レスポンスタイムの上限など。処理系は、その QoS に従って、適切なリソース割り当てや特殊なハードウェアへの割り当てを実行する.

ジョブ毎にプライオリティが存在しても面白い。

データストリーム処理とバッチ処理への動的リソース割り当て

　Web への応用を考えているのだが、DSMS の一つの応用はリアルタイムログ解析、ログ圧縮などである。しかしながら、トラフィックが少ない時には、DSMS に割り当てたハードウェアリソースもアイドル状態のままになる。データ解析は DSMS のみで完結するわけではなく、後処理としてやはり従来手法のバッチ処理が必要になる。よって、上記のアイドル状態のときは、バッチ処理にアイドル状態のリソースを割り当てる方が好ましい。一つの研究テーマとしては、これをダイナミックにコントロールするメカニズムを考案することである。

上記の考え方は、Web のみならず、様々なアプリケーションに適用可能なはずだ。

DSMS と GPU との相性

本日は GSIC にて GPU 講習。松浦君、森田君、B3 の老木君、そして鈴村が参加。有益な一日になってくれたことを期待します。

　少し冷静になって、DSMS と GPU との相性を考える必要がありますが、計算が比較的軽く、低レイテンシが必要なアプリケーションは、CPU と GPU のデータ転送のコストがネックとなるため、向かなさそうです. アプリケーションの種類としては 以下のA～D がありますが、B, C が DSMS に向くのでしょう。

A) 計算が軽い。低レイテンシが必要
- 例) アルゴリズムトレーディング。1msec のレイテンシを争う。
B) 計算が軽い。レイテンシはそこまで重要でない。
- 例) 株の移動平均線
C) 計算が重い。レイテンシは低い方がよい
- 例）　画像解析、音声解析. 異常解析（O(N^3) の固有値問題などを解く等)
D) 計算が重い。レイテンシはそこまで重要でない
- DSMS である必要がない。従来のバッチ型処理で良い

アプリケーションによって、要求するレイテンシがある程度決まっていると思うので、実行時にそれを指定して、その範囲内で、CPU と GPU の実行をバランスさせるのも面白いです。

コールセンターにおけるストリーム処理

　世の中には、様々な業界のコールセンターがある。生命保険、自動車保険、証券会社、航空会社、メーカーの修理受付窓口など。コールセンターは、企業と顧客とを結ぶインタフェースでもあり、Customer Satisfaction を向上させる重要な役割を持つ。あまりにも対応が悪いコールセンターだと、その企業の信用度も落ちる。

しかし、昨今は、アメリカのコールセンターは、コスト削減のために、インドのコールセンターを使っていたりする。アウトソーシング化だ。アメリカに滞在していたときに、United Airlines のコールセンターにかけたのに、明らかにかかったのは、インド人なまりの英語。IP 電話で音質もあまり良くなく、あれは、明らかにインドにかかっていたのであろう。

そのような中、次世代のコールセンターではオペレータの品質を保ち、Customer Satisfaction を向上させるために、以下のようなことをリアルタイムで検知する実証実験をしたいらしい。

- オペレータが言うべきことをちゃんと言っているか？
- お客さんがどのような状態か？怒っているか？Neutral か？
- お客様からどのような苦情を受けているか？それが Critical か？

入力データは、音声データか、または、音声解析技術によって落としたテキストデータかもしれない。いずれにしても、膨大な数のデータが流れてきて、かつ、それらをリアルタイムにオペレータに伝えなくてはならない。

これはまさにストリームコンピューティングの一つの適用分野だ。

皆さんも、世の中を見て、どのような分野にストリームコンピューティングを適用できるかを考えてみてください。

仮想マシンを活用したデータストリーム処理

Xen や VMWare などの仮想マシン上でデータストリーム処理システムを動かすことにより、耐故障性に対応し、かつ、突発的に負荷が増大したときにも、仮想マシンの数を増やすことにより対応する。

データストリーム処理システムのジョブスケジューリング手法に関する研究

分散システムでは長年、ジョブスケジューリングが一つの研究テーマとなっている。データストリーム処理システムにおいても同様に、DAG (Directed Acyclic Graph) として表現されるアプリケーションを計算リソースにどのように割り振るかは一つのチャレンジングなテーマだ。

様々な手法が考えられるが、スループットを最大化することを目的とすると、グラフ理論の最大フローアルゴリズム (Ford-Fulkerson など) が活用できるのではないだろうか。