(57)【要約】      （修正有）【課題】リアルタイムに到来するストリーム・データを処理するためのシステムにおいて、クエリに対する応答レポート生成処理を高速化する。【解決手段】リアルタイム・ストリーム・データを処理するためのシステムにおいて、時間解像度を含めたクエリ定義を事前に設定する際に、クエリ定義に従いプリアグリゲーション・テーブルを生成し、アグリゲーション結果を時間解像度ごとに更新(追加・削除)し、作成したプリアグリゲーション・テーブルを利用して、ある時間範囲に生成されたデータに対してクエリを行うことを可能にする。【選択図】図２
2014157510.tif 000002
【特許請求の範囲】
【請求項１】
コンピュータの処理により、リアルタイムに生成されるデータをクエリ処理するためのシステムにおいて、
指定された時間解像度と保存期間の値を用いて、現在の時間から該時間解像度の時間が経過するまでの前記データからクエリ処理結果を生成するための中間結果を保存することと、前記保存期間の経過後は前記保存した中間結果を削除する中間結果処理ステップと、
前記中間結果処理ステップを、指定された時間解像度の時間ごとに動作させるようにスケジュールするステップと、
クエリを受信することに応答して、前記中間結果からクエリ結果を生成するステップを有する、
方法。
【請求項２】
前記中間結果処理ステップは、MapReduceを実装するシステムによって処理される、請求項１に記載の方法。
【請求項３】
前記中間結果処理ステップを実行する、前記MapReduceを実装するシステム上で動作するプログラムを、前記コンピュータの処理により、前記指定された時間解像度と保存期間の値から生成するステップをさらに有する、請求項２に記載の方法。
【請求項４】
前記コンピュータの処理により、前記スケジュールするステップを実行するプログラムを、前記指定された時間解像度と保存期間の値から生成するステップをさらに有する、請求項１に記載の方法。
【請求項５】
前記中間結果は、時間を含む値をキーとするフィールドを有する、請求項１に記載の方法。
【請求項６】
前記クエリが時間の範囲を含み、前記クエリ結果を生成するステップは、前記時間を含む値をキーが該時間の範囲にあるデータを集計するステップを含む、請求項５に記載の方法。
【請求項７】
コンピュータの処理により、リアルタイムに生成されるデータをクエリ処理するためのシステムにおいて、
前記コンピュータに、
指定された時間解像度と保存期間の値を用いて、現在の時間から該時間解像度の時間が経過するまでの前記データからクエリ処理結果を生成するための中間結果を保存することと、前記保存期間の経過後は前記保存した中間結果を削除する中間結果処理ステップと、
前記中間結果処理ステップを、指定された時間解像度の時間ごとに動作させるようにスケジュールするステップと、
クエリを受信することに応答して、前記中間結果からクエリ結果を生成するステップを実行させる、
プログラム。
【請求項８】
前記中間結果処理ステップは、MapReduceを実装するシステムによって処理される、請求項７に記載のプログラム。
【請求項９】
前記中間結果処理ステップを実行する、前記MapReduceを実装するシステム上で動作するプログラムを、前記コンピュータの処理により、前記指定された時間解像度と保存期間の値から生成するステップをさらに有する、請求項８に記載のプログラム。
【請求項１０】
前記コンピュータの処理により、前記スケジュールするステップを実行するプログラムを、前記指定された時間解像度と保存期間の値から生成するステップをさらに有する、請求項７に記載のプログラム。
【請求項１１】
前記中間結果は、時間を含む値をキーとするフィールドを有する、請求項７に記載のプログラム。
【請求項１２】
前記クエリが時間の範囲を含み、前記クエリ結果を生成するステップは、前記時間を含む値をキーが該時間の範囲にあるデータを集計するステップを含む、請求項１１に記載のプログラム。
【請求項１３】
コンピュータの処理により、リアルタイムに生成されるデータをクエリ処理するためのシステムにおいて、
指定された時間解像度と保存期間の値を用いて、現在の時間から該時間解像度の時間が経過するまでの前記データからクエリ処理結果を生成するための中間結果を保存することと、前記保存期間の経過後は前記保存した中間結果を削除する中間結果処理手段と、
前記中間結果処理ステップを、指定された時間解像度の時間ごとに動作させるようにスケジュールする手段と、
クエリを受信することに応答して、前記中間結果からクエリ結果を生成する手段を有する、
システム。
【請求項１４】
前記中間結果処理手段は、MapReduceを実装するシステムによって処理される、請求項１３に記載のシステム。
【請求項１５】
前記中間結果処理手段を実行する、前記MapReduceを実装するシステム上で動作するプログラムを、前記コンピュータの処理により、前記指定された時間解像度と保存期間の値から生成する手段をさらに有する、請求項１４に記載のシステム。
【請求項１６】
前記スケジュールするステップを実行するプログラムを、前記指定された時間解像度と保存期間の値から生成する手段をさらに有する、請求項１３に記載のシステム。
【請求項１７】
前記中間結果は、時間を含む値をキーとするフィールドを有する、請求項１３に記載のシステム。
【請求項１８】
前記クエリが時間の範囲を含み、前記クエリ結果を生成する手段は、前記時間を含む値をキーが該時間の範囲にあるデータを集計する機能を実行する、請求項１７に記載のシステム。
【発明の詳細な説明】
【技術分野】
【０００１】
この発明は、リアルタイムに到来するストリーム・データを処理するための技法に関し、より詳しくは、リアルタイムに到来するストリーム・データのクエリ結果を得るための技法に関する。
【背景技術】
【０００２】
例えば、道路交通シミュレーションにおいて、時々刻々と到来する車の速度やエリアなどの多数のデータに基づき、クエリ結果を迅速に得たいという要望がある。
【０００３】
リアルタイム・データに対するクエリの例として、SQLライクな疑似コードで記述すると以下のようなものがある。
SELECT MAX(speed) from "TABLE"
TIME
FROM time.now()
TO   time.now() - time.MIN(10)
GroupBy AREA
この式で、"TABLE"とは、時々刻々とハードディスクに蓄積されたリアルタイム・データのテーブルのことである。この式は、現時点から10分前までの間のデータで、最高速度を求めるものであるが、クエリが投入された時点で、全データをスキャンする、ディスクアクセスなどのコストが生じるので、クエリに対する応答レポート生成に多くの時間が必要になる。
【０００４】
もし計算の範囲(時間窓)が事前に決まっているなら、プリアグリゲーション(pre-aggregation)テーブルが事前集計でき、それによってクエリ結果生成の時間を短縮できるのだが、一般的には道路交通シミュレーションによって蓄積されるデータの場合は刻一刻と計算の範囲が変化して一意に決められないため、プリアグリゲーション・テーブルを用意できない。
【０００５】
特開２００６−３３８４３２号公報は、ストリームデータの一部、もしくは全部複製し、不揮発性の記憶媒体にアーカイブし、リアルタイムデータとアーカイブデータをシームレスに利用できる機構と、複数のストリームデータ処理システムを連携させ、クエリ処理の性能を向上させる機構を開示する。
【０００６】
特開２０１０−１０８０７３号公報は、時間ウィンドウを含むクエリのメモリ使用量を一定に保ち、かつ全ての入力データを考慮した正確な計算処理を実現するストリームデータを提供するために、ストリームデータ処理サーバが、クエリ時間解像度変更部で、時間ウィンドウをより小さい幅のサブウィンドウに区切り、クエリ処理エンジンがストリームデータ受付時にサブウィンドウ単位の集計処理を実行して集約タプルを生成し、時間ウィンドウを含むクエリの計算結果を、この集約タプルに対する集計処理によって計算する技法を開示する。
【０００７】
特開２０１０−２１７９６８号公報は、時刻情報が付与されたデータを時系列順に受信し、あらかじめ登録された複数のクエリによって処理する計算機システムで、受信したデータを処理する現用系の計算機と、障害発生時に現用系の計算機の替わりに受信したデータを処理する待機系の計算機とが含まれ、現用系の計算機は、あらかじめ定義された順序で各クエリによって受信したデータを処理し、各クエリによる受信したデータの処理結果を中間データとして、所定のタイミングで、クエリごとに待機系の計算機に送信し、待機系の計算機は、現用系の計算機に障害が発生した場合には、受信したデータ及び中間データを処理することによってデータを復元することを開示する。
【０００８】
特開２０１１−５９９６７号公報は、時系列に時刻情報が付与されたストリームデータを生成し、その生成したストリームデータについて、登録されたクエリに基づいてストリームデータ処理を行う計算機システムのストリームデータの生成方法であって、計算機システムに、クエリと、ストリームデータを構成する複数種類の構成要素を表すストリーム定義とからそのクエリに対応する前記構成要素を示すクエリ情報を記憶する記憶部と、ストリームデータを生成・送信するデータ生成部と、データ生成部から送信されたストリームデータの処理を行うストリームデータ処理部と、を設け、データ生成部が、そのクエリ情報に基づいて、ストリームデータ処理部に送信するストリームデータからより少ないストリームデータを生成することを開示する。
【０００９】
しかし、これらの従来技術は、クエリに対する応答レポート生成の時間を短縮する問題に取り組むものではない。
【先行技術文献】
【特許文献】
【００１０】
【特許文献１】特開２００６−３３８４３２号公報
【特許文献２】特開２０１０−１０８０７３号公報
【特許文献３】特開２０１０−２１７９６８号公報
【特許文献４】特開２０１１−５９９６７号公報
【発明の概要】
【発明が解決しようとする課題】
【００１１】
この発明の目的は、リアルタイムに到来するストリーム・データを処理するためのシステムにおいて、クエリに対する応答レポート生成処理を高速化することにある。
【課題を解決するための手段】
【００１２】
この発明によれば、時間に関する制約を前提とした、リアルタイム・ストリーム・データを処理するためのシステムにおいて、時間解像度を含めたクエリ定義を事前に設定する際に、クエリ定義に従いプリアグリゲーション・テーブルを生成し、アグリゲーション結果を時間解像度ごとに更新(追加・削除)し、作成したプリアグリゲーション・テーブルを利用して、ある時間範囲に生成されたデータに対してクエリを行うことを可能ならしめるようにシステムを構成することによって、上記課題が解決される。
【００１３】
この発明に従うシステムは、予めユーザが作成した、時間解像度と、時間ウインドウ情報を含むクエリ定義に従い、プリアグリゲーション・テーブル要求を解析して、プリアグリゲーション・テーブル更新スケジューラを登録するとともに、プリアグリゲーション・テーブル更新プログラムを自動生成する。
【００１４】
すなわち、プリアグリゲーション・テーブル更新スケジューラは、時間解像度毎に、時間解像度範囲のデータ抽出フィルタを生成し、プリアグリゲーション・テーブル更新プログラムは、センサなどからリアルタイムに到来するストリーム・データに時間解像度範囲のデータ抽出フィルタを適用することによって、集計結果のみを含むKey, Valueの値を生成し、それをプリアグリゲーション・テーブルに追加する。
【００１５】
この発明に従うシステムは、クライアント・システムからクエリを受信することに応答して、そのクエリ内の時間範囲のデータをフィルタするキーを用いたリダクション・プロセスをプリアグリゲーション・テーブルに適用することで、高速にクエリに対応するレポートを生成する。
【００１６】
この発明に従うシステムは、プリアグリゲーション・テーブル更新プログラムによって、時間ウインドウの時間よりも古いデータをプリアグリゲーション・テーブルの末尾から削除し、これによって、プリアグリゲーション・テーブルのデータが増大し続けることを防ぐ。
【発明の効果】
【００１７】
以上のように、この発明によれば、集計されたエントリをもち、サイズが時間ウインドウの範囲に収まるようにエントリを削除していくようにしたプリアグリゲーション・テーブルを生成して、このプリアグリゲーション・テーブルに基づきクエリに対するレポートを生成するようにすることによって、集計されたエントリを利用して集計処理を合理化するとともに、限定されたサイズのプリアグリゲーション・テーブルをスキャンすればいいので、短時間でクエリに対応するレポートを生成できるという効果を奏する。
【図面の簡単な説明】
【００１８】
【図１】この発明を実施するためのハードウェア構成の概要を示す図である。
【図２】この発明を実施するための機能構成の概要を示す図である。
【図３】プリアグリケーション・テーブルを生成するための機能構成を示す図である。
【図４】プリアグリケーション・テーブルを生成するためのフローチャートを示す図である。
【図５】プリアグリケーション・テーブルを生成するためのスキーマと、プリアグリケーション・テーブルのエントリの例を示す図である。
【図６】プリアグリケーション・テーブルにおいて、時間窓から外れたエントリを削除する様子を示す図である。
【図７】クエリ処理の機能構成を示す図である。
【図８】クエリ処理のフローチャートを示す図である。
【図９】クエリ処理の例を図式的に示す図である。
【発明を実施するための形態】
【００１９】
以下、図面に従って、本発明の実施例を説明する。これらの実施例は、本発明の好適な態様を説明するためのものであり、発明の範囲をここで示すものに限定する意図はないことを理解されたい。また、以下の図を通して、特に断わらない限り、同一符号は、同一の対象を指すものとする。
【００２０】
図１は、本発明を実施するためのシステムの一例の概要構成を示す図である。図１において、車両１０４、１０６は、パケット通信網１０２を通じて、サーバ１１０ａに、プローブカー・データを送信する。尚、実際は、何千台あるいは何万台レベルの車両からプローブカー・データを受信するが、ここでは例示的に２台だけ表示されていることに留意されたい。
【００２１】
プローブカーデータを利用した交通データ収集技法は公知であり、例えば、特開２００４−１１０４５８号公報、特開２００４−１５６９８２号公報、特開２００７−１９３７０５号公報、特開２００４−２４１９８７号公報などを参照されたい。このシステムは、そのような周知の技法を使用して、車両の速度、位置、加速度などのデータを時々刻々し受け取る。
【００２２】
サーバ１１０ａは、ＬＡＮ、ＷＡＮ、ＦＨＨＴなどのどれかの接続構成である、ネットワーク１０８に接続され、ネットワーク１０８にはさらに、サーバ１１０ｂ、１１０ｃ、・・・、１１０ｍ、サーバ１１２、１１６、１１８ａ、・・・、１１８ｋが接続されている。これらのサーバとして、これには限定されないが、インターナョナル・ビジネス・マシーンズ・コーポレーションから購入可能な、IBM(R) System X、System i、System pなどの機種のサーバを使うことができる。これらのサーバで使用可能なオペレーティング・システムとして、AIX（商標）、UNIX(商標)、Linux(商標)、Windows(商標)2003 Serverなどがあるが、この実施例では、Linux(商標)を導入するものとする。
【００２３】
サーバ１１０ａ、サーバ１１０ｂ、１１０ｃ、・・・、１１０ｍには、IBM(R) InfoSphere Streamsが導入され、サーバ１１０ａ、サーバ１１０ｂ、１１０ｃ、・・・、１１０ｍは、ストリーム・サーバを構成する。
【００２４】
サーバ１１６にはHadoopが導入され、サーバ１１６において、Hadoopのプロジェクトの１つであるHbaseを使って、カラムナー・テーブル(columnar table)１１６ａが実現される。
【００２５】
サーバ１１２にもHadoopが導入され、サーバ１１２において、Hadoopのプロジェクトの１つであるHiveを使って、クエリ・サーバ１１２ａが実現される。サーバ１１２には、クライアント・コンピュータ１１４が接続され、クライアント・コンピュータ１１４には、ＧＵＩの操作により、後述する時間粒度(time resolution)と、時間窓(time window)の情報を含むクエリ定義ファイルを生成するためのプログラムが導入されている。
【００２６】
サーバ１１８ａ、・・・、１１８ｋは、複数台の各々に、JobTracker、NameNode、TaskTracker、DataNodeの役割を割り当てることにより、MapReduceを構成する。あるいは、単一のサーバで、これらの全ての役割を担わせることもできる。
【００２７】
次に、図２を参照して、本発明のシステムの機能構成の概要について説明する。図２において、車両１０４は、センサ１０４ａを備え、アラータ(alerter)１０４ｂを介して、プローブカー・データを、サーバ１１０ａ、サーバ１１０ｂ、１１０ｃ、・・・、１１０ｍから構成されるストリーム・システム２００に送信する。ストリーム・システム２００は、接続マネジャ２０２、アラータ(alerter)２０４、マテリアライズド・ビュー用ＥＴＬ２０６をもち、接続マネジャ２０２は、受信したプローブカー・データを、アラータ２０４と、マテリアライズド・ビュー用ＥＴＬ２０６に送る。
【００２８】
サーバ１１６は、マテリアライズド・ビュー用ＥＴＬ２０６から受け取ったデータに基づき、主記憶上に、Hbaseにより、カラムナー(columnar)・テーブル１１６ａを作成する。
【００２９】
サーバ１１６は、サーバ１１８ａ、・・・、１１８ｋのうちの所定のサーバ群により構成されたMapReduce２０８により、プリアグリゲーション・テーブル（PAT）１１６ｂを主記憶上に生成する。サーバ１１６はまた、サーバ１１８ａ、・・・、１１８ｋのうちの所定のサーバ群により構成されたMapReduce２１０により、クエリ・サーバ112aから受け取ったクエリに従い、プリアグリゲーション・テーブル１１６ｂからデータを抽出してレポートを作成する処理を行う。
【００３０】
MapReduce２０８上で実行されるプログラムは、クライアント・コンピュータ１１４でユーザにより作成される、クエリ定義ファイルの内容に基づき生成される。
【００３１】
クエリアプリ定義ファイルの内容の例は、以下のとおりである。
Application: {
AggFunction :{avg,sum,cnt,max,min},
GroupBy: {
"Area" : Hash1(gps), // 47 types
"Age"  : Hash2(age)  // 6 types
},
Measures: {
"speed"  : float, // per hour
"LineChange" : int, // per intr
"quickBrake" : int, // per intr
}
resolution : 60      // 1 min
window : 3600   // 60 mins
}
【００３２】
クエリアプリ定義ファイルの内容の例において、AggFunctionは使用される集計関数をリストする。avgは平均、sumは合計、cntはカウント、maxは最大、minは最小をあらわす。
【００３３】
GroupByは、どのキーで集約するかを示す。ここでは、Area(区域)と、Age(年代)で集約すると指定されている。ここで、例えば、Hash1(gps)とあるのは、gpsの値を入力して、都道府県名を返す関数であり、次のようなコードである。
Hash1(gps)  {
if gps.lt > 10 and gps.lg < 100:
return "tokyo"
elif gps.lt > 12 and gps.lg > 10:
return "osaka"
...
}
【００３４】
Hash2(age)も、年代を層別する関数として適宜定義される。
【００３５】
プリアグリゲーション・テーブルで、レコードのキーは、時間+集約キーで生成される。一例では、キーは、2012/01/01_10:00:00_tokyo_10のようになる。
【００３６】
resolutionとあるのは、プリアグリゲーション・テーブルでデータを集約する時間粒度であり、windowとあるのは、その時間が経過したら、プリアグリゲーション・テーブルからレコードが消されてしまう時間の幅である。
【００３７】
図３は、そのようにして作成されたクエリアプリ定義ファイルに基づき、プリアグリゲーション・テーブル(PAT)を生成し、PATスケジューラを登録し、更にはMapReduce２０８を動作させるプログラムを生成するための機能ブロックを示す図である。
【００３８】
図３において、クエリアプリ定義ファイル３０２は、クライアント・コンピュータ１１４で作成されて、例えば、クライアント・コンピュータ１１４のハードディスク上または、主記憶上に保持されているとする。
【００３９】
クエリサービスを実行するサーバ１１２のハードディスクには、クエリアプリ定義ファイル３０２を要求解析するPAT要求解析モジュール３０４と、PAT要求解析モジュール３０４の解析結果に基づきプリアグリゲーション・テーブル１１６ｂを作成するPAT作成モジュール３０６と、PAT要求解析モジュール３０４の解析結果に基づきPATスケジューラ３１２にスケジュールを登録するPATスケジューラ登録モジュール３０８と、PAT要求解析モジュール３０４の解析結果に基づきMapReduce２０８を動作させるためのMAPPER/REDUCERプログラムを生成するためのPAT更新プログラム作成モジュール３１０が保存され、クライアント・コンピュータ１１４などからの操作に応答して、適宜呼び出されて主記憶にロードされ実行される。
【００４０】
尚、この実施例では、クエリサービスを実行するサーバ１１２を、PAT要求解析などを行わせるサーバとして兼用しているが、別のサーバを立てて、PAT要求解析、PAT作成、PATスケジューラ登録、及びPAT更新プログラム作成などの処理を実行させるようにしてもよい。
【００４１】
サーバ１１６においては、ストリーム・システム２００から受領したデータに従い主記憶上に形成したカラムナー・テーブル１１６ａに基づき、MapReduce２０８が、PATスケジューラ３１２のスケジューリングに従い動作して、集約したデータ３１４を作成し、プリアグリゲーション・テーブル１１６ｂに追加する。
【００４２】
次に図４のフローチャートを参照して、プリアグリゲーション・テーブル(PAT)の生成に関連する処理について説明する。図４のステップ４０２で、PAT要求解析モジュール３０４が、クエリアプリ定義ファイル３０２を読み込んで、クエリ定義Dに対するPAT生成処理を開始する。
【００４３】
ステップ４０４で、PAT要求解析モジュール３０４は、PATが存在するかどうか判断し、もしそうでないなら、ステップ４０６で、PAT定義と合致する(K,V)のスキーマを定義してPAT作成モジュール３０６を呼び出し、データを保存するPATを生成する。ここでPAT定義とは、時間解像度、(任意個の)グループ化指示列、対象データ列、集約関数の組み合わせをKEYとするとき、グループ化後の対象データ列に対して集約関数を適用結果のスカラー値をVALUEとして参照可能なKey Value Store型のテーブルである。図５は、スキーマに基づくPAT作成モジュール３０６の処理の例を示す図である。すなわち、PAT作成モジュール３０６は、ステップ５０２で新規テーブルを作成し、クエリアプリ定義ファイル３０２に基づき、ステップ５０４でキーを生成し、ステップ５０６でカラム・ファミリ(CF)を定義し、プリアグリゲーション・テーブル１１６ｂを生成する。
【００４４】
キーは、時間+集約フィールドであらわされる。この例では、Area+Ageが集約フィールドである。
【００４５】
この段階では、空のプリアグリゲーション・テーブル１１６ｂだけであるが、参考のために、具体的なエントリの例も示す。この例では、時間解像度(resolution) = 60秒なので、先頭のエントリ2012/01/01_10:00:00_tokyo_10 {“avg”:42.3, “max”:120.2, “min”:0.0, “cnt”:10, “sum”:423}...は、60秒間の集約データである。
【００４６】
図４のフローチャートに戻って、ステップ４０８では、PAT要求解析モジュール３０４は、クエリ定義Dに対応したPATへのデータ追加を行う集約プログラム(=P)が存在するか判断し、そうでないなら、ステップ４１０で、PAT更新プログラム作成モジュール310を呼び出して、センサーデータを入力として、Dの定義に従い集計し、集計時刻、グループ化列、対象データ列、集約関数をKEY、この集計結果をVALUEとするレコードを生成するプログラムを生成する。
【００４７】
下記に示すのは、
Application: {
AggFunction :{avg,sum,cnt,max,min},
GroupBy: {
"Area" : Hash1(gps), // 47 types
"Age"  : Hash2(age)  // 6 types
},
Measures: {
"speed"  : float, // per hour
"LineChange" : int, // per intr
"quickBrake" : int, // per intr
}
resolution : 60      // 1 min
window : 3600   // 60 mins
}
というクエリアプリ定義ファイル３０２からPAT作成モジュール３０６が生成するMap/Reduce用の集約プログラムのコードの例を示す。ここで示すのは疑似コードであるが、実際は、Java(R)、Pythonなどの既存の任意のプログラミング言語で記述することができる。
class MAPPER:
method INITIALIZE():
Set<K,V> set = new Set<K,V>()     // K=str, V=tuple
method MAP(string t, row r):
str s = String.format(“%s-%s-%s”, time, Hash1(r.gps), Hash2(r.age))   
// for CF:Speed
float sum1 = r.speed, float cnt1 = 1
float max1 = max(r.speed), float min1 = min(r.speed)
tuple t1 = new tuple(sum:sum1, cnt:cnt1, max:max1, min:min1)
set.add(“SP”, t1)   
// for CF:LineChange
float sum2 = r.lineChange, float cnt2 = 1 
float max2 = max(r.lineChange), float min2 = min(r.lineChange)
tuple t2 = new tuple(sum:sum2, cnt:cnt2, max:max2, min:min2)
set.add(“LC”, t2)   
EMIT(str s, Set set)
class REDUCER:
method INITIALIZE():
float sum1,sum2,cnt1,cnt2,max1,max2,min1,min2 = 0
method REDUCE(str s, sets[r1, r2, ..]):
for all Set set ∈ sets[r1, r2, …] do
sum1 = sum1 + set.get(“SP”).sum, sum2 = sum2 + set.get(“LC”).sum
cnt1 = cnt1 + set.get(“SP”).cnt, cnt2 = cnt2 + set.get(“LC”).cnt
max1 = set.get(“SP”).max > max1 ? Set.get(“SP”).max : max1
done
writeToIncrementalDB(s, CF=“SP”, sum1, cnt1, max1, min1)
writeToIncrementalDB(s, CF=“LC”, sum2, cnt2, max2, min2)
このようにして生成された集約プログラムのコードは、MapReduceシステム２０８を動作させる。尚、ここではresolutionとwindowの値はあらわれないが、それらの設定値は、PATスケジューラ３１２に設定される。
【００４８】
図４のフローチャートに戻って、PAT要求解析モジュール３０４は、ステップ４１２で、PATスケジューラ登録集約モジュール３０８によって、ＰＡＴスケジューラ３１２に、プログラムを時間解像度ごとに起動するようにスケジューリングする。ＰＡＴスケジューラ３１２は、実質的に、上記のMapReduce用のコードの外にあるドライバの中でハンドルすることができる。例えば:
*/1 * * * * hadoop jar PAT-Update-MR.jar
などとすると、1分ごとに更新プログラムが起動する。起動されたプログラムのドライバrでは、まず現在時刻を取得し、どこの時点に対する集約をするかを決定する。例えば、現在時刻を取得して12:01:02 だとしたとき、基準となる集約時刻を12:01にすると決める。これは実装の問題であり、自由に変更できる。この12:01という基準がプログラム実行時に動的に決定されると、ここからresolution時間分をさかのぼって、12:00:00 - 12:00:59 という時間範囲(=resolution)のデータをMapの入力とするというように一意に決定される。この範囲内のデータは、12:01という時刻に関して紐付けられるので、最終的なReduceの結果をPATに書き込む段階で、keyの一つとして利用される。
【００４９】
ステップ４１４では、集約プログラムPが、MapReduce２０８上で、現在の時刻=T1として保存し、ステップ416で、PATスケジューラ３１２が、センサーデータの入力のうち、最後に実行した時刻(=T0)と現在の時刻(=T1)を範囲にフィルタするルールをPに与えて集約プログラムPを実行し、結果を出力(=R)する。ここでいうルールとは、保存されているデータを現在の時刻から時間解像度前までのデータを入力とするようなフィルタ・ルールであり、PAT要求解析モジュール３０４の解析結果に基づき、PATスケジューラ３１２が予め作成する。そして、MapReduce２０８は、時間解像度の幅にフィルタされたデータを入力に指定し、集約プログラムをMapReduce実行する。その際、スケジューリングとは、時間解像度経過後に、上記ルール作成・MR実行を繰り返し行うようにスケジューリングすることである。
【００５０】
より詳細には、Map処理とは、時間解像度（現在の時刻）とGroupByのIDを組み合わせたkeyを、入力データからMeasureで指定したvalueを選択したTupleを生成し、Map処理の出力とすることであり、Reduce処理は、keyごとにグループされたTuple群を入力とし、集約関数をTuple内のvalueごとに分けて集約演算を行うことである。Keyに対する集約結果を新たなTupleとして、reduce処理の出力に利用する。Reduce処理はさらに、この出力をアグリゲーションテーブルに追加し、同時に現在の時刻からwindow time前のデータを削除することを含む。これは、図４のステップ４１８の、PATの先頭にRを追加、および、Dに設定されたTime Window時間よりも古いデータをPATの末尾から削除する処理に対応する。図６は、PATの末尾のデータが削除される様子を示す。
【００５１】
ステップ４２０で、PATスケジューラ３１２は、T0にT1の時刻を保存後、時間解像度時間経過するまで停止し、ステップ４２２で、クライアント・コンピュータ１１４などから、実行停止要求が来ているかどうか判断し、もしそうなら、実行を終了し、そうでないなら、ステップ４１４に戻る。
【００５２】
図７は、クライアント・コンピュータ１１４で作成したＯＬＡＰクエリ７０２に基づき、この発明に従いクエリに対するレポートを作成する処理の機能ブロックを示す図である。
【００５３】
ＯＬＡＰクエリ７０２は例えば、SQLライクな疑似コードで示すと、以下のとおりである。
SELECT AVG(speed) from PAT
TIME
FROM time.now()
TO   time.now() - time.MIN(2)
GroupBy area
これは、今から2分前までの時間の間の速度の平均値を求めるクエリである。
【００５４】
サーバ１１２にあるクエリ・サーバ１１２ａは、ＯＬＡＰクエリ７０２を、好適にはHadoopのプロジェクトの1つである、Hiveを用いて、フィルタ条件に変換する。このフィルタ条件は、ルールとして、サーバ１１６内のフィルタ７０６に加えられる。これに応答して、フィルタ７０６は、プリアグリゲーション・テーブル１１６ｂから、設定されたルールに従い、選択した範囲のデータを抽出する。
【００５５】
クエリ・サーバ１１２ａは一方で、MapReduce２１０に、MapReduceプログラム７０８を提供し、これにより、MapReduce２１０は、MapReduceプログラム７０８によって指定された集約処理を行って、結果のレポート７１０を生成する。結果のレポート７１０は、サーバ１１６の通信モジュール７１２によってクライアント・コンピュータ１１４の表示モジュール７０４に送られ、表示モジュール７０４は結果のレポート７１０を適宜、グラフ表示する。
【００５６】
次に図８のフローチャートを参照して、クエリ処理を説明する。図８のステップ８０２では、クエリ・サーバ１１２ａが、サブミットされたクエリ要求に対する結果の生成を開始する。尚、図８のフローチャートを通じて、Dはクエリアプリ定義ファイル、QはサブミットされたＯＬＡＰクエリとする。
【００５７】
ステップ８０４で、クエリ・サーバ１１２ａは、Qの結果生成に必要なデータがDで定義されているかどうかを判断し、そうでないなら、ステップ８０６で、終了処理エラーをクエリ要求元に返し、そうならステップ８０８に進む。
【００５８】
ステップ８０８で、クエリ・サーバ１１２ａは、Qで要求する時間解像度がDに定義された時間解像度よりも長いかどうか判断し、そうでないなら、ステップ８０６で、終了処理エラーをクエリ要求元に返し、そうならステップ８１０で、現在の時刻 = T0として保存する。
【００５９】
ステップ８１２で、クエリ・サーバ１１２ａは、Qの結果生成で要求される時間の範囲がT0に対して未来の時間を含むか判断し、そうでないなら、フィルタ７０６にルールを送ることにより、ステップ８１４でQに記述された時間範囲をフィルタしPATから該当するレコードを抽出し、ステップ８１６で、Qで要求される集約関数、グループ化列、対象データ列に対応する集約関数適用済みの値をこのフィルタされたレコードから抽出し、同じ集約関数をこれらにさらに適用して、計算結果を出力(=R)し、ステップ８１８でクエリ結果をクエリ要求元に返して、処理を終了する。
【００６０】
ステップ８１２に戻って、クエリ・サーバ１１２ａが、Qの結果生成で要求される時間の範囲がT0に対して未来の時間を含むと判断すると、クエリ・サーバ１１２ａは、ステップ８２０で、処理１（すなわち、ステップ８１４と、それに続くステップ８１６）を行い、T0より過去の時間のレコードをPATから抽出し、Qの結果を出力する (=R)。
【００６１】
次にクエリ・サーバ１１２ａは、ステップ８２２で、Qで要求される最も先の時間 = T2として保存し、ステップ８２４で、時間解像度時間経過するまで停止する。
【００６２】
そしてクエリ・サーバ１１２ａは、ステップ８２６で、PATの先頭を読み込み、最新のエントリのQに対応する集約結果を取得し、Rと合成して結果を出力 (=R)し、ステップ８２８で、(現在の時刻=T1) < T2かどうか判断し、もしそうなら、ステップ８２４に戻る。そして、T1がT2を過ぎると、クエリ・サーバ１１２ａは、ステップ８１８でクエリ結果をクエリ要求元に返して、処理を終了する。
【００６３】
図９は、図７及び図８で示すクエリ処理の例を図式的に示す図である。図９において、ユーザがOLAPクエリ７０２をサブミットすると、クエリ・サーバ１１２ａは、そこから、図示されているように、時間のキー・フィルタリング９０２と、カラム・ファミリの値フィルタリング９０４を生成する。そして、クエリ・サーバ１１２ａは、これらのフィルタリングからルールを生成し、フィルタ７０６に適用し、これによって選択された範囲のテーブル・データ９０６が得られる。
【００６４】
クエリ・サーバ１１２ａは、選択された範囲のテーブル・データ９０６に対して、MapReduce２１０上で、Map関数９０８と、Reduce関数９１０を呼び出して最終リダクションを行い、結果のレポート７１０を得る。このような一連の処理は、リダクション・プロセスと呼ばれる。結果のレポート７１０は、ユーザ側に返送される。
【００６５】
以上、本発明を、Hadoop(商標)フレームワークのMapReduceで実装する実施例において、プリアグリゲーション・テーブルの生成にHbaseを使用し、クエリ・サーバにHiveを使用する実施例として説明してきたが、プリアグリゲーション・テーブルの生成には、Hbase以外に、Cassandra、Memcachedなどを使用することができる。
【００６６】
また、この発明は、HadoopあるいはMapReduceに限定されず、任意のオペレーティング・システム及びハードウェアをもつコンピュータ・プラットフォーム上で実装可能である。
【００６７】
さらに、この発明は、プローブカー・データのみならず、気象データ、金融データ、その他シミュレーションなど、時間付でデータを受け取ってテーブルに集約する任意の適用例に使用することができる。
【符号の説明】
【００６８】
１１２・・・クエリ用サーバ
１１２ａ・・・クエリ・サーバ
１１４・・・クライアント・コンピュータ
１１６ａ・・・プリアグリゲーション・テーブル
２００・・・ストリーム・システム
２０８、２１０・・・MapReduceシステム
３０２・・・クエリアプリ定義ファイル
３０４・・・PAT解析要求モジュール
３０６・・・PAT作成モジュール
３０８・・・PATスケジューラ登録モジュール
３１０・・・PAT更新プログラム作成モジュール
３１２・・・PATスケジューラ
７０２・・・OLAPクエリ
【図１】
2014157510.tif 000003
【図２】
2014157510.tif 000004
【図３】
2014157510.tif 000005
【図４】
2014157510.tif 000006
【図５】
2014157510.tif 000007
【図６】
2014157510.tif 000008
【図７】
2014157510.tif 000009
【図８】
2014157510.tif 000010
【図９】
2014157510.tif 000011