PDF 本研究室志望者へ鈴村研究室大規模データ処理・ストリームコンピューティング

(1)

2009/10/23

ストリームコンピューティング

- データ ^{データ爆発時代} ^データ ^データ ^爆発時代 ^爆発時代 ^{爆発時代の} ^の ^の ^の新 ^新たな ^新 ^新 ^たな ^たな計算 ^たな ^計算 ^{計算パラダイム} ^計算 ^{パラダイム} ^{パラダイム} ^{パラダイム}

IBM 東京基礎研究所鈴村豊太郎

(2)

本日本日本日

本日ごごごご紹介紹介紹介する紹介するするする内容内容内容内容

ストリームコンピューティングの概要 InfoSphere Streams の紹介

実証実験のご紹介

SPADE の主要機能のご紹介

(3)

背景背景背景背景

ネットワーク技術の進化やセンサーデバイスの普及により、様々なデータがリアルタイムに利用可能になりつつある

利用可能なデータは爆発的に増加している（例：科研費・情報爆発）

たとえば、

– 製造業：各製造工程のセンサー, RFID – 自動車： GPS センサー、テレマティックス

– ^医療：医療機器のセンサー（心拍数、心電図、脈拍数 etc) – ^交通： ^交通管理

– 科学技術計算：電波天文学、地震センサー – セキュリティ：遠隔監視、サイバーセキュリティ

(4)

データの特性の変化：データ爆発の時代の到来

Explosion in market data volumes

New York stock exchange trade

& quote volume

Capital market data volumes rose by 1,750% during 2003-06

Explosion in market data volumes

New York stock exchange trade

& quote volume

Capital market data volumes rose by 1,750% during 2003-06

Consumer Generated

Media

Intelligence & Surveillance

Sensors & Embedded

Internet Devices

データデータ

データデータ量量量量のののの爆発爆発爆発、爆発、、及、及及及びびデータびびデータデータのデータのの種類の種類種類の種類ののの多様化多様化多様化多様化

NYSE の^の^の^{のマーケットデータ}^{マーケットデータ}^{マーケットデータ}^{マーケットデータの}^の^の^の総量^総量が^総量^総量^が^が^{が指数関数的}^{指数関数的}^{指数関数的に}^{指数関数的}^に^に上昇^に^上昇^上昇^上昇

(5)

アプリケーションアプリケーションアプリケーション

アプリケーション側側側側からのからのからのからの要求要求要求の要求ののの変化変化変化変化

即応性・意思決定までの時間の最小化

– 特定の条件に合致したときに即座に対応することで、新たな価値を生む

• 例）半導体の製造工程における異常検知

• 例）集中治療室の患者の状態の検知

様々な情報ソースを利用した高度な分析

– 様々なデータを利用してこれまで不可能だった高度な分析を実現する

• 例）アルゴリズムトレーディング

（22箇所のトレーディング情報や、ニュース情報を組み合わせた自動売買） – ^{データ形式が異なる}

• 構造化データ、非構造化データ

– データ到着レートが時間帯によって異なる

(6)

新新新

新たなたなたなたな計算計算計算パラダイム計算パラダイムパラダイムパラダイムのののの必要性必要性必要性必要性

データデータデータ

データののの特性の特性特性特性ののの変化の変化変化変化

アプリケーションアプリケーションアプリケーション

アプリケーションののの特性の特性特性特性ののの変化の変化変化変化

新

新たなたなたなたな計算計算計算計算パラダイムパラダイムパラダイムパラダイム

の

のの

の必要性必要性必要性必要性

(7)

ストリームコンピューティングストリームコンピューティングストリームコンピューティング

ストリームコンピューティングとはとはとはとは？？？？

データウェアデータウェアデータウェアデータウェア

ハウスハウスハウスハウス

センサーネットワークセンサーネットワークセンサーネットワークセンサーネットワーク

データ

ストリームストリームストリーム

ストリーム処理処理処理処理システムシステムシステムシステム

Process data as it is continuously generated

Extracting and organizing information and intelligence

意思決定意思決定意思決定

意思決定までのまでのまでのまでの時間時間時間時間ををを最小化を最小化最小化最小化

センサーセンサー

センサーセンサーなどなどなど複数など複数複数の複数の情報のの情報情報ソース情報ソースソースソース（（（構造化（構造化、構造化構造化、、非構造化、非構造化非構造化データ非構造化データデータデータ含含む含含むむ）む）））かかかからら

らら時時時時々々々々刻刻刻刻々々到達々々到達到達到達するするするするデータデータデータからデータから、からから、、複合的、複合的複合的に複合的ににに分析判断分析判断し分析判断分析判断しし、し、、、即座即座即座即座のののの意思決定

意思決定意思決定

意思決定ををを可能を可能可能可能にするにするにするにする計算計算計算計算パラダイムパラダイムパラダイムパラダイム

(8)

ストリームコンピューティングにににに適合適合適合適合するする応用領域するする応用領域応用領域応用領域ののの条件の条件条件条件

以下以下

以下以下のののの条件条件条件条件のうちのうち_{のうち1つ}のうちつつ以上つ以上以上以上をを満をを満満たす満たすたすアプリケーションたすアプリケーションアプリケーションアプリケーションがががが候補候補候補候補

1. 情報ソース（センサー）の数の存在または増加の可能性 2. 多様なデータ形式（構造化、非構造化）を活用した、複合的

なデータ分析の必要性

3. 実時間（低レスポンス）への要求

4. システムのキャパシティに対して（相対的に）

• ^{高い到着データレート}

• 必要なストレージの必要性

(9)

例例

例）例）））ハリケーンハリケーンハリケーンのハリケーンののの影響影響影響による影響による株売買によるによる株売買株売買の株売買ののの自動自動自動リコメンデーション自動リコメンデーションリコメンデーションリコメンデーション

ハリケーンハリケーンハリケーンハリケーンのののの

進路予測進路予測進路予測進路予測

インダストリーインダストリーインダストリーインダストリーへのへのへのへの

ハリケーンハリケーンハリケーンハリケーンののののインパクトインパクトインパクトインパクト

ポートフォリオポートフォリオポートフォリオポートフォリオ

へのへのへの

へのインパクトインパクトインパクトインパクト米国大洋大気庁

米国大洋大気庁米国大洋大気庁

米国大洋大気庁からのからのからの_からのRSS ニュース

ニュースニュースニュースををを取得を取得取得取得

NYSE (ニューヨークニューヨークニューヨーク証券取引所ニューヨーク証券取引所証券取引所_{証券取引所) から}からからから VWAP(出来高加重平均出来高加重平均出来高加重平均)を出来高加重平均ををを計算計算計算計算

企業財務情報開示企業財務情報開示企業財務情報開示

企業財務情報開示ののの情報の情報情報情報ををを取得を取得取得取得

ニュースサイトニュースサイトニュースサイト

ニュースサイトのののの情報情報情報情報をををを取得取得取得取得

ハリケーンハリケーンハリケーン

ハリケーンののの影響の影響影響影響によるによるによるによる株株株株のの

のの売売売り売りりり買買買買いのいの判断いのいの判断判断を判断ををを決決決決定

定定定 P/E (株収益率^株収益率株収益率)を^株収益率 ^を^を^を計算^計算^計算^計算

ストリームコンピューティングをををを必要必要必要必要とするとするとするとする条件条件条件を条件を満をを満満満たすたすたすたすアプリケーションアプリケーションのアプリケーションアプリケーションののの例例例例

(10)

従来従来従来

従来のののの処理方式処理方式処理方式処理方式でで実現困難でで実現困難実現困難なことは実現困難なことはなことはなことは？？？？

ログログログ

ログややややデータウェアハウスデータウェアハウスデータウェアハウスにデータウェアハウスににすべてのにすべてのすべてのすべてのデータデータをデータデータををを一旦一旦一旦一旦ためてためてためて、ためて、、、バッチ

バッチバッチ

バッチでででで処理処理処理処理するのがするのがするのがするのが基本基本基本基本

– Google MapReduce, Data Grid (例^例^例：^例^：^：加速器^：^加速器^加速器^{加速器データ}^データ^データ^データの^{のペタバイト}^の^の^{ペタバイト}^{ペタバイト}^{ペタバイト級}^級^級^級データ^データの^データ^データ^の^の^の解析^解析^解析解析) この

このこの

この処理方式処理方式処理方式処理方式でででで困難困難困難困難なことはなことはなことは？なことは？？？

– ^{データソースの}^{データソース}^{データソース}^{データソース}^の^の増加^の^増加に^増加^増加^に^に従^に^従^従って^従^って^って^って、^{、すべての}^、^、^すべての^すべての^{すべてのデータ}^データ^データ^データを^を保存^を^を^保存^保存^{保存することが}^{することが}^{することが困難}^{することが}^困難^困難^困難

• ^必^必^必^{必ずしもすべての}^{ずしもすべての}^{ずしもすべての}^{ずしもすべてのデータ}^データを^データ^データ^を^を保存^を^保存^保存する^保存^する^する^する必要^{必要はなし}^必要^必要^はなし^はなし（^はなし^（^（^（例^例^例^例：^{：連続}^：^： ^連続^連続^{連続データ}^データ^データ^データ）^）^）^） – ^到達する^到達^到達^到達^する^{するデータ}^する^データ^データの^データ^の^の^{のリアルタイム}^{リアルタイム}^{リアルタイム}^{リアルタイム性}^性を^性^性^を^を十分^を^十分^十分^十分に^に^に^に生^生^生^{生かしきれない}^{かしきれない}^{かしきれない}^{かしきれない}

Database/d ata warehouse

Sensor Network

data

Process data as it is continuously

generated

Minimizing time to react

Transaction

processing _Batch

processing

(11)

2009/10/23

ストリームコンピューティング

ストリームコンピューティング処理処理処理システム処理システムシステムシステム

InfoSphere Streams

(12)

InfoSphere Streams: ストリームコンピューティングシステムストリームコンピューティングシステムストリームコンピューティングシステムストリームコンピューティングシステム

ストリームコンピューティングを実現するソフトウェアシステムシステム構成技術

– SPADE (System S Stream Processing Application Declarative Engine)

• ストリーム型アプリケーションの生産性を高めるプログラミング言語

• 下位の分散環境に最適化されたコードを SPADE コンパイラが生成 – ランタイム：最適スケジューリング、耐故障性を実現

数多くの実証実験プロジェクト – ホームランドセキュリティ

– 証券取引・アルゴリズムトレーディング (カナダ TD Bank) – 製造業・リアルタイム異常検知、ITS (高度交通システム） – 環境, 遠隔医療など

(13)

InfoSphere InfoSphereInfoSphere

InfoSphere Streams Streams Streams Streams プラットフォームプラットフォームプラットフォームプラットフォーム

開発環境 ^{ランタイム環境} ^{ツールキット＆} アダプター

InfoSphere Streams ランタイム

SPADE Eclipse IDE

ツールキット StreamSight

アウトプット・アダプター & ソース・アダプター

Cognos Now WebSphere MQ Mashup Center

WebSphere Front Office RSS フィード

(14)

SPADE: SPADE: SPADE:

SPADE: SSSStream tream tream tream PPPProcessing rocessing rocessing Arocessing Application AApplication pplication Dpplication DDDeclarative eclarative Eeclarative eclarative EEEngine ngine ngine ngine – 高^高^高^{高パフォーマンス}^{パフォーマンス}^{パフォーマンス}^{パフォーマンス・}^・^{・ストリーム}^・^{ストリーム・}^{ストリーム}^{ストリーム}^・^・^{・コンピューティング}^{コンピューティング}^{コンピューティング}^{コンピューティングに}^に^に^に最適化^最適化^最適化^最適化

主な特徴

プログラミング言語とコンパイル基盤

高パフォーマンス・ストリーミング・システムを念頭においた設計

幅広いバリエーションのターゲット･アーキテクチャへの効率的なマッピング

主なコンポーネント

ストリーム・プロセッシング演算子および実行ディレクティブのツールキット

ストリーム関係型演算子ユーザー定義演算子

エッジ・ストリーム・アダプター

効果的な配置、パーティショニング等を行う管理機能

主な利点

ストリーム・プロセッシングに最適化デプロイメントまでの時間短縮 Streams

プロセッシング・コア

コンパイラー ^コードジェネレータオプティマイザークエリー

Stream プロセッシンググラフ

SPADE を使ってアプリケーションを記述

(15)

プログラミングプログラミングプログラミング

プログラミング・・・・モデルモデルモデルモデル

アプリケーション・プログラミング (SPADE/INQ)

使いやすい

再利用可能な演算子群

外部の静的データソースまたはストリーミング・データソースおよびシンク群

ソース・アダプター演算子リポジトリ ^{シンク・アダプター}

プラットフォーム最適化コンパイル

(16)

シンプルシンプルシンプル

シンプルななな例な例例例

[Application] アプリケーション SourceSink trace

[Typedefs] 型定義 typespace sourcesink typedef id_t Integer typedef timestamp_t Long [Program] プログラム // 仮想スキーマ定義

vstream Sensor (id : id_t, location : Double, light : Float, temperature : Float, timestamp : timestamp_t) // ソース･ストリームは source 演算子が生成 – この場合タプルは入力ファイルから投入される

stream SenSource ( schemaof(Sensor) )

:= Source( ) [ “file:///SenSource.dat” ] {}

// この中間ストリームは Aggregate 演算子がSenSourceストリームを入力に使って生成。 stream SenAggregator ( schemaof(Sensor) )

:= Aggregate( SenSource <count(100),count(1)> ) [ id . location ]

{ Any(id), Any(location), Max(light), Min(temperature), Avg(timestamp) } // この中間ストリームは functor 演算子が生成

stream SenFunctor ( id: Integer, location: Double, message: String ) := Functor( SenAggregator ) [ log(temperature,2.0)>6.0 ]

{ id, location, “Node ”+toString(id)+ “ at location ”+toString(location) } // リザルト管理は sink 演算子が実施 – この場合生成されたタプル群はソケット宛てに送られる Null := Sink( SenFunctor ) [ “udp://192.168.0.144:5500/” ] {}

Source Aggregate Functor Sink

(17)

1 7

X86 Box

X86 ブレード

Cell ブレード X86 ブ

レード FPGA

ブレード x86

ブレード

x86 ブレード

x86 ブレード x86

ブレード x86

ブレード

Streams Streams Streams

Streams ランタイムランタイムランタイムランタイム

トランスポート

処理要素コンテナ

最適化スケジューラーは演算子を処理ノードへ割り付け、資源配分を継続的に行う

市販ハードウェアで稼働 – 単一ノードから BladeCenter から高パフォーマンス・マルチラック・クラスターまで

Streams データ・ファブリック

(18)

1 8

X86 Box

X86 ブレード

Cell ブレード Blue

Gene FPGA

ブレード x86

ブレード

x86 ブレード

x86 ブレード x86

ブレード x86

ブレードトランスポート

Streams Streams Streams

Streams ランタイムランタイムランタイムランタイム

処理要素コンテナ

最適化スケジューラーは演算子を処理ノードへ割り付け、資源配分を継続的に行う

資源、ワークロード、データ量の変動に適応

特殊ハードウェアの活用も可能特殊ハードウェアの活用も可能市販ハードウェアで稼働 _{– 単一ノードから} BladeCenterから高パフォーマンス・マルチラック・クラスターまで

市販ハードウェアで稼働 _– 単一ノードから BladeCenterから高パフォーマンス・マルチラック・クラスターまで

Streams データ・ファブリック

(19)

C 拠点 B 拠点

A 拠点

分散分散分散

分散オペレーションオペレーションオペレーションオペレーション

C 拠点 B 拠点

A 拠点

(20)

Streamsight Streamsight Streamsight

Streamsight によるによるによるによる稼動中稼動中稼動中アプリケーション稼動中アプリケーションアプリケーションアプリケーションのののの管理管理管理管理

アプリケーション群の論理ビューおよび物理ビュー

(21)

並列並列並列

並列モデルモデルモデルとしてのモデルとしてのとしてのとしてのストリームストリームストリームストリーム･･プロセッシング･･プロセッシングプロセッシングプロセッシングのののの優位性優位性優位性優位性

ファーストクラスとしてのStreams

– 明示的なタスクおよびデータの並列性 – マルチコア活用の直感的手法

より効率的なリソース管理手段としての演算子およびデータソースのプロファイリング

ストアされたデータおよびライブ・データを通じた演算子の再利用

自動化、最適化された構築手順 (SPADE, INQ)

ソースアダプター

Sink アダプター演算子リポジトリ

自動化、最適化された

デプロイおよび管理（スケジューラー）

(22)

INQ

DGM SODA

SPADE から^から^{からランタイム}^から^{ランタイム}^{ランタイムで}^{ランタイム}^で^で^で実行^{実行されるまでの}^実行^実行^{されるまでの}^{されるまでの}^{されるまでの流}^流^流^流れ^れ^れ^れ

JDL (Job Description Language)

PE bin PE bin PE bin PE bin PE

bin SPADE

JMN (Job Manager) SPADE Compiler & Optimizer Simulink

StreamSQL

System S IDE

SPC (Stream Processing Core)

言語言語言語言語ののののコンバーターコンバーターコンバーターコンバーター

SPADE プログラムから、System Sへのジョブサブミッションファイル、設定ファイル、シェルスクリプト、デバッグ用ファイルなどが生成されます

(23)

StreamStudio の^の^の^{のプログラミング}^{プログラミング}^{プログラミング}^{プログラミング環境}^環境^環境^環境

シンタックスハイライトシンタックスハイライトシンタックスハイライトシンタックスハイライト

関数関数

関数関数のののの補完補完補完補完

グラフグラフグラフ

グラフをををを視覚的視覚的視覚的視覚的ににに表示に表示表示表示

(24)

ストリームアプリケーションストリームアプリケーションストリームアプリケーション

ストリームアプリケーション用用用用デバッガデバッガデバッガデバッガ：：：： Stream Debugger

オペレータオペレータオペレータ

オペレータののの実行前の実行前実行前実行前でででブレークポイントでブレークポイントブレークポイントブレークポイント入力入力入力

入力タプルタプルタプル、タプル、、出力、出力出力タプル出力タプルタプルタプルのの検査のの検査検査・検査・・修正・修正修正修正などなどなどなど

(25)

早期アクセス・プログラム (Early Access Program) 3大陸10拠点にて実施

2009年第二四半期に20～30拠点にて追加実施予定

InfoSphere Streams として2009年第二四半期に Early Access Program を発表

主な予定日

2009年第二四半期: – ^{限定的なリリース} 2010年前半:

– ^製品出荷

劇的なボリューム

劇的な分析劇的な

スピード

(26)

2009/10/23

実証実験

実証実験のごのごのご紹介のご紹介紹介紹介

(27)

Streams を活用した実証実験の事例

アルゴリズムトレーディングアルゴリズムトレーディングアルゴリズムトレーディングアルゴリズムトレーディング

公共公共公共

公共：：：：広域環境広域環境広域環境広域環境モニタリングモニタリングモニタリングモニタリング

製造業製造業製造業製造業

電波天文学電波天文学電波天文学電波天文学

半導体製造工程におけるリアルタイム異常検知によるテストプロセスの品質向上

リアルタイムの市場データを用いた金融取引への活用

• 河川、河口の水質管理

• 数千の低コストのアンテナを用いた電波宇宙天文学の研究

• セキュリティ^{セキュリティ}^{セキュリティ}^{セキュリティ分野}^分野分野：サイバーセキュリティ、災害復旧^分野

World’s largest and first fully digital radio

observatory for astrophysics, space and earth sciences, and radio research

毎秒5百万イベントマイクロ秒のレイテンシ

毎秒2500イベント 10ミリ秒のレイテンシ

毎秒150万イベント

Semiconductor Solutions

(28)

金融：次世代アルゴリズムトレーディング

超低レイテンシ、高データレート、データ形式の多様化などストリームコンピューティングの条件をすべて満たすフラグシップアプリケーション毎秒 160万イベントを処理

ミリ秒単位でトレードチャンスを検知し、実行

2010 年までには、160万イベントから 500万イベントに増加

マーケットデータの他に、企業のフィード、ニュース、音声、ビデオなどを用いた複合的な判断を可能に

次世代次世代

次世代次世代のののアルゴリズムトレーディングのアルゴリズムトレーディングアルゴリズムトレーディングのアルゴリズムトレーディングののの基盤

基盤基盤

基盤をををを Streams 上^上^上に^上^に^に構築^に^構築^構築^構築

(29)

マイクロチップ製造の製造監視・制御への応用

チャレンジチャレンジチャレンジチャレンジ

半導体の製造工程の品質を管理する自動製造監視・制御への応用

リアルタイムのデータ統計分析が必要 100以上の製造工程からなり、それぞれの工程で相関のあるステップを複数束ねた分析が必要

画像データなどを含めた非構造化データなど、様々な情報ソースからデータが到着する

• 100以上のセンサー

• リアルタイムのプロセス及びセンサーデータの分析

• 画像など、より非構造化データ分析の潜在的なニーズがある

Confidence

ITL JSQ

Test Data JPQWQA ^YPV

Prediction Repository FDC Summary Data

TSA

OTP

RAN

Yield Prediction Monitoring and Feedback

RUI

Confidence and Confusion Matrix User Defined Decision Tree PEs SOC

Link to external Very Fast Machine Learning (VFML) library (C) User Defined Decision Tree PEs Link to external Very Fast Machine Learning (VFML) library (C)

Incremental Learning through Delayed Feedback of Ground Truth

Online prediction using incrementally learnt model

. . . Event

Sensor Defect Tool log

Alarm

Trace data

技術的技術的技術的

技術的なななチャレンジなチャレンジチャレンジチャレンジ

ソリューションソリューションソリューションソリューション

(30)

次世代次世代次世代

次世代のののの河川河川河川・河川・・・水質管理水質管理：水質管理水質管理：： Beacon Institute と： ^と^とと IBM との^との^との協業^との^協業^協業^協業

Beacon Institute との^との^との^との協業^協業^協業^{協業により}^により^により、^により^、^、^、ニューヨーク

ニューヨークニューヨーク

ニューヨークののハドソンののハドソンハドソンハドソン河河河河のの生態系のの生態系生態系を生態系をををモモモモニタリング

ニタリングニタリング

ニタリングしししし、、、、常時常時常時常時そのその状況そのその状況状況状況をををを分析分析分析・分析・・・評価

評価評価

評価できるこれまでにないできるこれまでにないできるこれまでにないシステムできるこれまでにないシステムシステムシステムをををを開発

開発開発開発

315マイル^マイル^マイルマイル(507km)にわたる^にわたる^にわたる^{にわたる全域}^全域^全域^全域の^の^の^のデータデータ

データデータをををを収集収集収集収集

Streams を^を^を^を用^用^用いて^用^いて^いて^いて、^{、センサー}^、^、^センサー^{センサーから}^センサー^から^から^からデー^デー^デー^データ

タタ

タををを集を集集め集めめめ、、水、、水水水のののの温度温度、温度温度、、塩分、塩分塩分、塩分、濁、、濁濁濁りりりり具具具具合

合合

合などをなどをなどをなどを調査調査調査調査さらに

さらにさらに

さらに、、、、ハドソンハドソンハドソンハドソン河河河に河にに棲息に棲息する棲息棲息するする種する種種種々々々の々ののの魚類

魚類魚類

魚類そのそのその他その他他他のののの生物生物の生物生物のの状況の状況状況も状況ももも調査調査調査調査

(31)

サイバーセキュリティ

様々なセンサー（ビデオ、電話通話記録、音声）を複合的に利用した監視システム受信するデータストリームは巨大だが、“異常値”を示すデータはほとんどないというのが特徴的 - 前処理で処理することに大いに意味がある

ボットネットボットネットボットネット

ボットネット検出検出検出検出監視監視監視

監視システムシステムシステムシステム

(32)

ヘルスケアヘルスケアヘルスケア

ヘルスケア分野分野分野：分野：：：患者患者患者患者のののリアルタイムモニタリングのリアルタイムモニタリングリアルタイムモニタリングリアルタイムモニタリング

Contextual Data Sources Biometric Sensor Data

+

狭心症狭心症狭心症狭心症アラートアラートアラートアラート

アラートアラートアラートアラート alert

alert

SPE

QRS

FA

RR

SP

BP

PS

AR

AP PT

SPA

BPA

alert

PSA EP

GL GLA

アラートアラートアラートアラート

WT WTA WB

不整脈不整脈不整脈不整脈

Well-Being

ユーザーインタフェースユーザーインタフェースユーザーインタフェースユーザーインタフェース

遠隔の患者のモニタリング

医学的に重要なイベントの自動検知し、医者や看護婦に配信 Provenance (問題の出所を追跡) のサポート

アラートが出たときのイベントをトラックできる

医者が下した医療処置の（定量的な）証明が可能になる監視する患者の数の増加に応じたスケールアウトが可能に

IBM と^と^とと NPO 法人^法人法人 Sick ^法人 Kids との^との^との協業^との^協業^協業^協業

(33)

2009/10/23

SPADE:

ストリームアプリケーション

ストリームアプリケーション開発言語開発言語開発言語開発言語

(34)

SPADE : ストリームアプリケーションストリームアプリケーションストリームアプリケーション記述言語ストリームアプリケーション記述言語記述言語記述言語

ストリームアプリケーションを記述するために特化した言語

多様な情報ソースのアダプタ、ウィンドウ定義、並列機構、耐故障性などを抽象度の高い言語で記述

オペレーター（ストリームデータの処理ロジック）を一つの単位とするオペレータの入出力データをタプルと呼ぶ

複数のオペレータのデータフローを定義

[Program]

vstream MySchema(symbol : String, tradedate : String, closingprice : Double, volume : Integer)

vstream aggregatedData(symbol: String, avgPrice : Double_）））） stream myODBCstream(schemaFor(MySchema))

:= Source()[““““stcp://sensorserver.ibm.com:12345””””, csvFormat, noDelays] stream StockMovingAverage (schemaFor(aggregatedData))

:= Aggregate(myODBCstream <count(20), count(1), pergroup>) [symbol]

{Any(symbol),Avg(closingprice)}

Nil := DbAppend(StockMovingAverage)[connection:"DB2Person"; access:"StockSink"]{}

(35)

オペレータととととストリームストリームストリームからストリームから構成からから構成構成構成されるされるされるされるデータフローネットワークデータフローネットワークデータフローネットワークデータフローネットワーク

Source 1

Sink 2

Functor 2 Aggregator

Functor 1

Sink 3 Functor 4 Functor 5

Sink 1 Functor 3

input file, socket,

...

output file, socket,

... output file,

socket, ... output file,

socket, ... Split

Join

シンクソースオペレータ

ソース

オペレータオペレータ

stream

stream strea

m

str^ea m

st^re a

m

str_e m a

シンク Source

入力入力入力入力ポーポーポーポー

トト

トト ^出力^出力^出力^出力ポー_ト_ト_ト_ト^ポー^ポー^ポー

(36)

簡単簡単簡単

簡単なななな例例例例

[Application] SourceSinktrace [Typedefs]

typespacesourcesink typedef id_t Integer

typedef timestamp_t Long [Program]

// virtual schema declaration

vstreamSensor (id : id_t, location : Double, light : Float, temperature : Float, timestamp : timestamp_t) // a source stream is generated by a Source operator – in this case tuples come from an input file

stream SenSource( schemaof(Sensor) ) := Source( ) [ “file:///SenSource.dat” ] {}

// this intermediate stream is produced by an Aggregate operator, using the SenSource stream as input stream SenAggregator( schemaof(Sensor) )

:= Aggregate( SenSource<count(100),count(1)>) [ id . location ]

{ Any(id), Any(location), Max(light), Min(temperature), Avg(timestamp) } // this intermediate stream is produced by a functor operator

stream SenFunctor( id: Integer, location: Double, message:String ) := Functor( SenAggregator) [ log(temperature,2.0)>6.0 ]

{ id, location, “Node ”+toString(id)+ “ at location ”+toString(location) }

// result management is done by a sink operator – in this case produced tuples are sent to a socket Null := Sink( SenFunctor ) [ “udp://192.168.0.144:5500/” ] {}

Source Aggregate Functor Sink

(37)

オペレータ・・・・ストリームストリームストリームストリームのの定義のの定義定義定義ののの仕方の仕方仕方仕方

stream myStream (Integer A, Integer B)

:= Functor (inputStream) [light >100.0 ] {sensorID, light} オペレータ

オペレータオペレータ

オペレータ名名名名オペレータオペレータ条件オペレータオペレータ条件条件条件 ^出力^{出力タプル}^出力^出力^タプル^タプル^タプル出力出力

出力出力タプルタプルタプルタプルののののスキーマスキーマスキーマスキーマ

入力入力

入力入力ストリームストリームストリームストリームストリーム

ストリームストリーム

ストリームのののの定義定義定義定義

オペレータ

オペレータオペレータ

オペレータ

オペレータパラメータオペレータパラメータオペレータパラメータオペレータパラメータ入力

入力入力

入力ストリームストリームストリームストリーム (スキーマ^スキーマ^{スキーマ定義}^スキーマ^定義^定義^定義）^）^）^）

出力出力

出力出力ストリームストリームストリームストリーム (スキーマ^スキーマ^スキーマ^{スキーマ定義}^定義^定義）^定義^）^）^）

(38)

SPADE の^の^の^の特徴^特徴^特徴特徴: 外部^外部^外部の^外部^の^の情報^の^情報^情報^{情報ソース}^{ソースとの}^ソース^ソース^との^との接続^との^接続^接続^接続を^を実現^を^を^実現^実現^実現する^する^する^{するソー}^ソー^ソー^ソースオペレータ

スオペレータスオペレータ

スオペレータ、、、、シンクオペレータシンクオペレータシンクオペレータシンクオペレータ

ネットワークソケット: TCP / UDP (クライアントモード、サーバーモード）ファイル、データベース、WFO (WebSphere Front Office) など

様々な形式のサポート： CVS フォーマット、バイナリフォーマット^{（パーサーの指定）}

input adapter (non-SPADE

program)

source operator

output adapter (non-SPADE

program) sink

operator

SPADE Program

TCP socket _{TCP socket}

stream myStream(schemaFor(MySchema))

:= Source()[“^““stcp://sensorserver.ibm.com:12345”^“ ^””, csvFormat, noDelays]^”

PDF 本研究室志望者へ 鈴村研究室 大規模データ処理・ストリームコンピューティング

ストリームコンピューティング

ストリームコンピューティング

ストリームコンピューティング

ストリームコンピューティング

- データ データ爆発時代 データ データ 爆発時代 爆発時代 爆発時代の の の の新 新たな 新 新 たな たな計算 たな 計算 計算パラダイム 計算 パラダイム パラダイム パラダイム

即応性・意思決定までの時間の最小化

様々な情報ソースを利用した高度な分析

新

新

新

新たな たな たな たな計算 計算 計算 計算パラダイム パラダイム パラダイム パラダイム

の

の の

の必要性 必要性 必要性 必要性

ストリームコンピューティング

ストリームコンピューティング

ストリームコンピューティング

ストリームコンピューティング処理 処理 処理システム 処理 システム システム システム

InfoSphere Streams

実証実験

実証実験

実証実験

実証実験のご のご のご紹介 のご 紹介 紹介 紹介

Streams を活用した実証実験の事例

マイクロチップ製造の製造監視・制御への応用

SPADE:

ストリームアプリケーション

ストリームアプリケーション

ストリームアプリケーション

ストリームアプリケーション開発言語 開発言語 開発言語 開発言語

オペレータ

オペレータ オペレータ

オペレータ

SPADE Program

PDF 本研究室志望者へ鈴村研究室大規模データ処理・ストリームコンピューティング

- データ ^{データ爆発時代} ^データ ^データ ^爆発時代 ^爆発時代 ^{爆発時代の} ^の ^の ^の新 ^新たな ^新 ^新 ^たな ^たな計算 ^たな ^計算 ^{計算パラダイム} ^計算 ^{パラダイム} ^{パラダイム} ^{パラダイム}

新たなたなたなたな計算計算計算計算パラダイムパラダイムパラダイムパラダイム

のの

の必要性必要性必要性必要性

ストリームコンピューティング処理処理処理システム処理システムシステムシステム

実証実験のごのごのご紹介のご紹介紹介紹介

ストリームアプリケーション開発言語開発言語開発言語開発言語

オペレータオペレータ