[StreamCloud] 研究概要

以下、StreamCloud プロジェクトの研究課題及び解決策案、実装案を書きます。

[前提]
- 有限個の計算資源しかなく、マシンの増強は様々な要因で困難
（マシン設置環境、電源容量の限界、経済的な問題）
- 複数のアプリケーション（ストリーム処理のみ）が混在し、それぞれ SLA (応答時間の上限など）が異なる
- ターゲットとするアプリケーションのトラフィックの変動が激しく予測困難（周期性はところどころある）
- Load Shedding などデータを間引いて処理することは考えず、すべてのデータを処理する


[課題]
上記の前提を受けて、すべてのアプリケーションの SLA を満たすような手法及びアーキテクチャとは何か

[我々の解決手法の概要]
自身が保有する計算資源で間に合わない場合には、その処理をクラウド上の DSMS に委譲する

[本研究の Technical Challenge]
(A) どのようなアルゴリズムを持って、クラウド上に処理を委譲するか？また、クラウドのVMMは何個立ち上げれば SLA を守れるか？
(B) VMM の起動・停止の時間がかかるため、どのようなタイミングでクラウドの VMM を起動しておくか、停止するか？
(C) 現行のデータストリーム処理システムでは実行時にノードを追加する仕組みがないのでそれをどのように解決するか？

[解決策の前提]
- クラウドの定義とは、Amazon 及び Eucalyptus などの Iaas のみをここでは指し、VMM をオンデマンドで提供する環境とする。また、条件としては、ローカルの計算資源と比較して、レイテンシが相対的に高く、バンド幅も制限される。また、VMM 上の DSMS の稼動なので、ランタイムのオーバーヘッドが存在する。
- ただし、クラウドを利用している他のユーザーのジョブの混入によるパフォーマンスの劣化は考えなく、占有できる
- 自身の保有する計算資源で処理が間に合わない場合のみを以下の議論の対象にする
- SLA は本研究では単純化して応答時間とする

[解決策の詳細]
(1) クラウドへの処理委譲の条件 (上記A に対する解決策)
　(a) 応答時間が緩やかなアプリケーションの場合には、クラウド実行時の実行時間 =
（クラウド上へのデータの送受信時間）＋（VMM によるオーバーヘッドを加味した計算実行時間)
と定義できるので、この時間が応答時間を上回ってなければ委譲する
　(b)クラウド実行時の実行時間は前処理によって統計値を取っておき、データレートに応じた
計算時間を線形式で定式化し、（重）回帰分析を行って係数を求め、数理的なモデル式を構築する
　(c)上記の実行時間予測よりも応答時間が下回る、または許容範囲内に収まっていれば、クラウドに委譲する

(2) VMM の起動・停止のタイミング
- 自身の保有する計算資源で処理できないデータレートをあらかじめ求めておく.これは待ち行列理論により
　レイテンシが大幅に増加する段階のデータレートで判断することができる。
- データレートの変化を定期的にチェックし、変化の勾配がある閾値を越えた場合には、即座に
クラウドに対して、VMM の起動リクエストを発行する。起動リクエストと VMM の起動（または Resume) の
時間はあらかじめ測定しておき、それを加味した上で、上記の閾値を設定する
- 逆にデータレートが自身の保有する計算資源で処理できる量以下の値が一定時間続けば、クラウドに対して
VMM の停止リクエストを発行する

[実装の詳細]
- クラウド環境は Eucalyptus を使用する.
- クラウドの高レイテンシ及び低バンド幅を再現するために ns2 などのネットワークシミュレータを用いる
- データストリーム処理システムは System S を使用する。
- VMM の起動、停止は UDOPオペレータ内で実装し、その中から Eucalyptus への起動・停止リクエストを発行する
- 現行の System S には動的にホストを追加する機構がない (ジョブの再投入時に有効になる）ので,
クラウド上の VMM のホストを実行時に認識させる仕組みを設計し、実装する

[評価アプリケーション]
アプリケーションおよびシナリオはいろいろ考えられるので、また考えましょう。
手持ちのものとしては、VWAP (低レイテンシ), SST (1次元のみ, レイテンシ緩やかな場合を想定), Twitter などが考えられます

重回帰分析による 適応的なウィンドウサイズの決定

森田君が計測してくれた以下のデータに対して 統計言語 R を用いた重回帰分析を行ったので報告します。

実行時間 T は、CPU におけるウィンドウサイズ Wcと GPUにおけるウィンドウサイズと Wg の2変数の組み合わせによって、実験的に得られているので、実行時間 Tを Xc, Xg から説明する重回帰式を求めてみましょう。

T= a0 + a1 * Wc + a2 * Wg

以下, R 言語を持ちた分析. 第１フィールドに実行時間、第2フィールドに CPU のウィンドウサイズ、第3フィールドに GPU のウィンドウサイズから構成されるCSV 形式のデータを読み込みます。

> data1<-read.csv("C:/temp/streamgpu1.csv")

Time,CPU,GPU

1.8883369,50,500

2.3832237,50,600

2.7971815,50,700

3.2233268,50,800

3.773477,50,900

4.4216436,50,1000

5.1433544,50,1100

5.8385782,50,1200

6.5608456,50,1300

...

R による重回帰分析の実行と結果の表示

> data1.lm<-lm(Time~CPU+GPU,data=data1)

> summary(data1.lm)

Call:

lm(formula = Time ~ CPU + GPU, data = data1)

Residuals:

Min 1Q Median 3Q Max

-26.076 -4.510 1.308 5.718 30.277

Coefficients:

Estimate Std. Error t value Pr(>|t|)

(Intercept) -2.638e+01 1.488e+00 -17.73 <2e-16>

CPU 4.471e-02 2.677e-03 16.70 <2e-16>

GPU 2.541e-02 6.373e-04 39.87 <2e-16>

---

Signif. codes: 0 ‘***’ 0.001 ‘**’ 0.01 ‘*’ 0.05 ‘.’ 0.1 ‘ ’ 1

Residual standard error: 8.565 on 249 degrees of freedom

Multiple R-squared: 0.8824, Adjusted R-squared: 0.8814

F-statistic: 934 on 2 and 249 DF, p-value: <>

決定係数は 0.8824, 調整済みの決定係数は 0.8814

次に係数の算出。

> round(coefficients(data1.lm),6)

(Intercept) CPU GPU

-26.376838 0.044705 0.025407

これによって、以下の関係式が得られることがわかります。

T (s) =-26.38 + 0.044705 * Wc + 0.025407 * Wg

次に、説明変数 Xg, Xc に相関関係があると思われるので、以下のように相互作用を考慮した回帰係数を求めます。

> data1.lm2<-lm(Time~(CPU+GPU)^2,data=data1)

Call:

lm(formula = Time ~ (CPU + GPU)^2, data = data1)

Residuals:

Min 1Q Median 3Q Max

-25.4525 -4.4603 0.4802 4.4581 31.8098

Coefficients:

Estimate Std. Error t value Pr(>|t|)

(Intercept) -1.258e+01 2.023e+00 -6.217 2.13e-09 ***

CPU 7.909e-03 4.752e-03 1.664 0.0973 .

GPU 1.621e-02 1.175e-03 13.798 <>

CPU:GPU 2.453e-05 2.759e-06 8.891 <>

---

Signif. codes: 0 ‘***’ 0.001 ‘**’ 0.01 ‘*’ 0.05 ‘.’ 0.1 ‘ ’ 1

Residual standard error: 7.473 on 248 degrees of freedom

Multiple R-squared: 0.9108, Adjusted R-squared: 0.9097

F-statistic: 844.2 on 3 and 248 DF, p-value: <>

決定係数は 0.9108, 調整済みの決定係数は 0.9097で、相互作用効果を考慮していない場合の 0.8824, 0.8814 より増加しており、相互作用効果を考慮したモデルの当てはめがよいと言える。

次に係数の決定。

> round(coefficients(data1.lm2),6)

(Intercept) CPU GPU CPU:GPU

-12.578282 0.007909 0.016208 0.000025

Y(g,c) = a0 + a1 * Wc + a2 * Wg + a3 Wc Wg

T (s) = -12.578282 + 0.007909 * Wc + 0.016208 * Wg + 0.000025 * Wc * Wg

前処理の実験データによって上記の回帰式を得ることができたので、応答時間（異常検知までの最大値）の T (秒) がアプリケーションによって与えられ、かつ Wg のウィンドウサイズを 1000 と設定すれば、この式によって、適切な CPU 側でのウィンドウサイズ Wc を以下の式によって求めることができます。

Wc = (T + 12.578282 - 0.016208 * Wg) / (0.007909 + 0.000025 * Wg)

例えば、T=5 と設定したときは、Wc = 41.6385。T=30 と設定したときは Wc = 801 と求められます。

以上です。

2010年度前期輪読テーマ

4月から始める Stream Computing SIG (Special Interest Group) のテーマをそろそろ考えていきたいと思います。輪読するだけでなく、研究に関する議論も行っていきましょう。プロジェクトミーティングは別途行うとして、こちらでは主に関連研究や論文を読んで、それぞれの研究力をつけていきたいと思います。

• Load Shedding に関するテーマ: Data Stream Load Shedding by Amit Ahuja の本
• Virtualization の仕組み (準仮想化と完全仮想化の違いなど）やクラウド関係
• SQL のストリームへの拡張 (StreamSQL, CQL など）
• StreamGPU 関連
• 耐故障性、高信頼性、データ保証などを取り入れた DSMS

[StreamDS/StreamShedder] Data Stream Load Shedding - Dynamically Managing Channel Capacity by Amit Ahuja

今年は、DSMS における Load Shedding にも本格的に取り組んでいきたいので、以下の本を誰か一人に読んでもらいたいと思います。本の内容は、ある研究者が提案している一手法なので、この周辺の研究の様々な手法を紹介しているわけではないことには注意が必要ですが、論文の書き方、数理的なモデルの構築方法、アルゴリズムの書き方は、Load Shedding に関わらない人にとっても多いに参考になると思います。

Data Stream Load Shedding - Dynamically Managing Channel Capacity by Amit Ahuja　(Amazon)

[当研究の大枠]

当研究では以下の2つの Load Shedding を取り上げているが、特に (1) の手法に焦点を当てて論じている。

(1) intra-stream load shedding: ある特定のストリームの中で重要度の低い属性を間引くことで、与えられた通信チャネルの限界容量を越えないようにする。この研究では、間引く単位は、”タプルではなく”、”属性”単位で間引くこと、そして、Source で間引いたデータを recovery する機能、の2つがユニークだと主張する。

(2) inter-stream load shedding: 単一のチャネルを複数のストリームが流れ、その総和がチャネルの限界容量をコ越える際の問題。すべてのストリームを管理する central load shedder というコンポーネントが存在し、それが各ストリームのデータ転送レートを監視し、どのストリームのどの属性をどのくらい間引けば（ぎりぎり）限界容量を越えないかを判断する

[分散を用いた属性のランキング決め]

ストリームの属性の判断の仕方は、すべての属性のデータの「分散」を求めて、分散が大きいものはより重要度が高く、小さいものは「重要度が小さい」とする。トレーニングセットを用いて、ストリームのデータセットのすべての分散を計算し、タプル中の各属性のランクを決める (3.2.3.2)

[属性のランキングを決める上での工夫]

タプルの属性のランクを決める上で、ある属性（例えば砂漠上の降水量）が偶然そのトレーニングセットで突発的な変化（ある時間帯だけ瞬間的に雨が降った、等）が含まれている場合には、その属性のランキングが高まってしまい、より重要な属性（例えば砂漠の温度の変化）を落としてしまう可能性がある。ランキングの精度を高めるために、当研究では "移動平均 (Moving Average）" を使用している。移動平均には、スライディングウィンドウ中に存在するあるデータリスト（時系列順に揃っているデータ）の各データに対して、すべて同じ重みをつけて計算する Simple Moving Average (SMA) と、より新しいデータにより高い重みを付けて計算する Exponential Moving Average (EMA) が存在する。EMA の計算方法は p35 の (3.2) によって計算される.(3.2.2, p33-) 後の評価では、SMA 及び EMA を比較するが、当然ながら EMA の方がより良い結果を得ている。

[間引く量]

Load Shedding では、「何を間引くか？」と「どれくらい間引くか？」の２つの課題があるが、3.2.3.1 に記載されているが、バンド幅を B とし、ストリームのデータレートを R とすると、当然、 (R-B) の分だけ取り除く必要があり、かつ上記の属性のランキング決定に基づいてどの属性を何個取り除くかを決定する。属性によってバイト数が異なるため、それももちろん考慮する。

[適応的な Load Shedding Schema の変更方法]

Load Shedding Scheme (間引く戦術）自体は、静的に決められるものではなく、ある時間間隔 (インターバル）を持って再評価し、その時間帯で最適な Scheme に更新する必要がある (3.2.3.3, p41). この再評価のインターバルだが、短すぎるとパフォーマンスに影響を与えるし、長すぎると短期的な変化に対応できない。そこで、当研究では、まず、初期値のある程度短いインターバルをもって計算。次の再評価の時に、前回の Scheme （属性のランキングなど）と比較し、差がほとんどなかったら、インターバルの時間を2倍に設定する。次の評価の際に、 Scheme と設定した閾値以上異なる場合には、インターバルを初期値にリセット。このアルゴリズムが p42 に記載されており、"Adaptive Load Shedding Scheme Re-evaluation Algorithm" と呼んでいる。

[間引いたデータの復旧方法]

エンドユーザーは、Load Shedding が実行されたときに、ソース側で間引かれた属性を要求する場合がある。Shed したすべてのデータをディスクに保存しておくことは無理なのでエラー閾値（Error Threshold Value: 返された値が間違っている場合）とソースサイドにおける計算・ディスク容量とのトレードオフ（保存するにもCPUパワーは当然使う）になる。この問題のために、当研究では、「Euclidean Distance」と 「Synopsis Recovery Algorithm」の２つを提案している。

　後者は、Synposis、つまりサマリーデータ (Synopsis Recovery Matrix と呼ぶ）を保存しておき、Shed しようとしているデータと最も直近ののデータの差がエラー閾値以上であったら、Synopsis に貯められているデータの更新をし、閾値以下であったら更新しない。このエラー閾値はアプリケーション毎に異なり、インターネットモニタリングよりも患者のモニタリングの場合の方が当然閾値がきつくなる。

性能分析ツール: nmon

性能測定の結果グラフの可視化ですが、nmon というものがあります。

nmon : http://nmon.sourceforge.net/

バイナリは Intel or AMD 64 bit processor & 64 bit Linux でRHEL54 はいまのところありませんが、

ソースコードがあるので、それをコンパイルしてみて試してみてください。

Xen vs KVM

Xen 対 KVM.

http://virt.kernelnewbies.org/XenVsKVM

KVM はここに来て急速に発展してきているので、上記の性能比較も変わってそうだ。

情報処理学会全国大会７２回

松浦くん、森田君ともに発表がありました。大変お疲れさまでした。また、松浦君は「学生奨励賞」を受賞されました。おめでとうございます。

Libvert

Xen などの Hypervisor を C などの言語からコントロールする API

http://www.libvirt.org/index.html