(57)【特許請求の範囲】
【請求項１】
（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびその処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムであって、
前記データ蓄積装置はそれぞれ、
前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータを蓄積するキュー部と、
前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信する位置情報送信部と、
前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するキュー管理部とを備え、
前記情報管理装置は、
前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶する記憶部を備え、
前記フィルタ処理装置はそれぞれ、
前記ストリームデータの取得元となる前記データ蓄積装置を選択する選択処理部と、
前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するデータ取得部と、
前記情報管理装置の定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定し、当該フィルタ処理装置の制御を行うフィルタ実行制御部とを備え、
前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、
前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信し、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信する情報源指定データ取得部と、
前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合する選択結合部と、
前記結合したストリームデータを、外部装置へ送信する出力部とを備えることを特徴とするストリームデータ処理システム。
【請求項２】
前記データ蓄積装置はそれぞれ、
前記フィルタ処理装置から、前記第２のストリームデータの取得要求を受け付けたとき、当該データ蓄積装置に記憶される前記ストリームデータの最新データを前記フィルタ処理装置へ送信する最新データ管理部を備えることを特徴とする請求項１に記載のフィルタ処理システム。
【請求項３】
前記最新データ管理部は、
前記フィルタ処理装置から、前記第２のストリームデータの取得要求を受け付けたときに、当該データ蓄積装置に記憶される前記第２のストリームデータの受信時刻を参照して、この第２のストリームデータの最新データが、所定時間以上前のものであると判断したとき、当該前記フィルタ処理装置へ、前記第２のストリームデータの最新データは無い旨の応答を送信することを特徴とする請求項２に記載のフィルタ処理システム。
【請求項４】
前記ストリームデータ処理システムに、フィルタ処理装置が追加されたとき、
前記情報管理装置は、
前記追加されたフィルタ処理装置へ、前記ストリームデータ処理システム内の既存の前記フィルタ処理装置それぞれが備える機能と同じ機能を実現するためのファイルを送信する情報管理部を備えることを特徴とする請求項１ないし請求項３のいずれか１項に記載のストリームデータ処理システム。
【請求項５】
前記データ蓄積装置それぞれのキュー管理部は、
前記キュー部に蓄積されたストリームデータが、前記フィルタ処理装置に取得された後、所定時間以内に当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信するまで、他の前記フィルタ処理装置からの取得要求に応じず、
当該ストリームデータが、前記フィルタ処理装置に取得された後、前記所定時間が経過しても当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信しなかったとき、他の前記フィルタ処理装置からの取得要求に応じ、前記キュー部に蓄積された当該ストリームデータを送信することを特徴とする請求項１ないし請求項４のいずれか１項に記載のストリームデータ処理システム。
【請求項６】
前記データ蓄積装置それぞれのキュー管理部は、
前記キュー部に蓄積されたストリームデータが、前記フィルタ処理装置に取得された後、当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信したとき、前記キュー部に蓄積された当該ストリームデータを削除することを特徴とする請求項１ないし請求項５のいずれか１項に記載のストリームデータ処理システム。
【請求項７】
前記フィルタ処理装置はそれぞれ、
前記フィルタ実行制御部において、前記取得したストリームデータに対し実行するフィルタ処理に必要なファイルが当該フィルタ処理装置にないと判断したとき、前記情報管理装置から、前記取得したストリームデータに必要なフィルタ処理を実行するためのファイルを取得するファイル取得部を備えることを特徴とする請求項１ないし請求項６のいずれか１項に記載のストリームデータ処理システム。
【請求項８】
（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対し処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびそのフィルタ処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムにおけるストリームデータ処理方法であって、
前記データ蓄積装置はそれぞれ、
前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータをキュー部に蓄積するステップと、
前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信するステップと、
前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するステップとを実行し、
前記情報管理装置は、
前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶するステップを実行し、
前記フィルタ処理装置はそれぞれ、
前記ストリームデータの取得元となる前記データ蓄積装置を選択するステップと、
前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するステップと、
前記情報管理装置の定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定するステップと、
前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、
前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信して、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信するステップと、
前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合するステップと、
前記結合したストリームデータを、外部装置へ送信するステップとを実行することを特徴とするストリームデータ処理方法。
【請求項９】
（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびそのフィルタ処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムに用いられるストリームデータ処理プログラムであって、
前記データ蓄積装置に、
前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータをキュー部に蓄積するステップと、
前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信するステップと、
前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するステップとを実行させ、
前記情報管理装置に、
前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶するステップを実行させ、
前記フィルタ処理装置に、
前記ストリームデータの取得元となる前記データ蓄積装置を選択するステップと、
前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するステップと、
前記情報管理装置の前記定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定するステップと、
前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、
前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信して、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信するステップと、
前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合するステップと、
前記結合したストリームデータを、外部装置へ送信するステップとを実行させることを特徴とするストリームデータ処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ストリームデータ処理技術に関する。
【背景技術】
【０００２】
従来から、各種センサやＩＣタグ、携帯電話等、様々な情報源からデータを収集し、この収集したデータにフィルタ処理を行い、所定の利用先へ流通させるストリームデータ処理技術がある。このようなストリームデータ処理技術において、異なる情報源から送信されたストリームデータ（異種ストリームデータ）を合流させたい場合もある。例えば、映像センサから送信されるストリームデータと、対象物の位置情報を検出する位置センサから送信されるストリームデータとを合流させて、位置センサから得られる地図上の座標位置に映像ストリームを表示させたい場合もある。ここで、このようなストリームデータの合流を実現するには、非特許文献１に開示されるColal8や、非特許文献２に開示されるストリーム管理システムを用いることが考えられる。
【非特許文献１】Colal8、[online]、[平成20年2月27日検索]、インターネット、<URL: http://www.colal8.com>
【非特許文献２】赤間他、「追記・参照型データ管理プラットフォーム  アーキテクチャの提案」、マルチメディア通信と分散処理ワークショップ論文集、社団法人情報処理学会、2006年11月、p199〜204
【発明の開示】
【発明が解決しようとする課題】
【０００３】
しかし、非特許文献１に記載されるColal8によりストリームデータ合流処理を行う場合、１つのストリームデータ合流処理を行うのは１つのマシンに限定されるため、マシンを増設したとしてもストリームデータ合流処理のパフォーマンスを向上させることはできない。つまり、Colal8は、ストリームデータ合流処理に関し、スケールアウト特性（マシン増設による処理能力の向上）を出せない。
【０００４】
また、非特許文献２に記載されるストリーム管理システムによりストリームデータ合流処理を行う場合、このストリームデータ合流処理は、各フィルタ処理装置からのデータを受信する参照部（参照装置）が行う。よって、非特許文献２に記載されるストリーム管理システムも、ストリームデータ合流処理を行うのは１つのマシン（参照装置）であり、マシンを増設したとしてもストリームデータ合流処理のパフォーマンスを向上させることはできない。つまり、非特許文献２に記載されるストリーム管理システムも、ストリームデータ合流処理に関し、スケールアウト特性を出せない。
【０００５】
さらに、前記した技術のいずれもが、（ストリームデータ合流処理において）データ処理の可用性確保のため、処理を実行するマシンが故障したとき、待機している他のマシン（待機用のマシン）がその処理を引き継ぐActive-Standby方式を採用している。よって、システム内のデータ処理の可用性を確保しようとすると、別途、待機用マシンを用意しなくてはならず、この待機用マシンの処理能力に無駄が生じていた。また、このActive-Standby方式において、システム内のActiveマシンが二重起動するのを避けるためには、Standbyマシンが、ActiveマシンのActive死活監視結果（Activeか否かの監視結果）の受信を待つ必要がある。よって、StandbyマシンがActiveマシンに切り替わるときに時間がかかり（通常、約３０秒間）、その間サービスが停止していた。つまり、高可用性を確保できていなかった。
【０００６】
そこで、本発明は、前記した問題を解決し、ストリームデータの合流処理において、スケールアウト特性を出し、かつ、高可用性を確保することを目的とする。
【課題を解決するための手段】
【０００７】
前記した課題を解決するため、請求項１に記載の発明は、（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびその処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムであって、前記データ蓄積装置はそれぞれ、前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータを蓄積するキュー部と、前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信する位置情報送信部と、前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するキュー管理部とを備え、前記情報管理装置は、前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶する記憶部を備え、前記フィルタ処理装置はそれぞれ、前記ストリームデータの取得元となる前記データ蓄積装置を選択する選択処理部と、前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するデータ取得部と、前記情報管理装置の定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定し、当該フィルタ処理装置の制御を行うフィルタ実行制御部とを備え、前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信し、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信する情報源指定データ取得部と、前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合する選択結合部と、前記結合したストリームデータを、外部装置へ送信する出力部とを備えることを特徴とするストリームデータ処理システムとした。
【０００８】
請求項８に記載の発明は、（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対し処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびそのフィルタ処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムにおけるストリームデータ処理方法であって、
前記データ蓄積装置はそれぞれ、前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータをキュー部に蓄積するステップと、前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信するステップと、前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するステップとを実行し、前記情報管理装置は、前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶するステップを実行し、前記フィルタ処理装置はそれぞれ、前記ストリームデータの取得元となる前記データ蓄積装置を選択するステップと、前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するステップと、前記情報管理装置の定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定するステップと、前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信して、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信するステップと、前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合するステップと、前記結合したストリームデータを、外部装置へ送信するステップとを実行することを特徴とするストリームデータ処理方法とした。
【０００９】
このようにすることで、ストリームデータ処理システム（以下、システムと略す）の情報管理装置は、データ蓄積装置それぞれから当該データ蓄積装置に追記されたストリームデータの情報源位置情報を受信する。そして、フィルタ処理装置はそれぞれ、情報管理装置に、情報源位置情報を問い合わせることで、自身が受信したストリームデータに合流させるストリームデータ（データ）の参照先を知ることができる。よって、システムにおけるストリームデータ合流処理を、各フィルタ処理装置で行うことができる。これにより、システムは、フィルタ処理装置を追加することによるスケールアウト特性を持たせることができる。
【００１０】
なお、システムは、フィルタ処理装置側でデータの取得先となるデータ蓄積装置を選択し、データを取得する。したがって、例えば、システム内のいずれかのフィルタ処理装置に故障が発生した場合でも、データ蓄積装置に蓄積されるストリームデータは他のいずれかのフィルタ処理装置により取得され、処理が実行される。つまり、システムは、フィルタ処理装置に関しActive-Standby方式ではなく、Active-Active方式（処理中のマシンが、故障したマシンの処理を引き継ぐ方式）を採用するので、高可用性を確保できる。
【００１１】
なお、フィルタ処理装置がデータの取得先となるデータ蓄積装置の選択するときの方法としては、各フィルタ処理装置がデータ蓄積装置それぞれのキュー状態（例えば、キュー長）を監視し、よりキューが長いデータ蓄積装置をデータの取得先として選択する方法が考えられる。このようにすることで、所定のデータ蓄積装置にデータが集中した場合でも、このデータ蓄積装置からデータが溢れることがなくなり、ストリームデータを効率的に処理できる。
【００１２】
また、システムのデータ蓄積装置は１以上の追記部を備え、その追記部それぞれが、前記したデータ蓄積装置と同様の処理を行うようにしてもよい。さらに、フィルタ処理装置も１以上の処理部を備え、その処理部それぞれが、前記したフィルタ処理装置と同様の処理を行うようにしてもよい。
【００１３】
請求項２に記載の発明は、請求項１に記載のフィルタ処理システムの前記データ蓄積装置がそれぞれ、前記受信したストリームデータをその受信時刻とともに当該データ蓄積装置の記憶部に記憶し、前記フィルタ処理装置から、前記第２のストリームデータの取得要求を受け付けたとき、当該データ蓄積装置の記憶部に記憶される前記第２のストリームデータの最新データを前記フィルタ処理装置へ送信する最新データ管理部を備えることを特徴とする。
【００１４】
このようにすることで、ストリームデータ処理システムのフィルタ処理装置は、データ蓄積装置に蓄積された第２のストリームデータの最新データと、この第１のストリームデータとを合流させることができる。
【００１５】
請求項３に記載の発明は、請求項２に記載のフィルタ処理システムの前記最新データ管理部が、前記受信したストリームデータをその受信時刻とともに当該データ蓄積装置の記憶部に記憶し、前記フィルタ処理装置から、前記第２のストリームデータの取得要求を受け付けたときに、当該データ蓄積装置に記憶される前記第２のストリームデータの受信時刻を参照して、この第２のストリームデータの最新データが、所定時間以上前のものであると判断したとき、当該前記フィルタ処理装置へ、前記第２のストリームデータの最新データは無い旨の応答を送信することを特徴とする。
【００１６】
このようにすることで、ストリームデータ処理システムのフィルタ処理装置は、データ蓄積装置に蓄積される第２のストリームデータの最新データが、受信してから所定時間以上経過したものであるとき、これを参照しないようにすることができる。つまり、フィルタ処理装置は、ストリームデータの合流処理を行うとき、第１のストリームデータに対しタイムラグの大きい第２のストリームデータを合流させないようにできる。
【００１７】
請求項４に記載の発明は、請求項１ないし請求項３のいずれか１項に記載のストリームデータ処理システムに、フィルタ処理装置が追加されたとき、前記情報管理装置は、前記追加されたフィルタ処理装置へ、前記ストリームデータ処理システム内の既存の前記フィルタ処理装置それぞれが備える機能と同じ機能を実現するためのファイルを送信する情報管理部を備えることを特徴とする。
【００１８】
このようにすることで、ストリームデータ処理システムは、フィルタ処理装置が追加されたときも、このフィルタ装置に既存のフィルタ処理装置と同じ機能を持たせることができる。
【００１９】
請求項５に記載の発明は、請求項１ないし請求項４のいずれか１項に記載のストリームデータ処理システムにおいて、前記データ蓄積装置それぞれのキュー管理部が、前記キュー部に蓄積されたストリームデータが、前記フィルタ処理装置に取得された後、所定時間以内に当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信するまで、他の前記フィルタ処理装置からの取得要求に応じず、当該ストリームデータが、前記フィルタ処理装置に取得された後、前記所定時間が経過しても当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信しなかったとき、他の前記フィルタ処理装置からの取得要求に応じ、前記キュー部に蓄積された当該ストリームデータを送信することを特徴とする。
【００２０】
このようにすることで、ストリームデータ処理システムにおいて、フィルタ処理装置が故障した場合でも、そのフィルタ処理装置において実行中であったデータと同じデータを他のいずれかのフィルタ処理装置がデータ蓄積装置から取得する。よって、ストリームデータ処理システムは、フィルタ処理装置が故障した場合でも、確実にフィルタ処理を実行することができる。
【００２１】
請求項６に記載の発明は、請求項１ないし請求項５のいずれか１項に記載のストリームデータ処理システムにおいて、前記データ蓄積装置それぞれのキュー管理部が、前記キュー部に蓄積されたストリームデータが、前記フィルタ処理装置に取得された後、当該フィルタ処理装置から当該ストリームデータに対する処理の完了通知を受信したとき、前記キュー部に蓄積された当該ストリームデータを削除することを特徴とする。
【００２２】
このようにすることで、ストリームデータ処理システムのフィルタ処理装置において処理中のストリームデータについて、データ蓄積装置から、他のフィルタ処理装置がこれと同じストリームデータを重複して取得することがなくなる。よって、各フィルタ処理装置における処理の重複を避けることができる。
【００２３】
請求項７に記載の発明は、請求項１ないし請求項６のいずれか１項に記載のストリームデータ処理システムにおいて、前記フィルタ処理装置がそれぞれ、前記フィルタ実行制御部において、前記取得したストリームデータに対し実行するフィルタ処理に必要なファイルが当該フィルタ処理装置にないと判断したとき、前記情報管理装置から、前記取得したストリームデータに必要なフィルタ処理を実行するためのファイルを取得するファイル取得部を備えることを特徴とする。
【００２４】
このようにすることで、ストリームデータ処理システムは、各フィルタ処理装置自身で、処理に必要なファイルを取得するので、このストリームデータ処理システムの管理者等が各フィルタ処理装置に必要な機能を設定する手間が省ける。
【００２５】
請求項９に記載の発明は、（１）複数の情報源端末から送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）このデータ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記ストリームデータの種別を示すＳＩＤごとに、このＳＩＤを持つストリームデータに対し実行すべき前記フィルタ処理およびそのフィルタ処理の適用手順を示した定義情報を記憶する情報管理装置とを備えるストリームデータ処理システムに用いられるストリームデータ処理プログラムであって、前記データ蓄積装置に、前記情報源端末から前記ストリームデータを受信したとき、この受信したストリームデータをキュー部に蓄積するステップと、前記受信したストリームデータのＳＩＤと、自データ蓄積装置の識別情報とを対応付けた情報源位置情報を作成し、前記情報管理装置へ送信するステップと、前記フィルタ処理装置からの要求に応じて、前記キュー部に蓄積されたストリームデータを出力するステップとを実行させ、前記情報管理装置に、前記データ蓄積装置それぞれから送信された前記情報源位置情報を記憶するステップを実行させ、前記フィルタ処理装置に、前記ストリームデータの取得元となる前記データ蓄積装置を選択するステップと、前記選択されたデータ蓄積装置のキュー部から、このキュー部に蓄積された第１のストリームデータを取得するステップと、前記情報管理装置の前記定義情報と、前記取得した第１のストリームデータに付されたＳＩＤとを参照して、前記取得した第１のストリームデータに対し実行すべき処理を決定するステップと、前記決定した処理が、取得した第１のストリームデータと、この第１のストリームデータ以外の第２のストリームデータとを合流させる処理であったとき、前記情報管理装置へ、前記第２のストリームデータのＳＩＤを含む前記第２のストリームデータの情報源位置情報の取得要求を送信して、その結果取得した第２のストリームデータの情報源位置情報を参照して、前記第２のストリームデータを蓄積するデータ蓄積装置から、前記第２のストリームデータを受信するステップと、前記定義情報に従い、前記取得した第１のストリームデータと前記第２のストリームデータとの所定カラムを選択して結合するステップと、前記結合したストリームデータを、外部装置へ送信するステップとを実行させることを特徴とするストリームデータ処理プログラムとした。
【００２６】
このようなプログラムによれば、一般的なコンピュータに請求項８に記載のストリームデータ処理方法を実行させることができる。
【発明の効果】
【００２７】
本発明によれば、ストリームデータ処理システムは、ストリームデータの合流処理において、スケールアウト特性を出し、かつ、高可用性を確保するができる。
【発明を実施するための最良の形態】
【００２８】
以下、本発明を実施するための最良の形態（以下、実施の形態という）を、図面を参照しながら説明する。まず、本実施の形態のストリームデータ処理システムの概要を説明する。
【００２９】
図１は、本実施の形態のストリームデータ処理システムを含むシステムの構成例を示した図である。図１に示すように、システムは、様々な情報源から取得したデータに情報源ＩＤ（ＳＩＤ）を付し、このデータをストリームデータとして送信する追記クライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）と、受信したストリームデータのフィルタ処理を行うストリームデータ処理システム１と、フィルタ処理されたデータの出力先である、参照装置６０（６０Ａ，６０Ｂ，…，６０Ｎ）と、この参照装置６０からデータを取得したり、検索したりする参照クライアント７０（７０Ａ，７０Ｂ，…，７０Ｎ）とを含んで構成される。
【００３０】
ストリームデータ処理システム１は、追記クライアント５０から送信されたストリームデータを蓄積するデータ蓄積装置２０（２０Ａ，２０Ｂ）と、ストリームデータのフィルタ処理を行うフィルタ処理装置１０（１０Ａ，１０Ｂ）と、このフィルタ処理装置１０におけるフィルタ処理部１２４（詳細は後記）の機能を実現するファイルデータ等を記憶する情報管理装置３０とを含んで構成される。データ蓄積装置２０は、ストリームデータの受け付けやキュー管理等を行う追記部（追記部２２１，２２２等）を１以上備える。また、フィルタ処理装置１０も、データの取得やデータのフィルタ処理を行う処理部１２（処理部１２Ａ，１２Ｂ等）を１以上備える。情報管理装置３０、データ蓄積装置２０およびフィルタ処理装置１０の数は、図１に示す数に限定されない。なお、このフィルタ処理装置１０（処理部１２）はそれぞれ同じフィルタ処理部１２４を備え、同じ処理手順情報３２４を参照して処理を実行するものとする。つまり、フィルタ処理装置１０はそれぞれ、同じＳＩＤを持つデータを受信したとき、処理手順情報３２４を参照し、このデータに対し同じ処理手順で同じ処理を実行するものとする。よって、ストリームデータ処理システム１のフィルタ処理装置１０（処理部１２）を増設すれば、システム全体の処理能力を向上させることができる。
【００３１】
参照装置６０は、フィルタ処理装置１０から出力されたデータの入力を受け付ける入力部６２と、入力されたデータについて、そのデータに付されたＳＩＤを用いて整列（ソート）する整列部６４と、データの整列に用いる整列バッファ６６と、整列されたデータを参照クライアント７０へ出力する出力部６５とを含んで構成される。
【００３２】
また、この参照装置６０は、参照装置６０Ｂのように、データの入力を受け付ける入力部６２と、フィルタ処理されたデータを蓄積するＤＢ（データベース）６１と、参照クライアント７０の要求に応じて、このＤＢ６１から所定のデータを検索する検索処理部６３とを含む構成としてもよい。
【００３３】
なお、（１）追記クライアント５０とストリームデータ処理システム１との間、（２）ストリームデータ処理システム１と参照装置６０との間はそれぞれ、図示しない伝送経路により接続される。また、参照装置６０、参照クライアント７０の数は、図１に示す数に限定されない。
【００３４】
＜動作概要＞
ここで、図１のシステムの動作概要を説明する。
【００３５】
＜単一ストリームデータの処理＞
まず、ストリームデータ処理システム１の動作概要を説明するため、図１のシステムが単一のストリームデータを処理する場合について説明する。ここでは、追記クライアント５０Ａがストリームデータをデータ蓄積装置２０Ａの追記部２２１へ出力し、この追記部２２１内に蓄積されたストリームデータ（データ）をフィルタ処理装置１０Ａの処理部１２Ａが取得する場合を例に説明する。
【００３６】
追記クライアント５０Ａは、様々な情報源からデータを取得し、そのデータに応じたＳＩＤを付与する。次に、追記クライアント５０Ａは、情報管理装置３０の接続先割当部３１（後記）により指示されたデータ蓄積装置２０（２０Ａ）へストリームデータを送信する。そして、データ蓄積装置２０Ａは、受信したストリームデータをキュー部２２４（後記）に蓄積する。また、このとき追記部２２１の位置情報送信部２２８（後記）は、受信したストリームデータのＳＩＤと、自身の追記部２２１のＩＰアドレスとポート（ポート番号）とを示す情報源位置情報を情報管理装置３０のデータ位置管理部３３へ送信する。データ位置管理部３３は、受信した情報源位置情報を情報源位置情報ＤＢ（データベース）３２５に登録しておく。このようにすることで、情報管理装置３０は、各ストリームデータが、どのデータ蓄積装置２０の、どの追記部に蓄積されているかを把握できる。
【００３７】
続いて、フィルタ処理装置１０Ａの処理部１２Ａは、データの取得先となるデータ蓄積装置２０を選択し、この選択したデータ蓄積装置２０Ａからストリームデータを取得する。そして、処理部１２Ａは、取得したストリームデータに対しフィルタ処理を行う。このストリームデータに対し行うフィルタ処理は、取得したデータに付されたＳＩＤと、情報管理装置３０の処理手順情報３２４に示される当該ＳＩＤのデータに関する処理手順内容を参照して判断される。
【００３８】
この後、フィルタ処理装置１０は、フィルタ処理後のデータを所定の参照装置６０Ａへ出力する。そして、この参照装置６０Ａの整列部６４は、整列バッファ６６内のデータを整列（例えば、同じＳＩＤを持つストリームデータについて時系列に並べる等）し、参照クライアント７０Ａへ出力する。
【００３９】
なお、詳細は後記するが、各データ蓄積装置２０の追記部のキュー管理部２２５は、フィルタ処理装置１０の処理部１２によりデータが取得され、この処理部１２における処理が完了したことを検知すると、このデータをキュー部２２４から削除する。このようにすることで、他のフィルタ処理装置１０の処理部１２が、同じデータを取得し、各処理部１２で重複した処理を行わないようにできる。
【００４０】
＜異種ストリームデータの合流処理＞
次に、このようなストリームデータ処理システム１において、フィルタ処理装置１０が異種ストリームデータの合流処理を行う場合について説明する。なお、以下の説明において、合流の契機になるストリームデータをメインストリームと呼び、合流対象のストリームデータで、メインストリームでないものをサブストリームと呼ぶ。なお、以下の説明において、このサブストリームは、最終的に処理部１２において破棄されるものとして説明するが、これに限定されない。
【００４１】
ここでは、フィルタ処理装置１０Ａの処理部１２Ａが、追記クライアント５０Ａから出力されたストリームデータをメインストリームとし、追記クライアント５０Ｂから出力されたストリームデータをサブストリームとして合流処理を行う場合を例に説明する。なお、この追記クライアント５０Ａは、ストリームデータをデータ蓄積装置２０Ａの追記部２２１へ出力し、追記クライアント５０Ｂはこのストリームデータをデータ蓄積装置２０Ｂの追記部２２１へ出力するものとする。
【００４２】
この追記部２２１は、追記クライアント５０Ｂからストリームデータを受信すると、位置情報送信部２２８により、このストリームデータに関する情報源位置情報（データのＳＩＤと、データ蓄積装置２０Ｂの追記部２２１のＩＰアドレスとポート番号）を情報管理装置３０へ送信する。なお、追記部２２１も、追記クライアント５０Ａからストリームデータを受信すると、位置情報送信部２２８により、情報源位置情報（データのＳＩＤと、データ蓄積装置２０Ａの追記部２２１のＩＰアドレスと、ポート）を情報管理装置３０へ送信する。
【００４３】
このような処理の後、フィルタ処理装置１０Ａの処理部１２Ａはデータ取得部１２２により、データ蓄積装置２０Ａからデータ（追記クライアント５０Ａから出力されたデータ）を取得する。この処理部１２Ａは、このデータに付されたＳＩＤと、情報管理装置３０の処理手順情報３２４等とを参照して、このデータに対し実行すべき処理を判断する。ここで、このデータに対し実行すべき処理が、他のストリームデータ（例えば、追記クライアント５０Ｂから出力されるストリームデータ）との合流処理であると判断したとき、処理部１２Ａは、以下のような処理を行う。
【００４４】
すなわち、処理部１２Ａは最新データ取得部１２７により、情報管理装置３０のデータ位置管理部３３に対し、合流させるストリームデータ（ここでは追記クライアント５０Ｂから出力されたストリームデータ）の情報源位置情報を問い合わせる。このような問い合わせを受けた情報管理装置３０のデータ位置管理部３３は、処理部１２Ａの最新データ取得部１２７へこのストリームデータの情報源位置情報（データのＳＩＤと、データ蓄積装置２０Ｂの追記部２２１のＩＰアドレスと、ポート番号）を送信する。
【００４５】
このような情報源位置情報を受信した処理部１２Ａの最新データ取得部１２７は、情報源位置情報に示される追記部２２１にアクセスし、この追記部２２１の最新データ管理部２２６から合流させるストリームデータ（追記クライアント５０Ｂから出力されたストリームデータ）の最新データの参照を開始する。
【００４６】
なお、以下の説明において、フィルタ処理装置１０の処理部１２がデータ取得部１２２により、データ蓄積装置２０の追記部２２１，２２２のキュー管理部２２５からデータを取得することを「データを取得する」という。一方、フィルタ処理装置１０の処理部１２が最新データ取得部１２７により、データ蓄積装置２０の追記部２２１，２２２のデータを取得することを「データを参照する」という。
【００４７】
そして、処理部１２Ａは、この参照したデータと、自身が保持するデータ（追記クライアント５０Ａから出力されたストリームデータ）とを用いてストリームデータの合流処理を行う。そして、処理部１２Ａは、フィルタ処理部１２４により、合流後のデータのフィルタ処理を行うと、フィルタ処理後のデータを参照装置６０（６０Ａ）へ出力する。
【００４８】
また、図示を省略しているが、このようなストリームデータの合流処理は、他の処理部１２（他のフィルタ処理装置１０）においても行われる。例えば、フィルタ処理装置１０Ｂの処理部１２Ａも、データ蓄積装置２０Ａの追記部２２１から、追記クライアント５０Ａより出力されたストリームデータを取得し、ストリームデータの合流処理を行い、参照装置６０（６０Ａ）へ出力する。そして、参照装置６０（６０Ａ）は、フィルタ処理装置１０の処理部１２それぞれにおいて処理されたデータ（つまり、一連のストリームデータについて各処理部１２においてデータ単位で処理されたもの）を受信すると、これを整列バッファ６６内で整列させて、参照クライアント７０（７０Ａ）へ出力する。
【００４９】
このように、フィルタ処理装置１０（処理部１２）それぞれが、このデータ蓄積装置２０のキュー部２２４からデータを取得してストリームデータの合流およびフィルタ処理を行う。つまり、ストリームデータの合流処理についてフィルタ処理装置１０（処理部１２）で分散して処理を行う。よって、データ蓄積装置２０に流れ込むストリームデータのデータ量が増加し、処理負荷が増加した場合でも、各フィルタ処理装置１０を増設することでストリームデータ処理システム１の処理能力を向上させることができる。なお、フィルタ処理装置１０の処理部１２は、データのバッファを持たないので、処理部１２の追加や変更を行っても、データが失われることはない。つまり、管理者等は、このフィルタ処理装置１０の処理部１２の追加や変更を行いやすくなる。
【００５０】
また、ここでは説明を省略したが、データ蓄積装置２０からフィルタ処理装置１０の最新データ取得部１２７により参照されたデータは、ストリームデータ処理システム１内のいずれかのフィルタ処理装置１０の処理部１２のデータ取得部１２２より取得されるが、このデータは、このデータを取得した処理部１２においてフィルタ処理の対象としてもよいし、データ破棄部１２６（後記）が破棄するようにしてもよい。このデータをフィルタ処理の対象とするか、破棄の対象とするかは、前記した情報管理装置３０の処理手順情報３２４の記載内容による。
【００５１】
なお、この追記クライアント５０、データ蓄積装置２０、情報管理装置３０、フィルタ処理装置１０、参照装置６０および参照クライアント７０は、入出力インタフェース、ＣＰＵ（Central Processing Unit）等の演算処理装置、ＲＡＭ（Random Access Memory）、ＲＯＭ（Read Only Memory）、ＨＤＤ（Hard Disk Drive）等の記憶装置等を備えるコンピュータにより実現される。また、各処理部は、ＣＰＵによるプログラム実行処理により実現される。
【００５２】
また、データ蓄積装置２０（２０Ａ，２０Ｂ）は、それぞれ別個のコンピュータ（装置）により実現するようにしてもよいし、これらをまとめて１つのコンピュータにより実現するようにしてもよい。さらに、フィルタ処理装置１０（１０Ａ，１０Ｂ）についても、それぞれ別個のコンピュータ（装置）により実現するようにしてもよいし、これらをまとめて１つのコンピュータにより実現するようにしてもよい。つまり、データ蓄積装置２０（２０Ａ，２０Ｂ）およびフィルタ処理装置１０（１０Ａ，１０Ｂ）をそれぞれ、プロセスにより実現するようにしてもよいし、コンピュータ（マシン）により実現するようにしてもよい。さらに、情報管理装置３０の接続先割当部３１は、各データ蓄積装置２０の処理部の負荷状態を監視するロードバランサを備え、このロードバランサにより追記クライアント５０の接続先となる追記部２２１,２２２を指示するようにしてもよい。
【００５３】
＜データ蓄積装置＞
引き続き図１を用いてデータ蓄積装置２０を詳細に説明する。前記したとおり、データ蓄積装置２０はそれぞれ、追記クライアント５０から出力されたストリームデータを一時的に蓄積する。このようなデータ蓄積装置２０は、１以上の追記部（追記部２２１，２２２）と、プロセス管理部２６とを含む。この追記部２２１，２２２は、入力部（図示省略）経由で追記クライアント５０から出力されたストリームデータを取得すると、これをキューとして蓄積する。そして、この追記部２２１，２２２は、フィルタ処理装置１０の処理部１２からデータの取得要求を受け付けると、このストリームデータを出力部（図示省略）経由で処理部１２へ送信する。プロセス管理部２６は、追記部２２１，２２２の制御を行う。なお、データ蓄積装置２０の備える追記部の数は、図１に示した数に限定されない。
【００５４】
このような追記部２２１，２２２はそれぞれ、ストリーム受付部２２３と、キュー部２２４と、キュー管理部２２５と、最新データ管理部２２６と、位置情報送信部２２８と、最新データ記憶部２２７とを含んで構成される。
【００５５】
ストリーム受付部２２３は、追記クライアント５０から、ストリームデータを受信する。そして、この受信したストリームデータをキュー管理部２２５および最新データ管理部２２６へ出力する。
【００５６】
キュー部２２４は、キュー管理部２２５から出力されたストリームデータをキューとして蓄積する。なお、このキュー部２２４は、データ蓄積装置２０のＲＡＭ等に形成される。
【００５７】
キュー管理部２２５は、ストリーム受付部２２３から出力されたストリームデータをキュー部２２４に出力する。そして、このキュー管理部２２５は、フィルタ処理装置１０の処理部１２からデータの取得要求を受け付けたとき、キュー部２２４に蓄積されたデータをフィルタ処理装置１０の処理部１２へ送信する。
【００５８】
なお、このキュー管理部２２５は、キュー部２２４のキュー状態（キュー長等）が所定の閾値以上に変化したとき、このキュー状態情報を同報送信等により各フィルタ処理装置１０の処理部１２へ送信するようにしてもよい。このようにすることで、各フィルタ処理装置１０の処理部１２は、データ蓄積装置２０の追記部２２１，２２２のキュー状態をもとにデータの取得元となる追記部２２１，２２２を選択することができる。例えば、処理部１２は、より多くのデータがキュー部２２４に溜まっているデータ蓄積装置２０Ａの追記部２２１を優先的にデータ取得先として選択する。
【００５９】
また、このキュー管理部２２５は、フィルタ処理装置１０の処理部１２へキュー部２２４のデータを出力した後、このデータをすぐに削除せず、所定時間、仕掛り中のフラグを立ててキュー部２２４に保存しておく。そして、キュー管理部２２５は、この仕掛り中のフラグの立ったデータについては、他の処理部１２からの取得要求を受け付けても出力しないようにする。ここで、キュー管理部２２５は、当該データを出力した処理部１２から、このデータについて、所定時間以内に当該データの処理が完了した旨の通知を受信したとき、このキュー部２２４に保存していた当該データを削除する。一方、この仕掛り中のフラグの立ったデータについて、所定時間経過しても、処理部１２から処理が完了した旨の通知が来ないとき、キュー管理部２２５は、当該データから仕掛り中のフラグをはずし、他の処理部１２からのデータ取得要求に応じる。このようにすることで、例えば、処理部１２Ａにおいてデータの処理中に障害等が発生し、処理できなかった場合でも、このデータを他の処理部（例えば、処理部１２Ｂ）に取得させ、処理させることができる。つまり、当該データの処理を行う処理部１２が別の処理部１２に切り替わったときでも、ストリームデータの処理を確実なものとすることができる。なお、このキュー管理部２２５が、当該データに仕掛り中のフラグを立ててから、フラグをはずすまでの時間（タイムアウト時間）は、前記したActive-Standby方式において、StandbyマシンがActiveマシンのActive死活監視結果の受信を待つ時間（約３０秒間）よりも非常に短い。これは、通常、処理部１２において、キュー管理部２２５からデータを取得してから処理が完了するまでの時間は非常に短いので（１秒未満）、これにあわせたタイムアウト時間を設定すれば、キュー管理部２２５は、処理部１２において処理が正常に完了したか否かを充分判断できるからである。つまり、ストリームデータ処理システム１は、Active-Standby方式に比べてタイムアウト時間を非常に短く設定できるので、フィルタ処理装置１０の処理部１２に故障が発生したときの切り替えを速やかに行うことができる。
【００６０】
最新データ管理部２２６は、ストリーム受付部２２３で受信したストリームデータを、その受信時刻（受付時刻）とともに最新データ記憶部２２７へ格納する。また、この最新データ管理部２２６は、フィルタ処理装置１０の処理部１２から最新データの参照要求を受け付けたとき、この最新データ記憶部２２７に格納されるデータのうち、この参照要求の対象データの最新データを読み出し、送信する。このようにすることで、フィルタ処理装置１０は、データ蓄積装置２０から所望するストリームデータの最新データを受け取ることができる。これにより、フィルタ処理装置１０は、自身がデータ取得部１２２により取得したストリームデータ（メインストリーム）とほぼ同期した状態で、他のストリームデータとの合流させることができる。
【００６１】
なお、この最新データ管理部２２６は、フィルタ処理装置１０の最新データ取得部１２７からのデータ参照要求を受信した場合において、このデータ取得要求の対象であるデータの最新データが、その受信から所定時間以上経過したものであったとき、この取得要求の対象であるデータは無い旨を応答するようにしてもよい。このようにすることで、フィルタ処理装置１０の最新データ取得部１２７は、追記部２２１，２２２から所定時間以上古いデータを参照することがなくなる。
【００６２】
位置情報送信部２２８は、ストリーム受付部２２３において、ストリームデータを受け付けると、このストリームデータのＳＩＤと、自身の属する処理部（追記部２２１，２２２）のＩＰアドレスとポート（ポート番号）とを対応付けた情報源位置情報を作成し、情報管理装置３０へ送信する。
【００６３】
最新データ記憶部２２７は、最新データ管理部２２６から出力されたストリームデータをその受信時刻とともに記憶する。なお、本実施の形態では、最新データ記憶部２２７と、キュー部２２４とは別個の構成としているが、これらをまとめて１つの記憶部とし、キュー管理部２２５と最新データ管理部２２６との両方からのアクセスを受け付けるようにしてもよい。また、追記部２２１，２２２が最新データ記憶部２２７とキュー部２２４との両方を用いる場合において、この最新データ記憶部２２７に蓄積されるデータに相当するデータが、キュー部２２４から削除されたときも、最新データ管理部２２６はそのデータは削除せずに保持し続ける。このようにすることで、当該追記部２２１，２２２のキュー部２２４に蓄積されていたデータが、フィルタ処理装置１０の処理部１２により取得された後、別の処理部１２の最新データ取得部１２７による当該データの参照要求があったときも、追記部２２１，２２２は、この参照要求に応えることができる。
【００６４】
＜フィルタ処理装置＞
次に、図２を用いてフィルタ処理装置１０を詳細に説明する。図２は、図１のフィルタ処理装置の機能ブロック図である。
【００６５】
フィルタ処理装置１０は、データ蓄積装置２０からのストリームデータや情報管理装置３０からのファイルの入力を受け付ける入力部１１と、受け付けたストリームデータに対し、他のストリームデータの合流処理やフィルタ処理を行う処理部１２と、この処理部１２から処理されたデータを参照装置６０等の外部装置へ出力する出力部１３とを含んで構成される。この入力部１１および出力部１３も、データ蓄積装置２０や参照装置６０との間でデータ入出力可能な入出力インタフェースにより実現され、処理部１２は、ＣＰＵによるプログラムの実行処理により実現される。なお、図示を省略しているが、この処理部１２は、この処理部１２の機能を実現するためのプログラムを記憶する記憶部を備えるものとする。
【００６６】
この処理部１２は、キュー選択部（選択処理部）１２１と、データ取得部１２２と、フィルタ実行制御部１２３と、フィルタ処理部１２４（１２４Ａ〜１２４Ｎ）と、選択結合部１２５と、データ破棄部１２６と、最新データ取得部（情報源指定データ取得部）１２７と、ファイル取得部１２８とを含んで構成される。
【００６７】
キュー選択部１２１は、データの取得先となるデータ蓄積装置２０の追記部２２１，２２２を選択する。このときの選択方法は、様々な方法が考えられるが、例えば、各データ蓄積装置２０の追記部２２１，２２２から送信されるキュー状態情報をもとに、最もキュー長の長い追記部２２１，２２２を選択する。
【００６８】
データ取得部１２２は、キュー選択部１２１により選択されたデータ蓄積装置２０の追記部２２１，２２２からデータを受信する。なお、データ取得部１２２がデータを取得するタイミングは、自身の処理部１２において処理しているデータがなくなったときでもよいし、処理部１２で処理しているデータが所定の閾値以下となったときでもよい。
【００６９】
フィルタ実行制御部１２３は、データ取得部１２２、フィルタ処理部１２４、選択結合部１２５、データ破棄部１２６、最新データ取得部１２７およびファイル取得部１２８を制御する。具体的には、このフィルタ実行制御部１２３は、データ取得部１２２から出力されたデータに付されたＳＩＤと、情報管理装置３０の情報源情報３２２、スキーマ情報３２３および処理手順情報３２４（後記する図３参照）を参照して、このストリームデータに対し実行すべき処理を決定する。そして、フィルタ実行制御部１２３は、選択結合部１２５、フィルタ処理部１２４、データ破棄部１２６、最新データ取得部１２７等を制御して、この決定した処理内容を実行させる。
【００７０】
例えば、フィルタ実行制御部１２３は、データ取得部１２２から「ＳＩＤ＝１」のデータの出力を受け付けると、情報源情報３２２、スキーマ情報３２３および処理手順情報３２４を参照して、この「ＳＩＤ＝１」のデータに対し行うべき処理を判断する。ここで、判断された処理が「ＳＩＤ＝１」のデータ（ストリームデータ）に「ＳＩＤ＝２」のデータ（ストリームデータ）を合流させる処理である場合、フィルタ実行制御部１２３は、最新データ取得部１２７に、この「ＳＩＤ＝２」のデータの参照処理を実行させる。次に、フィルタ実行制御部１２３は、選択結合部１２５に、取得した「ＳＩＤ＝１」のデータの所定カラムと、「ＳＩＤ＝２」のデータの所定カラムとを結合させる。そして、フィルタ実行制御部１２３は、この結合後のデータに対し、フィルタ処理部１２４に、処理手順情報３２４に示される手順でフィルタ処理を実行させ、処理後のデータを出力部１３経由で参照装置６０へ出力する。
【００７１】
フィルタ処理部１２４は、フィルタ実行制御部１２３からの指示に基づき、ストリームデータに対し所定のフィルタ処理を実行する。このフィルタ処理部１２４は、図２に示すように１以上のフィルタ処理部１２４（１２４Ａ〜１２４Ｎ）により構成され、どのフィルタ処理部１２４にフィルタを実行させるかは、フィルタ実行制御部１２３が決定する。
【００７２】
選択結合部１２５は、フィルタ実行制御部１２３からの指示に基づき、ストリームデータの所定カラムのデータ同士を結合させる。例えば、取得した「ＳＩＤ＝１」のデータの所定カラムと、「ＳＩＤ＝２」のデータの所定カラムとを結合させる。なお、最新データ取得部１２７において、追記部２２１，２２２から当該ストリームデータの最新データを受信できなかったとき、当該データについてブランクの状態で結合させるようにしてもよい。
【００７３】
なお、この選択結合部１２５は、データ（カラム）の結合を行うとき、メインストリームのデータの到着を契機として行うので、メインストリームとして受信したデータのデータ数と、合流後のデータのデータ数とは同じになる。例えば、メインストリームのデータ数が、１分間に「３０」であり、サブストリームのデータ数が、１分間に「１０」であるとき、合流後のデータ数は１分間に「３０」となる。つまり、メインストリームのデータ数と、サブストリームのデータ数とは必ずしも同じにならない。よって、メインストリームのデータに対応するサブストリームのデータがなかった場合には、選択結合部１２５は、当該データについてはブランクの状態で処理する。
【００７４】
データ破棄部１２６は、フィルタ実行制御部１２３からの指示に基づき所定のＳＩＤを持つデータを破棄する。例えば、フィルタ実行制御部１２３が、前記した処理手順情報３２４を参照して、当該ＳＩＤのデータについて特に実行すべき処理がないと判断したとき、データ破棄部１２６はこのデータを破棄する。
【００７５】
最新データ取得部１２７は、フィルタ実行制御部１２３からの指示に従い、データ蓄積装置２０の追記部２２１，２２２にアクセスし、この追記部２２１，２２２に蓄積されるデータ（例えば、ストリームデータの最新データ）を参照する。このときのデータの参照先となるデータ蓄積装置２０の追記部２２１，２２２は、情報管理装置３０から取得した当該ＳＩＤのデータに関する情報源位置情報に基づき選択する。また、ここで最新データ取得部１２７が取得するデータの数は、最新データ１件だけではなく、最近のデータを所定個としてもよい。なお、この最新データ取得部１２７は、以前接続した追記部２２１，２２２の通信コネクション情報がキャッシュとして残っていれば、そのキャッシュに残っている通信コネクション情報を用いて追記部２２１，２２２に接続してデータを参照するようにしてもよい。このようにすることで、最新データ取得部１２７のデータ参照処理を高速化できる。また、このキャッシュは、データ取得部１２２および最新データ取得部１２７でお互いに共有するようにしてもよい。例えば、最新データ取得部１２７がデータ蓄積装置２０Ａの追記部２２１から最新データの参照処理を行う場合において、データ取得部１２２がこの追記部２２１からデータを取得したときの通信コネクション情報がキャッシュとして残っていれば、これを用いて通信コネクションを確立して最新データの参照処理を行う。このようにすることで、各処理部１２がデータ蓄積装置２０の追記部２２１，２２２からデータを取得したり、参照したりするときの処理を高速化できる。
【００７６】
ファイル取得部１２８は、情報管理装置３０から、処理部１２においてフィルタ処理機能を実現するためのファイル等を情報管理装置３０から受信する。このデータは、ファイル取得部１２８側からプル型で取得してもよいし、情報管理装置３０側からプッシュ型でファイル取得部１２８へ送信するようにしてもよい。なお、ファイル取得部１２８がプル型で取得する場合の取得タイミングは、当該処理部１２において必要なフィルタ処理機能がないことを検知したときでもよいし、所定期間ごとでもよい。また、このファイルを情報管理装置３０がプッシュ型で送信する場合の送信タイミングは、例えば、情報管理装置３０に格納される情報が最新の情報に変わったタイミング等である。
【００７７】
＜情報管理装置＞
次に、適宜、図１および図２を参照しつつ、図３を用いて、情報管理装置３０を詳細に説明する。図３は、図１の情報管理装置の機能ブロック図である。
【００７８】
情報管理装置３０は、フィルタ処理装置１０における処理に用いられる各種データやデータ処理における定義情報を蓄積する。また、追記クライアント５０の接続先の割り当てや、データ蓄積装置２０（２０Ａ，２０Ｂ）におけるストリームデータの蓄積位置（情報源位置情報）の記録等を行う。
【００７９】
このような情報管理装置３０は、記憶部３２と、接続先割当部３１と、データ位置管理部３３と、情報管理部３４とを備える。なお、ここでは図示を省略しているが、情報管理装置３０は、各種データの入力を司る入力部と、出力を司る出力部と、ＣＰＵ（Central Processing Unit）、メモリ、通信インタフェース等を備える。
【００８０】
記憶部３２は、システム情報３２１、情報源情報（ＳＩＤ情報）３２２、スキーマ情報３２３、処理手順情報３２４および情報源位置情報ＤＢ３２５を所定領域に備える。
【００８１】
システム情報３２１は、ＳＩＤごとに、そのＳＩＤのストリームデータの送信先を示した情報である。ここでの送信先は、例えば、データ蓄積装置２０の識別情報（ＩＰアドレス等）およびそのデータ蓄積装置２０における追記部２２１，２２２の識別情報（ポート番号等）を示した情報（表１参照）である。このシステム情報３２１は、接続先割当部３１が接続先の割り当てを行うときに参照される。
【００８２】
【表１】
0004733721.tif 000002
【００８３】
図４は、図１の情報源情報およびスキーマ情報を例示した図である。
【００８４】
情報源情報（ＳＩＤ情報）３２２は、当該ＳＩＤの定義情報であり、図４の符号４０１に示すようにＳＩＤと、当該ＳＩＤを持つストリームデータのスキーマ情報の識別情報（ＳＣＨＥＭＡ）と、このストリームデータに用いるフィルタの識別情報（ＦＩＬＴＥＲ）、このストリームデータのwindow幅（ＷＩＮＤＯＷ＿ＳＩＺＥ）等を示した情報である。図４の符号４０１に示す情報源情報は、ＳＩＤ「ＤＭＬ００１」の定義情報であり、このストリームデータのスキーマは「ＤＭＬ００１＿ＳＣＨＥＭＡ」に定義されるスキーマであり、このストリームデータに用いるフィルタは「ＸｐａｔｈＦｉｌｔｅｒ（“ｕｐｄａｔｅ”）」であり、このストリームデータのwindow幅は「１」であることを示す。
【００８５】
図３のスキーマ情報３２３は、当該ＳＩＤを持つストリームデータのスキーマの定義情報である。図４の符号４０２に示すスキーマ情報は、「ＤＭＬ００１＿ＳＣＨＥＭＡ」、つまり、ＳＩＤ「ＤＭＬ００１」のスキーマ情報であり、このストリームデータのスキーマは、「ｃｏｌ＿ｔ」は「ｔｉｍｅｓｔａｍｐ」であり、「ｃｏｌ＿ｉｄ」は「ｉｎｔ」であり、「ｃｏｌ＿ｄ」は「ｘｍｌ」であることを示す。
【００８６】
図３の処理手順情報３２４は、ＳＩＤごとに、そのＳＩＤのストリームデータに対しフィルタ処理装置１０が実行すべき処理を示した情報である。例えば、この処理手順情報３２４には、「ＳＩＤ＝１のストリームデータに、ＳＩＤ＝２のストリームデータを合流」といった内容が記載される、より詳細には「（１）ＳＩＤ＝１のデータに対しフィルタ処理１を実行し、（２）ＳＩＤ＝２のストリームデータの最新データを受信（参照）し、（３）このＳＩＤ＝１のデータとＳＩＤ＝２のデータそれぞれから所定カラムを選択して結合し、（４）この結合したデータに対し、フィルタ処理２を実行し、（５）参照装置６０Ａへの出力処理を行う」といった内容が記載される。この処理手順情報３２４は、図示しない入力部経由で書き換え可能である。
【００８７】
情報源位置情報ＤＢ３２５は、当該ＳＩＤごとに、当該ＳＩＤのデータが蓄積されるデータ蓄積装置２０の追記部２２１，２２２のＩＰアドレスとポート（ポート番号）とを示した情報源位置情報を蓄積したものである。例えば、この情報源位置情報ＤＢ３２５には「ＳＩＤ＝１のストリームデータは、ＩＰアドレス＝ＸＸ、ポート＝ＹＹの追記部に蓄積される」という情報源位置情報が含まれる。この情報源位置情報は、データ蓄積装置２０の追記部（追記部２２１，２２２等）から送信されたものをデータ位置管理部３３が情報源位置情報ＤＢ３２５に登録する。そして、データ位置管理部３３がフィルタ処理装置１０の処理部１２からの情報源位置情報の取得要求を受け付けたとき、この取得要求に含まれるＳＩＤに対応する情報源位置情報を情報源位置情報ＤＢ３２５から検索して送信する。
【００８８】
接続先割当部３１は、システム情報３２１を参照して、追記クライアント５０に接続先となるデータ蓄積装置２０の追記部２２１，２２２を指示する。
【００８９】
データ位置管理部３３は、データ蓄積装置２０の追記部（例えば、追記部２２１）から、情報源位置情報を受信すると、この情報源位置情報を記憶部３２の情報源位置情報ＤＢ３２５に登録する。また、このデータ位置管理部３３は、フィルタ処理装置１０の処理部１２から、当該ＳＩＤのストリームデータの情報源位置情報の問い合わせを受けたとき、このＳＩＤをキーとして当該ＳＩＤのデータの情報源位置情報ＤＢ３２５を検索し、検索した情報源位置情報を、問い合わせ元であるフィルタ処理装置１０の処理部１２へ送信する。
【００９０】
情報管理部３４は、記憶部３２に記憶されるストリームデータに必要なフィルタ処理機能を実現するためのファイル（図示省略）等を、フィルタ処理装置１０へ送信したり、記憶部３２内の各情報の書き換えをしたりする。例えば、この情報管理部３４は、ストリームデータ処理システム１内の各フィルタ処理装置１０（図１参照）のフィルタ処理機能を管理し、各フィルタ処理装置１０のフィルタ処理機能が同じものとなるよう、必要なファイルを各フィルタ処理装置１０へ送信するようにしてもよい。例えば、ストリームデータ処理システム１に、新たなフィルタ処理装置１０が追加されたとき、この情報管理装置３０から、各フィルタ処理装置１０と同じ機能を実現するためのファイルを送信するようにしてもよい。このようにすることで、ストリームデータ処理システム１のフィルタ処理装置１０のスケールアウトが容易になる。
【００９１】
＜動作手順＞
次に、ストリームデータ処理システム１の動作手順を説明する。まず、図１〜図３を参照しつつ、図５を用いて、データ蓄積装置２０の追記部２２１，２２２から送信された情報源位置情報の登録手順を説明する。図５は、図１のストリームデータ処理システムにおける情報源位置情報の登録手順を示したシーケンス図である。ここでは、データ蓄積装置２０Ｂの追記部２２１が、追記クライアント５０Ｂからのデータを受信し、データ位置管理部３３がこのデータの情報源位置情報を情報管理装置３０に登録する場合を例に説明する。
【００９２】
まず、追記クライアント５０Ｂは、情報源からデータを受信すると、このデータにＳＩＤを付し、情報管理装置３０の接続先割当部３１へ、当該ＳＩＤのデータの追記先払出し依頼を送信する（Ｓ１１）。この追記先払出し依頼を受信した接続先割当部３１は、システム情報３２１（図３参照）を参照して、当該ＳＩＤのデータの追記先となるデータ蓄積装置２０Ｂの追記部２２１を選択する。そして、接続先割当部３１は、この選択したデータ蓄積装置２０Ｂの追記部２２１のＩＰアドレスとポートとを追記先として追記クライアント５０Ｂへ送信する（Ｓ１２：追記先払出し）。
【００９３】
次に、データ蓄積装置２０Ｂの追記部２２１は、追記クライアント５０Ｂからの接続を受け付ける（Ｓ１３）。そして、この追記部２２１のストリーム受付部２２３（図１参照）は、追記クライアント５０Ｂからのデータの追記（送信）を検知すると（Ｓ１４のＹｅｓ）、位置情報送信部２２８は、このデータの情報源位置情報を情報管理装置３０へ送信する（Ｓ１５）。この情報源位置情報は、この追記部２２１に追記されたデータのＳＩＤと、この追記部２２１のＩＰアドレスとポートとを示した情報である。なお、追記クライアント５０Ｂからのデータの追記（送信）を検知しないとき（Ｓ１４のＮｏ）、位置情報送信部２２８は、そのままの状態でデータの追記を待つ。
【００９４】
情報管理装置３０のデータ位置管理部３３（図１参照）は、追記部２２１から送信された情報源位置情報を情報源位置情報ＤＢ３２５に登録する（Ｓ１６）。また、データ蓄積装置２０Ｂの追記部２２１におけるキュー管理部２２５（図１参照）は、追記クライアント５０Ｂから受信したデータをキュー部２２４へ出力する（Ｓ１７）。つまり、受信したデータをキュー部２２４に蓄積する。さらに、最新データ管理部２２６は、受信したデータを、その受信時刻とともに最新データ記憶部２２７へ出力する（Ｓ１８）。つまり、受信したデータを最新データ記憶部２２７に蓄積する。Ｓ１７とＳ１８の順は、前後してもよい。そして、追記部２２１のストリーム受付部２２３（図１参照）において追記クライアント５０Ｂからのデータの追記（送信）を検知すると（Ｓ１９のＹｅｓ）、Ｓ１７へ戻る。なお、追記クライアント５０Ｂからのデータの追記がないときは（Ｓ１９のＮｏ）、キュー管理部２２５は、そのままの状態でデータの追記を待つ。
【００９５】
追記部２２１は、このＳ１７〜Ｓ１９の処理を、追記クライアント５０Ｂの接続が切れるまで繰り返す。
【００９６】
このようにして、情報管理装置３０は、データ蓄積装置２０それぞれの追記部２２１，２２２から当該追記部２２１，２２２に追記されたデータの情報源位置情報を取得し、情報源位置情報ＤＢ３２５に登録する。
【００９７】
次に、図１〜図３を参照しつつ、図６を用いてストリームデータ処理システム１におけるストリームデータ合流処理の手順について説明する。図６は、図１のストリームデータ処理におけるストリームデータ合流処理の手順を示したシーケンス図である。ここでは、フィルタ処理装置１０が、データ蓄積装置２０Ａの追記部２２１に蓄積されるデータおよびデータ蓄積装置２０Ｂの追記部２２１に蓄積されるデータを合流させる場合を例に説明する。
【００９８】
まず、フィルタ処理装置１０の処理部１２は、キュー選択部１２１（図２参照）によりデータの取得先となる追記部２２１を選択し、データ取得部１２２は、この選択された追記部２２１に対しデータ取得要求を送信する（Ｓ２１）。このような取得要求を受け付けたデータ蓄積装置２０は、この取得要求の対象であるデータをキュー部２２４から読み出し、処理部１２のデータ取得部１２２へ出力する。これにより、データ取得部１２２は、追記部２２１から取得要求の対象であるデータを取得する（Ｓ２２）。
【００９９】
フィルタ処理装置１０の処理部１２のフィルタ実行制御部１２３（図２参照）は、この取得したデータに付されたＳＩＤと、情報管理装置３０の情報源情報３２２、スキーマ情報３２３および処理手順情報３２４（図３参照）等の定義情報を参照して、このデータに対し実行する処理内容を決定する（Ｓ２３）。そして、このフィルタ実行制御１２３は、この決定した処理内容を実行するため、このフィルタ処理装置１０の各部を制御する。ここで、この決定した処理内容がストリームデータの合流処理であった場合（Ｓ２４のＹｅｓ）、フィルタ実行制御部１２３は、処理手順情報３２４から、この受信したデータに合流させるストリームデータ（データ）のＳＩＤを読み出す（Ｓ２５）。つまり、フィルタ実行制御部１２３は、参照するデータのＳＩＤを調べる。一方、Ｓ２３において決定した処理内容がストリームデータの合流処理でなかった場合（Ｓ２４のＮｏ）、Ｓ２５、Ｓ３１をスキップしてＳ３２へ進む。Ｓ３２については後記する。
【０１００】
フィルタ処理装置１０の処理部１２は、情報管理装置３０に対し、Ｓ２５で読み出したＳＩＤの情報源位置情報の取得要求を送信する（Ｓ２６）。この情報源位置情報の取得要求を受信した情報管理装置３０のデータ位置管理部３３は、受信した情報源位置情報に示されるＳＩＤをキーとして、情報源位置情報ＤＢ３２５から、このＳＩＤのデータの情報源位置情報（このデータを持つ追記部２２１のＩＰアドレスとポート）を検索する。データ位置管理部３３は、この検索した情報源位置情報を処理部１２へ送信する（Ｓ２８）。
【０１０１】
そして、この情報源位置情報を受信した処理部１２の最新データ取得部１２７（図２参照）は、この情報源位置情報に示されるＩＰアドレスとポートを持つ追記部２２１へ当該ＳＩＤのデータの参照要求を送信する（Ｓ２９）。そして、この追記部２２１に蓄積される当該ＳＩＤのデータ（最新データ）の参照処理を行う（Ｓ３０）。
【０１０２】
フィルタ処理装置１０の処理部１２の選択結合部１２５（図２参照）は、Ｓ２２でデータ蓄積装置２０Ａの追記部２２１から取得したデータおよびＳ３０でデータ蓄積装置２０Ｂの追記部２２１から取得したデータそれぞれの所定カラムの選択結合を行う（Ｓ３１）。そして、フィルタ処理部１２４は、この結合したデータに対し、フィルタ処理を行い、この処理後のデータを参照装置６０へ出力する（Ｓ３３：データ出力）。この後、参照装置６０において、フィルタ処理装置１０から出力されたデータの整列を行い（Ｓ３４）、参照クライアント７０へ出力する（Ｓ３５：データ出力）。
【０１０３】
このようにすることで、ストリームデータ処理システム１は、異種ストリームデータの合流処理を行うことができる。
【０１０４】
次に、図１〜図３を参照しつつ、図７を用いて、データ蓄積装置２０が、フィルタ処理装置１０からのデータ取得要求に基づき、キュー部２２４に蓄積されるデータを送信する手順を詳細に説明する。図７は、図１のデータ蓄積装置が、フィルタ処理装置からのデータ取得要求に基づき、キュー部に蓄積されるデータを送信する手順を示したシーケンス図である。ここでは、データ蓄積装置２０Ａの追記部２２１が、フィルタ処理装置１０の処理部１２からのデータ取得要求に基づきデータを送信する場合を例に説明する。
【０１０５】
まず、データ蓄積装置２０Ａの追記部２２１は、フィルタ処理装置１０の処理部１２からデータ取得要求を受信する（Ｓ４１）。そして、この追記部２２１のキュー管理部２２５（図１参照）は、キュー部２２４からデータを取り出し、この取り出したデータに仕掛り中のフラグがあるか否かを判断する（Ｓ４２）。ここで、キュー管理部２２５が、この取り出したデータに仕掛り中のフラグはないと判断したとき（Ｓ４２のＮｏ）、この取り出したデータのコピーを、フィルタ処理装置１０の処理部１２へ送信する（Ｓ４３）。
【０１０６】
一方、キュー管理部２２５が、この取り出したデータに仕掛り中のフラグがあると判断したとき（Ｓ４２のＹｅｓ）、このデータを処理部１２へ送信しない（Ｓ４８）。そして、追記部２２１は、フィルタ処理装置１０の処理部１２からのデータ取得要求を待つ。
【０１０７】
Ｓ４３の後、追記部２２１のキュー管理部２２５は、キュー部２２４に蓄積されたデータのうち、処理部１２へ送信したデータに仕掛り中のフラグを付す（Ｓ４４）。そして、キュー管理部２２５は、この仕掛り中のフラグを付したデータについて、このフラグを付してから所定時間以内に、このデータの送信先である処理部１２から、このデータの処理完了通知を受信したか否かを判断する（Ｓ４５）。
【０１０８】
ここで、キュー管理部２２５は、この仕掛り中のフラグを付したデータについて、所定時間以内に、このデータの送信先である処理部１２から、このデータの処理完了通知を受信したとき（Ｓ４５のＹｅｓ）、当該データをキュー部２２４から削除する（Ｓ４６）。そして、処理部１２からのデータ取得要求（Ｓ４１）を待つ。つまり、キュー管理部２２５は、フィルタ処理装置１０の処理部１２へ送信したデータについて、この追記部１２において確かに処理完了したことを確認してからデータを削除する。
【０１０９】
一方、キュー管理部２２５は、この仕掛り中のフラグを付したデータについて、所定時間経過しても、処理部１２から、このデータの処理完了通知を受信しなかったとき（Ｓ４５のＮｏ）、当該データに付された仕掛り中のフラグをはずす（Ｓ４７）。そして、キュー管理部２２５は、このデータについて、処理部１２からのデータ取得要求（Ｓ４１）を待つ。
【０１１０】
つまり、キュー管理部２２５は、自身が送信したデータについて処理部１２において確かに処理が完了したことを確認するまで、そのデータをキュー部２２４に保持しておく。そして、キュー管理部２２５は、追記部１２において確かに処理完了したことを確認したとき、このデータを削除する。一方、キュー管理部２２５は、所定時間待っても、処理部１２から当該データの処理完了通知を受信できなかったときは、このデータについて仕掛り中のフラグをはずし、他の処理部１２からのデータ取得に応じるようにする。つまり、所定時間待っても、処理部１２から処理完了通知を受信できなかったということは、この処理部１２において当該データの処理が完了できなった可能性がある。よって、このデータを別の処理部１２が取得できるようにする。
【０１１１】
このようにすることで、例えば、フィルタ処理装置１０（または、その一部の処理部１２）のデータの処理中に故障が発生した場合でも、このデータについて、他のフィルタ処理装置１０や処理部１２が取得して処理を行うので、サービスが停止することがなくなる。
【０１１２】
なお、このフィルタ処理装置１０はそれぞれ、取得したストリームデータに対し必要なフィルタ処理機能が自身のフィルタ処理装置１０にないとき、フィルタ処理装置１０自身がファイル取得部１２８により、情報管理装置３０から、必要なフィルタ処理機能を取得するようにしてもよい。
【０１１３】
≪実施の形態の例≫
次に、図１〜図７を参照して、ストリームデータ処理システム１の実施の形態の例を示す。
【０１１４】
＜実施の形態の例１＞
ここでは、ストリームデータ処理システム１が、位置センサから得られる地図上の座標位置のストリームに、映像センサから得られるMotion  jpegの映像ストリームを組み合わせる場合を説明する。ここでは、映像センサからのデータ（ストリームデータ）をメインストリームとし、位置センサからのデータ（ストリームデータ）をサブストリームとする。この場合、情報管理装置３０の情報源情報３２２、スキーマ情報３２３および処理手順情報３２４（図３参照）に記憶される各センサの定義情報は、例えば、以下のようになる。
【０１１５】
位置センサのＳＩＤ：
location02
location02のスキーマ定義：
double  x;
double  y;
location02の処理定義：
begin
dropRecord（）; 
end-begin
【０１１６】
映像センサのＳＩＤ：
webcam02
webcam02のスキーマ定義：
jpeg  img;
double  x;
double  y;
webcam02の処理定義：
begin
getRecentRecord（location02）select  in1，out1，out2;
emitter(192.168.100.1:10000)；
end-begin
【０１１７】
この例では、図１に例示したストリームデータ処理システム１のデータ蓄積装置２０Ａの追記部２２１は、位置センサからのdouble型のデータｘとデータｙとを受信し続け、そのデータをキュー部２２４に蓄積する。また、追記部２２１は、このデータを最新データ記憶部２２７にも蓄積する。さらに、この位置センサのデータをデータ取得部１２２により受信したフィルタ処理装置１０の処理部１２は、このデータをデータ破棄部１２６により破棄（dropRecord（））する。つまり、所定の処理部１２の最新データ取得部１２７により、位置センサのデータが参照され、映像センサのデータと合流されれば、この位置センサのデータは不要であるので、当該データを取得した処理部１２はこのデータを破棄する。
【０１１８】
また、データ蓄積装置２０Ａの追記部２２１はjpeg型のデータimgを受信し、このデータ蓄積装置２０Ａのそのデータをキュー部２２４に蓄積する。また、このデータを最新データ記憶部２２７にも蓄積する。さらに、この映像センサのデータを受信したフィルタ処理装置１０の処理部１２の最新データ取得部１２７は、getRecentRecord（location02）によって、このlocation02のデータを蓄積する追記部２２１，２２２に対し、このlocation02の最新データとその受信時刻（timestamp）の参照処理を行う。そして、その最新データのカラムが、x，y，timestampという３つのカラムで構成される場合には、選択結合部１２５が、select  in1，out1，out2;によって、imgカラム、xカラム、yカラムから構成されるデータを構成し、このデータを後続のフィルタ処理に渡す。
【０１１９】
フィルタ実行制御部１２３は、emitter(192.168.100.1:10000)；により、imgカラム、xカラム、yカラムから構成されるデータを、出力部１３経由でＩＰアドレス192.168.100.1、ポート番号10000の装置（例えば、参照装置６０）へ送信し、この装置において、jpeg画像を座標（x，y）に表示させる。
【０１２０】
このようにすることで、ストリームデータ処理システム１は、位置センサから得られる地図上の座標位置に、映像センサから得られるMotion  jpegの映像ストリームを表示させることができる。
【０１２１】
＜実施の形態の例２＞
なお、ストリームデータ処理システム１は、ストリームデータの合流処理において条件付き処理を実行するようにしてもよい。つまり、処理手順情報３２４（図３参照）に、ストリームデータの合流処理を実行するときの条件文（if文）を含めるようにしてもよい。ここでは、ストリームデータ処理システム１が、対象物に対する電波の反射応答時間を数値として出し続ける人感センサからのデータを受信し、この人感センサの反射反応時間が変化している場合（つまり対象物が動いている場合）に、Motion  jpegの映像ストリームを出し続ける映像センサのデータを取り込んで、記録または表示する場合を説明する。なお、この例では、ストリームデータ処理システム１は、人感センサからのデータ（ストリームデータ）をメインストリームとし、映像センサからのストリームデータ（データ）をサブストリームのデータとして処理するものとする。この場合における各センサの定義情報は、例えば、以下のようになる。
【０１２２】
人感センサのＳＩＤの定義：
beamsensor01
beamsensor01のwindow幅の定義：
2
beamsensor01のスキーマ定義：
double  val；
beamsensor01の処理定義：
begin
beamsensor_diff() select  in1，out1;
bindcolums val，res; 
if res==1 then
getRecentRecord（webcam01）select  out1;
emitter(192.168.100.1:10000)；
end‐if
end‐begin
【０１２３】
映像センサのＳＩＤ：
webcam01
webcam01のwindow幅の定義：
1
webcam01のスキーマ定義：
jpeg  img;
webcam02の処理定義：
begin
dropRecord（）; 
end‐begin
とする。
【０１２４】
前記した定義情報に基づき、例えば、データ蓄積装置２０Ａの追記部２２１は、映像センサからのjpeg型のデータimgを受信し続け、そのデータをキュー部２２４に蓄積する。また、このデータを最新データ記憶部２２７にも蓄積する。さらに、この映像センサのデータを受信したフィルタ処理装置１０の処理部１２はデータ破棄部１２６により、データを破棄（dropRecord（））することを続ける。また、データ蓄積装置２０Ａの追記部２２１は、人感センサから、前記した反射反応時間をdouble型のvalデータとして受信し続ける。
【０１２５】
この人感センサからのデータを受信したフィルタ処理装置１０のフィルタ処理部１２４は、まず人感センサからのデータの差分の有無を確認するためのbeamsensor_diff() を実行する。ここで、人感センサからのデータのwindow幅の定義情報を活用する。このwindow幅とは、現在のデータに対して、過去の何件のデータを参照できるかを示した値である。ここでは人感センサからのデータのwindow幅を「２」としている。よって、フィルタ処理部１２４は、現在のデータと、その直前のデータとを参照することになり、反射応答時間の変化（beamsensor_diff()）を計算し、それが所定の閾値を超えていれば何らかの変化があったものと判断して、戻り値「１」を返すようにする。一方、反射応答時間の変化（beamsensor_diff() ）を計算し、それが所定の閾値を超えていなければ変化がないものと判断して、戻り値「０」を返す。
【０１２６】
beamsensor_diff() の後のselect  in1，out1は、選択結合部１２５がbeamsensor_diff()の入力の１カラム目であるin１と、出力の１カラム目であるout１とをあわせた２カラムを、後続の処理に渡すことを示す。bindcolums val，res; は、その２カラムをvalとresという名称で以下参照できるように定義したものである。
【０１２７】
if res==1 thenは、resカラムの値が「１」であれば、フィルタ処理部１２４がthenブロック内の処理を実行することを示す。ここで、res＝＝１とは、反射応答時間に変化があったと見なした（つまり人感センサが反応した）ということであり、res＝＝０であれば、人感センサが反応していないことを示す。ここで、人感センサが反応していない場合には、フィルタ処理部１２４は何もせず、そのデータに対する処理を終了し、次のデータの取得処理に入ることを示している。
【０１２８】
なお、人感センサが反応した場合には、getRecentRecord（webcam01）処理によって、フィルタ処理部１２４はwebcam01のデータを蓄積する追記部２２１，２２２から、このwebcam01の最新データとその受信時刻（timestamp）の参照処理を行う。そして、そのデータのカラムが、jpeg，timestampという２つのカラムで構成される場合には、選択結合部１２５が、select out1;によって、imgカラムだけを後続の処理に渡す。
【０１２９】
フィルタ実行制御部１２３は、emitter(192.168.100.1:10000)；により、imgカラムから構成されるデータを、出力部１３経由でＩＰアドレス192.168.100.1、ポート番号10000の装置（例えば、参照装置６０）へ送信し、この装置において、jpeg画像を記録または表示する。
【０１３０】
なお、ここでストリームデータ処理システム１は、人感センサの反応の有無により、jpeg画像の参照先となる映像センサをwebcam01，webcam02のいずれかに切り替えるようにしてもよい。この場合、webcam02の映像センサに関し、例えば、以下のような定義情報を追加する。
【０１３１】
映像センサのＳＩＤ：
webcam02
webcam02のwindow幅の定義：
2
webcam02のスキーマ定義：
jpeg  img;
webcam02の処理定義：
begin
dropRecord（）; 
end‐begin
【０１３２】
また、beamsensor01の処理定義に、以下の下線部のelse文を書き加える。
【０１３３】
begin
beamsensor_diff() select  in1，out1;
bindcolums val，res; 
if res==1 then
getRecentRecord（webcam01）select  out1;
emitter(192.168.100.1:10000)；
else
getRecentRecord（webcam02）select  out1;
emitter(192.168.100.1:10000)；
end‐if
end‐begin
【０１３４】
このようにすることでトリームデータ処理システム１は、人感センサの反応があるとき、webcam01のデータを参照し、人感センサの反応が無いとき、webcam02のデータを参照する処理を行うことができる。
【０１３５】
＜実施の形態の例３＞
また、ストリームデータ処理システム１は、ストリームデータ合流処理において、対象となるストリームデータをすべてメインストリームとして処理してもよい。ここでは、ストリームデータ処理システム１が、３つの位置センサＡ，Ｂ，Ｃからのデータを取得し、この３つの位置センサからのデータすべてをメインストリームとして処理する場合について説明する。この場合、ストリームデータ処理システム１は、その位置センサＡ，Ｂ，Ｃからのデータに基づき、その位置センサＡ，Ｂ，Ｃそれぞれにより測定された対象物（例えば、対象物a，b，c）間の距離を計算する。そして、その距離が所定の閾値（例えば、ユークリッド距離で「10.0」）以内に近づいたか否かを判断する場合を例に説明する。ここで、位置センサＡ，Ｂ，ＣのＳＩＤをそれぞれ、location11、location12、location13とする。ここでストリームデータ処理システム１は、（１）対象物ａと、対象物ｂ，ｃそれぞれとの距離、（２）対象物ｂと、対象物ａ，ｃそれぞれとの距離、（３）対象物ｃと、対象物ａ，ｂそれぞれとの距離が所定の閾値（ユークリッド距離で「10.0」）以内になったときhttpget（）を実行する。このときの各センサの定義情報は、例えば、以下のようになる。
【０１３６】
位置センサＡのＳＩＤの定義：
location11
location11のスキーマ定義：
double  x1;
double  y1;
location11の処理定義：
begin
getRecentRecord（location12）select  in1，in2，out1，out2;
bindcolums（x1，y1，x2，y2）;
if（sqrt((x1−x2)＊(x1−x2)+(y1−y2)＊(y1−y2))<10.0）then
httpget（http：//www.example.com/aaa.cgi?alert=on&src=11&dst=12);
end‐if
getRecentRecord（location13）select  in1，in2，out1，out2;
bindcolums（x1，y1，x3，y3）;
if（sqrt((x1−x3)＊(x1−x3)+(y1−y3)＊(y1−y3))<10.0）then
httpget（http：//www.example.com/aaa.cgi?alert=on&src=11&dst=13);
end‐if
end‐begin
【０１３７】
位置センサＢのＳＩＤの定義：
location12
location12のスキーマ定義：
double  x2;
double  y2;
location12の処理定義：
begin
getRecentRecord（location11）select  in1，in2，out1，out2;
bindcolums（x2，y2，x1，y1）;
if（sqrt((x2−x1)＊(x2−x1)+(y2−y1)＊(y2−y1))<10.0）then
httpget（http：//www.example.com/aaa.cgi?alert=on&src=12&dst=11);
end‐if
getRecentRecord（location13）select  in1，in2，out1，out2;
bindcolums（x2，y2，x3，y3）;
if（sqrt((x2−x3)＊(x2−x3)+(y2−y3)＊(y2−y3))<10.0）then
httpget(http：//www.example.com/aaa.cgi?alert=on&src=12&dst=13);
end‐if
end‐begin
【０１３８】
位置センサＣのＳＩＤの定義：
location13
location13のスキーマ定義：
double  x3;
double  y3;
location13の処理定義：
begin
getRecentRecord（location11）select  in1，in2，out1，out2;
bindcolums（x3，y3，x1，y1）;
if（sqrt((x3−x1)＊(x3−x1)+(y3−y1)＊(y3−y1))<10.0）then
httpget（http：//www.example.com/aaa.cgi?alert=on&src=13&dst=11);
end‐if
getRecentRecord（location12）select  in1，in2，out1，out2;
bindcolums（x3，y3，x1，y1）;
if（sqrt((x3−x2)＊(x3−x2)+(y3−y2)＊(y3−y2))<10.0）then
httpget（http：//www.example.com/aaa.cgi?alert=on&src=13&dst=12);
end‐if
end‐begin
【０１３９】
このような定義情報に基づき、例えば、位置センサＡからのデータを受信したデータ蓄積装置２０Ａの追記部２２１は、double型の２カラムからなるデータ（x1，y1）を受信し、そのデータをキュー部２２４に蓄積する。また、このデータを最新データ記憶部２２７にも蓄積する。このような処理を位置センサＢ，Ｃも実行する。
【０１４０】
この位置センサＡ，Ｂ，Ｃいずれかのデータを受信したフィルタ処理部１２４は、この位置センサ以外の２つの位置センサの最新データを、そのデータを蓄積する追記部から取得する。そして、フィルタ処理部１２４は、そのデータに示される位置座標のユークリッド距離が「10.0」よりも小さかった場合に、httpget（）を実行することで、所定の警報を出力する。
【０１４１】
なお、前記した実施の形態の例２において、ストリームデータ処理システム１は、当初サブストリームとして処理していた映像センサからのデータをメインストリームとして処理するよう変更してもよい。例えば、ストリームデータ処理システム１において、映像センサ（webcam01）からのデータについて、location02の位置センサからのデータと合流させたい場合、前記したwebcam01の映像センサからの定義情報に、以下の情報を追加すればよい。
【０１４２】
webcam01の処理定義：
begin
getRecentRecord( location02)  select  in1， out1， out2;
emitter( 192.168.100.1:10000);
end‐begin
【０１４３】
このような定義情報の設定により、ストリームデータ処理システム１は様々なストリームデータの合流処理を実行できる。なお、このようにストリームデータの合流処理について、その定義情報が変更された場合でも、図１のフィルタ処理装置１０の処理部１２はそれぞれ、この変更された定義情報を参照して処理を行う。よって、定義情報が変更された場合でもサービスを停止することなく処理を実行できる。
【０１４４】
＜適用例＞
なお、前記したストリームデータ処理システム１は、様々なサービスに適用可能である。ここでは、ストリームデータ処理システム１を子供の見守りサービスに適用した場合について説明する。この場合、ストリームデータ処理システム１は、各子供が携帯する携帯電話機から時々刻々送信されるＧＰＳ位置情報と、その子供の母親から指定されたエリア情報、警察から指定されたエリア情報とを取得する。そして、ストリームデータ処理システム１は、これらの情報をもとに、母親から指定されたエリア以外に子供がいると判断したとき、その母親に警報を出力する。また、警察から指定されたエリア（危険エリア）に子供がいると判断したとき、母親に警報を出力する。
【０１４５】
このときに処理の流れを、図８を用いて説明する。図８は、図１のストリームデータ処理システムの適用例を説明した図である。図８において、ストリームデータ処理システム１の構成のうち、データ蓄積装置２０の追記部２２１、追記部２２２およびフィルタ処理装置１０の処理部１２以外は図示を省略している。
【０１４６】
例えば、図１の追記部２２２Ａは、子供Ｃ１のＧＰＳ等位置情報を受信し、追記部２２１は、エリア情報として母親Ｍ１（子供Ｃ１の母親）の指定エリア７１を受信する。また、追記部２２２Ｂは、子供Ｃ２のＧＰＳ等位置情報を受信し、追記部２２１Ｂは、エリア情報として母親Ｍ２（子供Ｃ２の母親）の指定エリア７３を受信する。さらに、追記部２２１Ｃは、警察（または地域団体や安全監視会社等）からエリア情報として警察指定マップ（危険エリア）７５を受信する。
【０１４７】
そして、処理部１２Ａは、処理手順情報３２４等を参照して以下の処理を実行する。
【０１４８】
１．追記部２２２Ａ，２２２Ｂから、子供（子供Ｃ１，Ｃ２）の位置（位置情報）を取得する。
２．追記部２２１Ｂ，２２２Ｂから、母親（母親Ｍ１，Ｍ２）の指定エリアを取得する。
３．子供の位置と、この子供の母親の指定エリアとの関係を演算し、子供の位置が母親の指定エリアからはみ出しそうなら子供（例えば、子供の携帯電話機等）に警報を出力する。また、はみ出したら母親（例えば、母親の持つ端末装置等）に警報を出力する。
４．追記部２２１Ｃから警察の指定エリア（警察指定マップ）を取得する。
５．子供の位置と、警察の指定エリアとの関係を演算し、この指定エリアに近づいたら子供に警報を出力し、この指定エリアに侵入したら母親に警報を出力する。
６．危険エリアと、母親の指定エリアと、子供の位置とを重ね合わせたエリア地図を母親へ送信する。
【０１４９】
処理部１２Ａが以上のような処理を行うことで、例えば、母親Ｍ１の端末装置等に、符号７２に示す監視画面を表示させたり、母親Ｍ２の端末装置等へ符号７４に示す監視画面を表示させたりすることができる。つまり、母親側から問い合わせて子供の位置を把握するのではなく、母親の端末装置上に子供の位置を表示し続けることができるので、子供の監視を容易なものとすることができる。
【０１５０】
また、前記した例では、指定エリアは固定されたものとして説明したが、処理部１２が時間帯（例えば、日没等）によって指定エリアを変更するようにしてもよい。さらに、米国のように犯罪者がＧＰＳを保有する場合には、犯罪者との距離が所定値より近くなったら警報を出すようにしてもよい。また、この処理部１２から送信される画面上に地域のイベント等の情報を含めて表示させるようにしてもよいし、子供の位置情報が消えたら警報を出力するようにしてもよい。
【０１５１】
なお、このような見守りサービスにおいて、指定エリアの形状が複雑なものになることも多い。また、指定エリアも、日々変更したい場合もあり、警報の出力先も母親、子供、警察等で変更したい場合もある。このような場合、情報管理装置３０の定義情報（処理手順情報３２４等、図３参照）に変更を加えればよいので、前記した変更にも柔軟に対応することができる。
【０１５２】
また、子供の監視エリアは、母親が指定した指定エリアの他に、地域の団体や警察、安全監視会社等が指定した危険エリア（例えば、工事、地盤不安定、交通量の増大、事故等が発生したエリア）も考えられる。ここで、このような危険エリアの情報をストリームデータ処理システム１へ送信すれば、ストリームデータ処理システム１は、これらのエリアや種別に応じて警報を出力することもできる。
【０１５３】
本実施の形態のストリームデータ処理システム１は、前記したとおりフィルタ処理装置１０の追加によりスケールアウト可能である。よって、ユーザの増大に応じてシステムの規模を徐々に大きくしていくことも可能となり、前記したような見守りサービス等、多数のユーザを抱えるサービスへの適用効果は大きいと考えられる。また、ストリームデータ処理システム１は、処理内容を途中で変更したり拡張したりした場合でも、サービスが停止することはないので高品質なサービスを提供できる。
【０１５４】
なお、前記した説明では、ストリームデータ処理システム１がストリームデータの合流処理を行う契機は、主に、所定のＳＩＤのデータが到着したこととしたが、これに限定されない。例えば、ストリームデータ処理システム１は、外部から与えられるタイマ・イベントを契機としてもよい。また、所定のＳＩＤを持ち、かつデータのタイムスタンプに示される時区間が所定の時間内のデータに受信したことを契機としてもよい。このようなストリームデータの合流の契機や、その合流対象のデータについては、前記した情報管理装置３０の処理手順情報３２４等に定義される。
【０１５５】
また、図１の各データ蓄積装置２０は、追記部２２１，２２２それぞれで受信したストリームデータを蓄積する蓄積データ記憶部を備え、このフィルタ処理装置１０の処理部１２がデータの再フィルタ処理の指示を受け付けたとき、この蓄積データ記憶部から、再フィルタ処理の対象となるデータを取得して再フィルタ処理を行うようにしてもよい。この場合も、データ蓄積装置２０は、蓄積データ記憶部にデータを蓄積したとき、この蓄積したデータのＳＩＤと、自身のデータ蓄積装置２０の識別情報（ＩＰアドレス）や、その蓄積データ記憶部における、当該データの記憶領域を示す位置情報を情報管理装置３０へ送信しておく。そして、フィルタ処理装置１０の処理部１２がデータの再フィルタ処理の指示を受け付けたとき、この再フィルタの対象となるデータの位置情報を情報管理装置３０に問い合わせ、その応答結果に基づき、データ蓄積装置２０から再フィルタ処理を行う。このようにすることで、例えば、いったんフィルタ処理装置１０の処理部１２がデータのフィルタ処理を行った後、再度そのデータについて別のフィルタをするとき等に大変便利である。
【０１５６】
本実施の形態に係るデータ蓄積装置２０、フィルタ処理装置１０および情報管理装置３０は、前記したような処理を実行させるプログラムによって実現することができ、そのプログラムをコンピュータによる読み取り可能な記憶媒体（ＣＤ−ＲＯＭ等）に記憶して提供することが可能である。また、そのプログラムを、インターネット等のネットワークを通して提供することも可能である。
【図面の簡単な説明】
【０１５７】
【図１】本実施の形態のストリームデータ処理システムを含むシステムの構成例を示した図である。
【図２】図１のフィルタ処理装置の機能ブロック図である。
【図３】図１の情報管理装置の機能ブロック図である。
【図４】図１の情報源情報およびスキーマ情報を例示した図である。
【図５】図１のストリームデータ処理システムにおける情報源位置情報の登録手順を示したシーケンス図である。
【図６】図１のストリームデータ処理におけるストリームデータの合流手順を示したシーケンス図である。
【図７】図１のデータ蓄積装置が、フィルタ処理装置からのデータ取得要求に基づき、キュー部に蓄積されるデータを送信する手順を示したシーケンス図である。
【図８】図１のストリームデータ処理システムの適用例を説明した図である
【符号の説明】
【０１５８】
１    ストリームデータ処理システム
１０（１０Ａ，１０Ｂ）  フィルタ処理装置
１１    入力部
１２（１２Ａ，１２Ｂ）  処理部
１３    出力部
２０（２０Ａ，２０Ｂ）  データ蓄積装置
２６    プロセス管理部
３０    情報管理装置
３１    接続先割当部
３２    記憶部
３３    データ位置管理部
３４    情報管理部
５０（５０Ａ，５０Ｂ，…，５０Ｎ）  追記クライアント
６０（６０Ａ，６０Ｂ，…，６０Ｎ）  参照装置
６１    ＤＢ（データベース）
６２    入力部
６３    検索部
６４    整列部
６５    出力部
６６    整列バッファ
７０（７０Ａ，…，７０Ｎ）  参照クライアント
１２１  キュー選択部
１２２  データ取得部
１２３  フィルタ実行制御部
１２４  フィルタ処理部
１２５  選択結合部
１２６  データ破棄部
１２７  最新データ取得部
１２８  ファイル取得部
２２１，２２２  追記部
２２３  ストリーム受付部
２２４  キュー部
２２５  キュー管理部
２２６  最新データ管理部
２２７  最新データ記憶部
２２８  位置情報送信部
３２１  システム情報
３２２  情報源情報
３２３  スキーマ情報
３２４  処理手順情報
３２５  情報源位置情報ＤＢ（データベース）
【図１】
0004733721.tif 000003
【図２】
0004733721.tif 000004
【図３】
0004733721.tif 000005
【図４】
0004733721.tif 000006
【図５】
0004733721.tif 000007
【図６】
0004733721.tif 000008
【図７】
0004733721.tif 000009
【図８】
0004733721.tif 000010