(57)【特許請求の範囲】
【請求項１】
複数のフィルタ処理装置によりストリームデータのフィルタ処理を行うストリームデータ管理システムに用いられる前記フィルタ処理装置であって、
各種データの入力を受け付ける入力部と、
複数の情報源より送信されたストリームデータを蓄積する１以上のデータ蓄積装置それぞれから、前記入力部経由で、当該データ蓄積装置におけるキュー負荷の値を示したキュー負荷情報を受信し、この受信したキュー負荷情報をキュー負荷テーブルへ格納するキュー負荷テーブル更新部と、
前記データ蓄積装置それぞれのキュー負荷を示した前記キュー負荷テーブルおよび前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐを記憶する記憶部と、
前記キュー負荷テーブルに示されるデータ蓄積装置のうち、そのキュー負荷の値が大きいデータ蓄積装置ほど高確率で選択し、選択した順に前記データ蓄積装置の識別情報を取得先リストに記録し、前記選択したデータ蓄積装置の識別情報を記録した取得先リストを前記記憶部に格納する取得先リスト作成部と、
前記取得先リストに示される順に前記ストリームデータの取得先となる前記データ蓄積装置を選択する選択処理部と、
前記選択されたデータ蓄積装置から、前記ストリームデータを取得するデータ取得処理部と、
前記取得したストリームデータに対しフィルタ処理を実行するフィルタ処理部と、
前記フィルタ処理後のストリームデータを出力する出力部と、
前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐと、前記キュー負荷テーブルに示されるデータ蓄積装置それぞれのキュー負荷の合計値Ｓとの差（Ｐ−Ｓ）が、所定の閾値Ｐ×ε（但し、εの値は、０．２〜０．３）を超えるとき、前記データ取得処理部に前記データ蓄積装置からストリームデータを取得させないスリープ時間の合計値が、（Ｐ−Ｓ）×（０．７〜０．８）に相当する時間の前記スリープ時間を示したスリープ情報を、前記取得先リストに設定するスリープ情報追加部とを備え、
前記選択処理部が、前記取得先リストから前記スリープ情報を読み出したとき、
前記データ取得処理部は、
前記読み出したスリープ情報に示されるスリープ時間が経過するまで前記ストリームデータを取得しないことを特徴とするフィルタ処理装置。
【請求項２】
前記キュー負荷情報は、
当該データ蓄積装置における（１）入力スループットと、（２）このデータ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間とを含むことを特徴とする請求項１に記載のフィルタ処理装置。
【請求項３】
前記スリープ情報追加部は、
前記取得先リストに前記スリープ情報を設定するとき、前記スリープ時間の合計値を、前記取得先リストに設定されるデータ取得の回数で割った値を、前記取得先リスト上で前記取得先となるデータ蓄積装置の識別情報それぞれの後に設定することを特徴とする請求項１または請求項２に記載のフィルタ処理装置。
【請求項４】
複数の情報源より送信されたストリームデータを蓄積する１以上のデータ蓄積装置それぞれが、
フィルタ処理装置へ前記ストリームデータを送出したとき、その時刻を記録し、前記フィルタ処理装置において当該データのフィルタ処理を完了すると、その旨の通知を受信し、当該フィルタ処理装置にストリームデータを送出してから前記通知を受信するまでの時間の平均値を計算し、この計算した平均値を前記送出したデータ数で割って、前記データ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間として記録するステップと、
前記データ蓄積装置における（１）入力スループットおよび（２）前記データ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間の少なくとも一方が、所定の閾値以上に変化したとき、前記フィルタ処理装置へ、前記データ蓄積装置におけるキュー負荷の値を示したキュー負荷情報を送信するステップとを実行し、
前記フィルタ処理装置が、
前記データ蓄積装置それぞれから、入力部経由で、当該データ蓄積装置におけるキュー負荷情報を受信し、この受信したキュー負荷情報を記憶部のキュー負荷テーブルへ格納するステップと、
前記キュー負荷テーブルに示されるデータ蓄積装置のうち、そのキュー負荷の値が大きいデータ蓄積装置ほど高確率で選択し、この選択したデータ蓄積装置の識別情報を順次取得先リストに記録し、前記選択したデータ蓄積装置の識別情報を記録した取得先リストを前記記憶部に格納するステップと、
前記取得先リストに示される順に前記ストリームデータの取得先となる前記データ蓄積装置を選択するステップと、
前記選択されたデータ蓄積装置から、前記ストリームデータを取得するステップと、
前記取得したストリームデータに対しフィルタ処理を実行するステップと、
前記フィルタ処理後のストリームデータを出力部経由で出力するステップとを実行し、
前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐと、前記キュー負荷テーブルに示されるデータ蓄積装置それぞれのキュー負荷の合計値Ｓとの差（Ｐ−Ｓ）が、所定の閾値Ｐ×ε（但し、εの値は、０．２〜０．３）を超えるとき、前記データ蓄積装置からストリームデータを取得させないスリープ時間の合計値が、前記（Ｐ−Ｓ）×（０．７〜０．８）に相当する時間となる前記スリープ時間を示したスリープ情報を、前記取得先リストに設定し、
前記取得先リストから前記スリープ情報を読み出したとき、前記読み出したスリープ情報に示されるスリープ時間が経過するまで前記ストリームデータを取得しないことを特徴とするフィルタ処理方法。
【請求項５】
前記キュー負荷情報は、
当該データ蓄積装置における（１）入力スループットと、（２）このデータ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間とを含むことを特徴とする請求項４に記載のフィルタ処理方法。
【請求項６】
前記フィルタ処理装置が、
前記取得先リストに前記スリープ情報を設定するとき、前記スリープ時間の合計値を、前記取得先リストに設定されるデータ取得の回数で割った値を、前記取得先リスト上で前記取得先となるデータ蓄積装置の識別情報それぞれの後に設定することを特徴とする請求項４または請求項５に記載のフィルタ処理方法。
【請求項７】
コンピュータを、請求項１ないし請求項３のいずれか１項に記載のフィルタ処理装置として機能させるためのフィルタ処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ストリームデータ処理技術に関する。
【背景技術】
【０００２】
従来、各種センサやＩＣタグ、携帯電話等で計測されたデータをストリームデータとして受信し、このストリームデータにフィルタ処理を行い、所定の利用先へ流通させるストリームデータ処理技術がある。このようなストリームデータ処理技術として、telegraphCQ（非特許文献１参照）や、borealis（非特許文献２参照）が提案されている。
【非特許文献１】Sirish Chandrasekaran,Owen Cooper,AmolDeshpande,Michael J.Franklin,JosephM. Hellerstein,Wei Hong,SaileshKrishnamurthy,Sam Madden,VijayshankarRaman,Fred Reiss,and Mehul Shah,“Te1egraphCQ:Continuous Dataflow Processing for an Uncertain World,” CIDR(2003)
【非特許文献２】D.J.Abadi,Y.Ahmad,M.Balazinska,U.Cetintemel,M.Cherniack,J.H.Hwang,W.Lindner,A.S.Maskey,A.Rasin,E.Ryvkina,N.Tatbul,Y.Xing,and S.tan Zdonik, “The Design of the Borealis Stream Processing Engine,”CIDR(2005)
【発明の開示】
【発明が解決しようとする課題】
【０００３】
前記したtelegraphCQは、入力されるデータの性質変化に合わせてストリームデータ処理を実行するものであるが、処理を実行するのは単一サーバである。つまり、前記したtelegraphCQにおいて、複数のサーバでストリームデータ処理を実行することについては考慮されていない。
【０００４】
また、前記したborealisは、複数のサーバでストリームデータ処理を実行するものであるが、各サーバの最適配置が決定した後に、ストリームデータ処理のマイグレーション処理（負荷の大きいサーバから、負荷の小さいサーバへオペレーションを移行する処理）を行う必要がある。しかし、各サーバにおいてこのようなマイグレーション処理を実行すると、マイグレーション自体の負荷が大きくなり、特に、各種センサやＩＣタグ等、様々な情報源からデータ入力を受け付けたり、また、その情報源からのデータ入力量の変化が激しかったりする場合、システム全体の処理能力が下がる可能性がある。
【０００５】
そこで、本発明は前記した問題を解決し、様々な情報源からストリームデータを受信する場合であっても、システム全体の処理能力が低下しないようにするフィルタ処理装置等を提供することを目的とする。
【課題を解決するための手段】
【０００６】
前記した課題を解決するため、請求項１に記載の発明は、複数のフィルタ処理装置によりストリームデータのフィルタ処理を行うストリームデータ管理システムに用いられる前記フィルタ処理装置であって、各種データの入力を受け付ける入力部と、複数の情報源より送信されたストリームデータを蓄積する１以上のデータ蓄積装置それぞれから、前記入力部経由で、当該データ蓄積装置におけるキュー負荷の値を示したキュー負荷情報を受信し、この受信したキュー負荷情報をキュー負荷テーブルへ格納するキュー負荷テーブル更新部と、前記データ蓄積装置それぞれのキュー負荷を示した前記キュー負荷テーブルおよび前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐを記憶する記憶部と、前記キュー負荷テーブルに示されるデータ蓄積装置のうち、そのキュー負荷の値が大きいデータ蓄積装置ほど高確率で選択し、選択した順に前記データ蓄積装置の識別情報を取得先リストに記録し、前記選択したデータ蓄積装置の識別情報を記録した取得先リストを前記記憶部に格納する取得先リスト作成部と、前記取得先リストに示される順に前記ストリームデータの取得先となる前記データ蓄積装置を選択する選択処理部と、前記選択されたデータ蓄積装置から、前記ストリームデータを取得するデータ取得処理部と、前記取得したストリームデータに対しフィルタ処理を実行するフィルタ処理部と、前記フィルタ処理後のストリームデータを出力する出力部と、前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐと、前記キュー負荷テーブルに示されるデータ蓄積装置それぞれのキュー負荷の合計値Ｓとの差（Ｐ−Ｓ）が、所定の閾値Ｐ×ε（但し、εの値は、０．２〜０．３）を超えるとき、前記データ取得処理部に前記データ蓄積装置からストリームデータを取得させないスリープ時間の合計値が、（Ｐ−Ｓ）×（０．７〜０．８）に相当する時間となる前記スリープ時間を示したスリープ情報を、前記取得先リストに設定するスリープ情報追加部とを備え、前記選択処理部が、前記取得先リストから前記スリープ情報を読み出したとき、前記データ取得処理部は、前記読み出したスリープ情報に示されるスリープ時間が経過するまで前記ストリームデータを取得しない構成とした。
【０００７】
請求項４に記載の発明は、複数の情報源より送信されたストリームデータを蓄積する１以上のデータ蓄積装置それぞれが、フィルタ処理装置へ前記ストリームデータを送出したとき、その時刻を記録し、前記フィルタ処理装置において当該データのフィルタ処理を完了すると、その旨の通知を受信し、当該フィルタ処理装置にストリームデータを送出してから前記通知を受信するまでの時間の平均値を計算し、この計算した平均値を前記送出したデータ数で割って、前記データ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間として記録するステップと、前記データ蓄積装置における（１）入力スループットおよび（２）前記データ蓄積装置から送出されたストリームデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間の少なくとも一方が、所定の閾値以上に変化したとき、前記フィルタ処理装置へ、前記データ蓄積装置におけるキュー負荷の値を示したキュー負荷情報を送信するステップとを実行し、前記フィルタ処理装置が、前記データ蓄積装置それぞれから、入力部経由で、当該データ蓄積装置におけるキュー負荷情報を受信し、この受信したキュー負荷情報を記憶部のキュー負荷テーブルへ格納するステップと、前記キュー負荷テーブルに示されるデータ蓄積装置のうち、そのキュー負荷の値が大きいデータ蓄積装置ほど高確率で選択し、この選択したデータ蓄積装置の識別情報を順次取得先リストに記録し、前記選択したデータ蓄積装置の識別情報を記録した取得先リストを前記記憶部に格納するステップと、前記取得先リストに示される順に前記ストリームデータの取得先となる前記データ蓄積装置を選択するステップと、前記選択されたデータ蓄積装置から、前記ストリームデータを取得するステップと、前記取得したストリームデータに対しフィルタ処理を実行するステップと、前記フィルタ処理後のストリームデータを出力部経由で出力するステップとを実行し、前記フィルタ処理装置それぞれのフィルタ処理能力の合計値Ｐと、前記キュー負荷テーブルに示されるデータ蓄積装置それぞれのキュー負荷の合計値Ｓとの差（Ｐ−Ｓ）が、所定の閾値Ｐ×ε（但し、εの値は、０．２〜０．３）を超えるとき、前記データ蓄積装置からストリームデータを取得させないスリープ時間の合計値が、前記（Ｐ−Ｓ）×（０．７〜０．８）に相当する時間となる前記スリープ時間を示したスリープ情報を、前記取得先リストに設定し、前記取得先リストから前記スリープ情報を読み出したとき、前記読み出したスリープ情報に示されるスリープ時間が経過するまで前記ストリームデータを取得しない方法とした。
【０００８】
このようにすることで、ストリームデータ管理システムのフィルタ処理装置はキュー負荷の値が大きいデータ蓄積装置ほど高確率で選択して、取得先リストを作成し、その取得先リストに従いデータ蓄積装置からストリームデータを取得する。従って、フィルタ処理装置は、キュー負荷が大きいデータ蓄積装置から優先的にデータを取得することができる。よって、特定のデータ蓄積装置にキュー負荷が偏ることがなくなる。また、このように取得先リストを参照して、データ取得先となるデータ蓄積装置を選択するので、フィルタ処理装置それぞれでマイグレーションを行う必要がなくなり、システム全体の処理能力の低下を防止できる。また、各フィルタ処理装置のフィルタ処理能力の合計値の方が、各データ蓄積装置におけるキュー負荷の合計値を上回る場合に、フィルタ処理装置は適宜スリープ時間を入れながらデータ蓄積装置からストリームデータを取得する。これにより、フィルタ処理装置は、データ蓄積装置から必要以上にデータ取得を行うこと（空振り）を低減できるので、無駄な電気消費を抑制できる。さらに、このようにフィルタ処理装置において適宜スリープ時間を入れながらデータ蓄積装置からストリームデータを取得することで、データ蓄積装置において過剰にストリームデータの取得要求を受け付けることがなくなる。これによりデータ蓄積装置における入力データの受け付け能力の低下を防止できる。また、各データ蓄積装置におけるデータの平均待ち行列長を短くできる。
【０００９】
請求項２に記載の発明は、請求項１に記載のフィルタ処理装置において、前記キュー負荷情報が、当該データ蓄積装置における（１）入力スループットと、（２）このデータ蓄積装置から送出されたデータの前記フィルタ処理装置におけるデータ１個あたりの平均処理時間とを含む構成とした。
【００１０】
このようにすることで、フィルタ処理装置は、各データ蓄積手段におけるキュー負荷を正確に把握することができる。
【００１５】
請求項３に記載の発明は、請求項１または請求項２に記載の前記スリープ情報追加部が、前記取得先リストに前記スリープ情報を設定するとき、前記スリープ時間の合計値を、記取得先リストに設定されるデータ取得の回数で割った値を、前記取得先リスト上で前記取得先となるデータ蓄積装置の識別情報それぞれの後に設定する構成とした。
【００１６】
このようにすることで、フィルタ処置装置は、各データ蓄積装置からデータを取得するとき、スリープ時間を均等にして取得するので、特定のデータ蓄積装置にデータが溜まることを防止できる。
【００１９】
請求項７に記載の発明は、コンピュータを、請求項１ないし請求項３のいずれか１項に記載のフィルタ処理装置として機能させるためのフィルタ処理プログラムとした。
【００２０】
このようなプログラムによれば、一般的なコンピュータを請求項１ないし請求項３のいずれか１項に記載のフィルタ処理装置として機能させることができる。
【発明の効果】
【００２１】
本発明によれば、ストリームデータ管理システムにおいて、情報源のデータ入力量の変化が激しい場合であっても、システム全体の処理能力を著しく低下させないようにすることができる。
【発明を実施するための最良の形態】
【００２２】
以下、本発明を実施するための最良の形態（以下、実施の形態という）を、第１の実施の形態および第２の実施の形態に分けて説明する。
【００２３】
≪第１の実施の形態≫
図１は、本実施の形態のストリームデータ管理システムを含むシステムの構成例を示した図である。図１に示すように、システムは、様々な情報源から取得したデータにサービスＩＤ（ＳＩＤ）を付し、このデータをストリームデータとして送信するクライアント５０（５０Ａ，５０Ｂ）と、受信したストリームデータのフィルタ処理を行うストリームデータ管理システム１と、フィルタ処理されたデータの出力先である、出力先６０（６０Ａ〜６０Ｎ）と、この出力先６０からデータの検索を行う検索クライアント７０（７０Ａ〜７０Ｎ）とを含んで構成される。
【００２４】
ストリームデータ管理システム１は、クライアント５０へ、ストリームデータの送信先となるデータ蓄積手段（装置）２０を指示するキュー割当手段４０と、クライアント５０から送信されたストリームデータを蓄積するデータ蓄積手段（装置）２０（２０Ａ，２０Ｂ，…，２０Ｎ）と、ストリームデータのフィルタ処理を行うフィルタ処理手段（装置）１０（１０Ａ〜１０Ｎ）と、このフィルタ処理手段１０におけるフィルタ処理部１２４（詳細は後記）の機能を実現するファイルデータ等を記憶する共有情報管理手段（装置）３０とを含んで構成される。
【００２５】
また、出力先６０は、ストリームデータ管理システム１においてフィルタ処理されたデータをＤＢ（データベース）６１に蓄積する蓄積処理部６２と、フィルタ処理されたデータを記憶するＤＢ６１と、検索クライアント７０からの要求に応じて、このＤＢ６１から所定のデータを検索する検索処理部６３とを含んで構成される。
【００２６】
（１）クライアント５０とストリームデータ管理システム１との間、（２）ストリームデータ管理システム１と出力先６０との間はそれぞれ、図示しない伝送経路により接続される。さらに、データ蓄積手段２０、フィルタ処理手段１０、出力先６０の数は、図１において、それぞれＮ個としたが、これに限定されない。なお、フィルタ処理手段１０はそれぞれ同じ機能を備えるものとする。
【００２７】
＜動作概要＞
ここで、図１のシステムの動作概要を説明する。まず、クライアント５０は、様々な情報源からデータを取得し、そのデータに応じたＳＩＤを付する。このＳＩＤはＳＩＤのバージョンの情報を含んでいてもよい。そして、クライアント５０は、このＳＩＤを付したデータをストリームデータとして、ストリームデータ管理システム１へ送信する。ストリームデータ管理システム１のキュー割当手段４０は、ストリームデータの送信先であるデータ蓄積手段２０を選択する。そして、クライアント５０に対し、ストリームデータの送信先のデータ蓄積手段２０を指示し、クライアント５０は、指示されたデータ蓄積手段２０にストリームデータを送信する。なお、データ蓄積手段２０はそれぞれ、自身のキュー負荷に所定の閾値以上の変化があったとき、そのキュー負荷の値を示したキュー負荷情報を含むキュー負荷情報変更通知を各フィルタ処理手段１０へ通知する。そして、キュー負荷テーブル更新部１２６はキュー負荷情報を、各データ蓄積手段２０におけるキュー負荷テーブル１４１に反映する。
【００２８】
続いて、フィルタ処理手段１０は、取得先リスト１４２を作成する。この取得先リスト１４２はストリームデータの取得先となるデータ蓄積手段２０を時系列で示した情報であえる。例えば、フィルタ処理手段１０は、データ蓄積手段２０（２０Ａ〜２０Ｎ）から、ランダム値を用いて、データ取得先となるデータ蓄積手段２０を順に選択する。このとき、フィルタ処理手段１０は、キュー負荷テーブル１４１に示される各データ蓄積手段２０のキュー負荷の値をもとに、キュー負荷が高いデータ蓄積手段２０ほど優先的に選択するようにする。例えば、データ蓄積手段２０Ａにおけるキュー負荷の値が「１０」であり、データ蓄積手段２０Ｂ,２０Ｃのキュー負荷の値がそれぞれ「５」であった場合、フィルタ処理手段１０は、１０：５：５の確率で選択するようにして、この選択したデータ蓄積手段２０の識別情報を取得先リスト１４２に順次書きこんでいく。このようにしてフィルタ処理手段１０は取得先リスト１４２を作成する。そして、フィルタ処理手段１０は、この作成した取得先リスト１４２に示される順にデータ蓄積手段２０からデータを取得する。そして、フィルタ処理手段１０は、この決定したデータ蓄積手段２０からストリームデータを取得し、フィルタ処理を行う。各ストリームデータに対し行うべきフィルタ処理は、このストリームデータに付されたＳＩＤに基づき判断する。なお、ここで、このストリームデータに必要なフィルタ処理機能が自身のフィルタ処理部１２４になかった場合、フィルタ処理手段１０は、共有情報管理手段３０からファイルデータを取得して、フィルタ処理機能の追加を行うようにしてもよい。
【００２９】
フィルタ処理手段１０は、フィルタ処理後、ストリームデータを所定の出力先６０へ出力する。そして、出力先６０においてフィルタ処理後のデータをＤＢ６１に蓄積し、このＤＢ６１のデータは、検索クライアント７０により検索可能となる。
【００３０】
また、このクライアント５０、キュー割当手段４０、データ蓄積手段２０、共有情報管理手段３０、フィルタ処理手段１０、出力先６０および検索クライアント７０は、入出力インタフェース、ＣＰＵ（Central Processing Unit）等の演算処理部、ＲＡＭ（Random Access Memory）、ＲＯＭ（Read Only Memory）、ＨＤＤ（Hard Disk Drive）等の記憶部等を備えるコンピュータにより実現される。また、各処理部は、ＣＰＵによるプログラム実行処理あるいは専用回路により実現される。
【００３１】
また、データ蓄積手段２０（２０Ａ〜２０Ｎ）はそれぞれ別個のコンピュータ（装置）により実現するようにしてもよいし、これらをまとめて１つのコンピュータにより実現するようにしてもよい。さらに、フィルタ処理手段１０（１０Ａ〜１０Ｎ）についても、それぞれ別個のコンピュータ（装置）により実現するようにしてもよいし、これらをまとめて１つのコンピュータにより実現するようにしてもよい。つまり、データ蓄積手段２０（２０Ａ〜２０Ｎ）およびフィルタ処理手段１０（１０Ａ〜１０Ｎ）をそれぞれ、プロセスにより実現するようにしてもよいし、コンピュータ（マシン）により実現するようにしてもよい。さらに、キュー割当手段４０は、ロードバランサをさらに備え、このロードバランサにより各データ蓄積手段２０へのストリームデータの割り当てを行うようにしてもよい。
【００３２】
＜データ蓄積手段＞
次に、図２を用いてデータ蓄積手段２０を詳細に説明する。図２は、図１のデータ蓄積手段の詳細を示した機能ブロック図である。
【００３３】
データ蓄積手段２０は、クライアント５０から送信されたストリームデータを一時的に蓄積する手段である。このようなデータ蓄積手段２０は、図２に示すように、クライアント５０からのストリームデータの入力を受け付ける入力部２１と、記憶部２４におけるキュー２４１の状態を監視し、所定の閾値以上にキュー負荷の変化があったとき、このデータ蓄積手段２０におけるキュー負荷情報（入力スループット、フィルタ処理手段１０におけるデータ１個あたりの平均処理時間、このデータ蓄積手段２０に溜まっているデータの個数等）を送信する処理部２２と、ストリームデータおよび前記したキュー負荷情報を出力するための出力部２３と、蓄積したストリームデータのキュー２４１が形成される記憶部２４とを含んで構成される。
【００３４】
なお、この入力部２１および出力部２３は、クライアント５０やフィルタ処理手段１０との間でデータ入出力可能な入出力インタフェースにより実現され、記憶部２４は、ＲＡＭ、ＨＤＤ等により実現される。また、処理部２２は、ＣＰＵによる記憶部２４に記憶されるプログラムの実行処理、専用回路等により実現される。なお、記憶部２４において、キュー２４１として蓄積されたストリームデータは、フィルタ処理手段１０からの要求に応じて、フィルタ処理手段１０へ送信される。
【００３５】
処理部２２は、キュー負荷監視部２２１と、キュー負荷情報送信部２２２とを含んで構成される。キュー負荷監視部２２１は、自身のデータ蓄積手段２０におけるキュー負荷を監視する。そして、キュー負荷監視部２２１は、所定の閾値（共有情報管理手段３０のキュー負荷情報通知閾値３２１）以上に、キュー負荷が変化したか否かを監視する
【００３６】
キュー負荷情報送信部２２２は、キュー負荷監視部２２１においてキュー負荷が共有情報管理手段３０のキュー負荷情報通知閾値３２１に示される閾値以上に変化したとき、このデータ蓄積手段２０におけるキュー負荷情報を各フィルタ処理手段１０へ送信する。例えば、キュー負荷情報のうち、入力スループット、フィルタ処理手段１０におけるデータ１個あたりの平均処理時間およびこのデータ蓄積手段２０に溜まっているデータの個数のうち少なくともいずれか１つがキュー負荷情報通知閾値３２１に示される値以上に変化したとき、キュー負荷情報を各フィルタ処理手段１０へ送信する。このときのキュー負荷情報には、自身のデータ蓄積手段２０の識別情報が付される。また、キュー負荷情報の送信は、例えば、各フィルタ処理手段１０への同報送信等により行われる。
【００３７】
このようにすることで、データ蓄積手段２０は、自身のキュー負荷に変化があったとき、自身のキュー負荷をいち早く各フィルタ処理手段１０へ通知することができる。なお、キュー負荷情報通知閾値３２１は、このデータ蓄積手段２０が共有情報管理手段３０からコピーして記憶部２４に記憶しておき、キュー負荷監視部２２１は、この記憶部２４に記憶されたキュー負荷情報通知閾値３２１を参照して、キュー負荷の監視を行うようにしてもよい。
【００３８】
＜フィルタ処理手段＞
図１に戻ってフィルタ処理手段１０を詳細に説明する。フィルタ処理手段１０は、図１に示すように、ストリームデータの入力を受け付ける入力部１１と、ストリームデータをデータ蓄積手段２０から取得し、フィルタ処理を行う処理部１２と、フィルタ処理を行ったデータを出力する出力部１３と、このフィルタ処理手段１０の機能を実現するためのプログラムや、各データ蓄積手段２０のキュー負荷を示したキュー負荷テーブル１４１、データの取得先となるデータ蓄積手段２０を時系列で示した取得先リスト１４２を記憶する記憶部１４とを含んで構成される。この入力部１１および出力部１３も、データ蓄積手段２０や出力先６０との間でデータ入出力可能な入出力インタフェースにより実現され、記憶部１４は、ＲＡＭ、ＨＤＤ等により実現される。また、処理部１２は、ＣＰＵによる記憶部１４に記憶されるプログラムの実行処理、専用回路等により実現される。
【００３９】
この処理部１２は、選択処理部１２１と、データ取得処理部１２２と、フィルタ実行制御部１２３と、フィルタ処理部１２４と、取得先リスト作成部１２５と、キュー負荷テーブル更新部１２６とを含んで構成される。スリープ情報追加部１２７については、別途第２の実施の形態で説明する。
【００４０】
選択処理部１２１は取得先リスト１４２に示される順にデータ蓄積手段２０を選択し、この選択したデータ蓄積手段２０をデータ取得先としてデータ取得処理部１２２へ指示する。
【００４１】
データ取得処理部１２２は、選択処理部１２１により指示されたデータ蓄積手段２０からストリームデータを取得する。なお、ストリームデータを取得するタイミングは、例えば、自身のフィルタ処理手段１０において処理しているデータがなくなったときでもよいし、処理中のデータが所定の閾値以下となったときでもよい。
【００４２】
フィルタ実行制御部１２３は、データ取得処理部１２２から取得したストリームデータに付されたＳＩＤと、共有情報管理手段３０のＳＩＤテーブル（図示省略）とを参照して、このストリームデータに対し実行すべきフィルタ処理（フィルタオペレーションの適用手順）を決定する。そして、フィルタ実行制御部１２３は、記憶部１４からフィルタオペレーション（ＦＯＰ）を読み出し、ＦＯＰにより記述されるフィルタ処理をフィルタ処理部１２４に実行させる。
【００４３】
なお、このフィルタ実行制御部１２３は、このストリームデータに対し実行すべきフィルタ処理機能が自身のフィルタ処理手段１０にないとき、共有情報管理手段３０から、このフィルタ処理機能を実現するためのファイルデータ（フィルタオペレーションを含む）をダウンロードし、機能追加を行うようにしてもよい。
【００４４】
フィルタ処理部１２４は、フィルタ実行制御部１２３からの指示に基づき、ストリームデータに対しフィルタ処理を実行する。このフィルタ処理部１２４におけるフィルタ処理の内容はフィルタオペレーション（ＦＯＰ）により記述され、フィルタ処理部１２４は、このＦＯＰに基づきフィルタ処理を実行する。このＦＯＰは例えば「画像フォーマットをＪＰＥＧ（Joint Photographic Experts Group）から、Ｂｉｔｍａｐフォーマットに変換する」というものや、「画像サイズを特徴抽出ルーチンに合うように正規化する」というもの等である。また、このＦＯＰは、「処理後のデータを出力先６０Ａに出力する」等、処理後のデータの出力先に関する情報を含んでいてもよい。このＦＯＰは、記憶部１４に記憶され、フィルタ実行制御部１２３により読み出される。
【００４５】
取得先リスト作成部１２５は、キュー負荷テーブル１４１をもとに取得先リスト１４２を作成する。このとき、取得先リスト作成部１２５は、キュー負荷テーブル１４１に示される各データ蓄積手段２０のうち、キュー負荷が高いデータ蓄積手段２０を高い確率で選択するようにする。そして、このような選択を所定回数（例えば、Ｍ回）繰り返し、選択した順に、この選択したデータ蓄積手段２０の識別情報を取得先リスト１４２に記録する。例えば、データ蓄積手段２０Ａにおけるキュー負荷の値が「１０」であり、データ蓄積手段２０Ｂ,２０Ｃのキュー負荷の値がそれぞれ「５」であった場合、フィルタ処理手段１０は、１０：５：５の確率で選択するようにして、この選択したデータ蓄積手段２０の識別情報を取得先リスト１４２に順次書きこんでいく。このときの取得先リスト１４２の作成手順の具体例は、図３を用いて後記する。なお、このようにして作成された取得先リスト１４２は、選択処理部１２１により、データ取得先のデータ蓄積手段２０を選択する際に参照される。
【００４６】
キュー負荷テーブル更新部１２６は、データ蓄積手段２０からキュー負荷情報変更通知を受信すると、この通知に含まれるキュー負荷情報を記憶部１４に格納する。そして、キュー負荷テーブル更新部１２６は、このキュー負荷情報をもとに各データ蓄積手段２０のキュー負荷ｃｉを計算し、その計算した負荷ｃｉをキュー負荷テーブル１４１に書き込む。
【００４７】
記憶部１４は、前記したＦＯＰ、このフィルタ処理手段１０の機能を実現するプログラム（図示省略）、キュー負荷テーブル１４１、取得先リスト１４２等を記憶する。なお、取得先リスト１４２ａについては、別途第２の実施の形態で説明する。このキュー負荷テーブル１４１および取得先リスト１４２を、図３を参照しながら説明する。図３は、図１のキュー負荷テーブルに基づく取得先リストの作成を概念的に説明した図である。
【００４８】
このキュー負荷テーブル１４１は、図３に示すように、データ蓄積手段２０ごとのキュー負荷（負荷）ｃｉ（データ蓄積手段ｉで計測されたキュー負荷）を示した情報である。例えば、図３に示すキュー負荷テーブル１４１において、ＩＤ「１」のデータ蓄積手段２０のキュー負荷は「１０」であることを示す。このキュー負荷ｃｉは以下の式（１）により定義される。
ｃｉ＝λｉｘｉ＋ｐｉｘｉ…式（１）
λｉ：単位時間あたりの平均入力データ数（入力スループット）
ｘｉ：データ１個あたりの平均処理時間（ＣＰＵコスト）
ｐｉ：データ蓄積手段に溜まっているデータ数
従って、これらのΣｃｉは、図３の式（２）のように定義される。なお、データ１個あたりの平均処理時間ｘｉは、データ蓄積手段２０がフィルタ処理手段１０へデータを渡した後に、そのデータが処理されるまでの時間である。つまり、データ蓄積手段２０は、フィルタ処理手段１０へデータを送出したとき、その送出時刻を記録しておく。そして、フィルタ処理手段１０において当該データのフィルタ処理を完了すると、その旨の通知をこのフィルタ処理手段１０から受信するので、データ蓄積手段２０は、フィルタ処理手段１０にデータを渡してからこの通知を受信するまでの時間の平均値を、データ数で割ってデータ１個あたりの平均処理時間ｘｉとする。また、前記した式（１）においてｐｉｘｉを含めないようにしてもよい。
【００４９】
また、取得先リスト１４２は、図３に示すように、データの取得先となるデータ蓄積手段２０の識別情報（ＩＤ）を取得順に示した情報である。例えば、図３に示す取得先リスト１４２において、データ取得処理部１２２は、ＩＤ「１」→「１」→「２」→「１」→「１」→…→「１」という順にデータ蓄積手段２０からデータを取得することを示す。
【００５０】
ここで、キュー負荷テーブル１４１を参照した取得先リスト１４２の作成手順を、図３を用いて説明する。（１）まず、図１の取得先リスト作成部１２５は、キュー負荷テーブル１４１における総負荷Ｓを求める。ここでの総負荷Ｓは、図３に例示したキュー負荷テーブル１４１に示されるｃｉの値から、総負荷Ｓ＝２０となる。（２）次に、取得先リスト作成部１２５は、１〜２０（総負荷Ｓ）までの間のランダム値ｒを生成する。（３）そして、取得先リスト作成部１２５は、前記した（２）で生成したランダム値ｒをもとに、データを取得するデータ蓄積手段２０を決定（選択）する。例えば、キュー負荷テーブル１４１に示されるＩＤ「１」、「２」、「３」のデータ蓄積手段２０のキュー負荷ｃｉはそれぞれ「１０」、「５」、「５」である。従って、総負荷Ｓの値「２０」を１０：５：５に分割し、（２）で生成したランダム値ｒが１≦ｒ≦１０を満たすものであれば、ＩＤ「１」のデータ蓄積手段２０を選択し、ランダム値ｒが１１≦ｒ≦１５を満たすものであれば、ＩＤ「２」のデータ蓄積手段２０を選択し、ランダム値ｒが１６≦ｒ≦２０を満たすものであれば、ＩＤ「３」のデータ蓄積手段２０を選択する。
【００５１】
すなわち、データ蓄積手段２０の数が３個であった場合、ＩＤ「１」のデータ蓄積手段２０が選択される確率は５０％、ＩＤ「２」のデータ蓄積手段２０が選択される確率は２５％、ＩＤ「３」のデータ蓄積手段２０が選択される確率は２５％となるようにする。つまり、各フィルタ処理手段１０の取得先リスト１４２は、ストリームデータを取得するデータ蓄積手段２０を選択するとき、当該データ蓄積手段２０のキュー負荷の値に比例した確率でデータ蓄積手段２０を選択する。
【００５２】
（４）取得先リスト作成部１２５は、前記した（２）、（３）を繰り返すことで、取得先リスト１４２を作成する。この取得先リスト１４２は、前記したとおり、データの取得先であるデータ蓄積手段２０のキュー負荷の値に比例した確率でデータ蓄積手段２０を選択したものである。従って、この取得先リスト１４２に示されるデータ取得元の各データ蓄積手段２０のＩＤの登場回数は、ほぼ各データ蓄積手段２０のキュー負荷の値に比例した回数になる。よって、このような取得先リスト１４２に基づきデータ取得を行うデータ取得処理部１２２は、データ蓄積手段２０それぞれのキュー負荷の値に比例した頻度でデータ蓄積手段２０からデータ取得を行うことになる。
【００５３】
＜動作手順＞
次に、図１〜図３を参照しつつ図４を用いて、図１のフィルタ処理手段１０の動作手順を説明する。図４は、図１のフィルタ処理手段の動作を示したフローチャートである。
【００５４】
図１のフィルタ処理手段１０のキュー負荷テーブル更新部１２６は、データ蓄積手段２０からのキュー負荷情報変更通知の受信を待つ（Ｓ１０１）。そして、キュー負荷テーブル更新部１２６は、キュー負荷情報変更通知を受信すると（Ｓ１０１のＹｅｓ）、この通知に含まれるキュー負荷情報を記憶部１４に格納する。そして、キュー負荷テーブル更新部１２６は、このキュー負荷情報をもとに各データ蓄積手段２０のキュー負荷ｃｉを計算し、その計算した負荷ｃｉをもとにキュー負荷テーブル１４１（図３参照）を更新する（Ｓ１０２）。
【００５５】
そして、取得先リスト作成部１２５は、更新したキュー負荷テーブル１４１をもとに取得先リスト１４２（図３参照）を作成する（Ｓ１０３）。この取得先リスト１４２の作成手順については、後記する。そして、取得先リスト作成部１２５は、作成した取得先リスト１４２を記憶部１４に格納する。
【００５６】
次に、選択処理部１２１は、この取得先リスト１４２の位置番号カウンタＣに初期値「１」をセットする（Ｓ１０４）。そして、新たなキュー負荷情報変更通知を受信したか否かを判断し（Ｓ１０５）、キュー負荷情報変更通知を受信していないことを確認すると（Ｓ１０５のＮｏ）、選択処理部１２１は、位置番号カウンタＣが取得先リスト１４２の大きさ（登録されているデータ数Ｍ）を超えているか否かを判断する（Ｓ１０６）。ここで、位置番号カウンタＣが取得先リスト１４２の大きさを超えていれば（Ｓ１０６のＹｅｓ）、選択処理部１２１は、位置番号カウンタＣを初期値「１」に戻し（Ｓ１０７）、Ｓ１０８へ進む。なお、Ｓ１０５でキュー負荷情報変更通知を受信していたとき（Ｓ１０５のＹｅｓ）、キュー負荷テーブル１４１の変更が必要なので、Ｓ１０２へ戻る。
【００５７】
一方、位置番号カウンタＣが取得先リスト１４２の大きさを超えていなければ（Ｓ１０６のＮｏ）、選択処理部１２１はこの位置番号カウンタＣに対応するデータ蓄積手段２０のＩＤを取得先リスト１４２から読み取り（Ｓ１０８）、データ取得処理部１２２はこの読み取ったＩＤのデータ蓄積手段２０からデータ取得を行う。そして、フィルタ実行制御部１２３において取得したデータのフィルタ処理を実行する（Ｓ１０９）。そして、選択処理部１２１は、位置番号カウンタＣをインクリメントし（Ｓ１１０）、Ｓ１０５へ戻る。
【００５８】
このような処理を行うことで、フィルタ処理手段１０は、各データ蓄積手段２０の最新のキュー負荷を反映した取得先リスト１４２を作成する。従って、フィルタ処理手段１０は、この最新のキュー負荷を反映した取得先リスト１４２に従ってデータ蓄積手段２０からデータを取得することができる。
【００５９】
次に、図５を用いて、取得先リスト１４２の作成手順を説明する。図５は、図１の取得先リストの作成手順を示すフローチャートである。
【００６０】
まず、取得先リスト作成部１２５は、ループカウンタａに初期値「０」をセットする（Ｓ２０１）。そして、キュー負荷テーブル１４１を参照して、すべてのデータ蓄積手段２０における負荷ｃｉの和Ｓ（総負荷Ｓ）を計算する（Ｓ２０２）。
【００６１】
次に、取得先リスト作成部１２５は、ループカウンタａが取得先リスト１４２の大きさＭよりも大きければ（Ｓ２０３のＹｅｓ）、取得先リスト１４２を出力し（Ｓ２０８）、処理を終了する。
【００６２】
一方、ループカウンタａの値が取得先リスト１４２の大きさＭ以下であれば（Ｓ２０３のＮｏ）、取得先リスト作成部１２５は、１〜Ｓ（Ｓ２０２で計算したすべてのデータ蓄積手段２０における負荷ｃｉの和）の間の値で、ランダム値ｒを生成する（Ｓ２０４）。このときのランダム値ｒの生成は、公知の技術を用いてよい。
【００６３】
そして、取得先リスト作成部１２５は、［ｃ1（１番目のデータ蓄積手段２０における負荷）からｃｈ（ｈ番目のデータ蓄積手段２０における負荷）までの和］≦［ｒ］＜［ｃ1（１番目のデータ蓄積手段２０における負荷）からｃｈ＋１（ｈ＋１番目のデータ蓄積手段２０における負荷）までの和］を満たすｈを決定する（Ｓ２０５）。なお、ストリームデータ管理システム１におけるデータ蓄積手段２０の数がＮであるとき、ｈは、１≦ｈ≦Ｎを満たす値である。
【００６４】
次に、取得先リスト作成部１２５は、取得先リスト１４２の第ａ番目の要素として、ｈ番目のデータ蓄積手段２０のＩＤを取得先リスト１４２に格納する（Ｓ２０６）。そして、取得先リスト作成部１２５は、ループカウンタａをインクリメントして（Ｓ２０７）、Ｓ２０３へ戻る。
【００６５】
このようにして作成された、取得先リスト１４２に基づきデータ取得処理部１２２はデータ蓄積手段２０のキュー負荷が高ければ高い頻度で、データ蓄積手段２０のキュー負荷が低ければ低い頻度で、データ取得を行う。従って、特定のデータ蓄積手段２０だけがデータの待ち行列長が長くなるのを防止できるの。つまり、データ蓄積手段２０へのデータ入力量の変化が激しい場合であっても、各データ蓄積手段２０においてキュー負荷や待ち行列の長さに偏りがなくなるので、ストリームデータ管理システム１全体の処理能力を著しく低下させないようにすることができる。
【００６６】
≪第２の実施の形態≫
次に、本発明の第２の実施の形態を説明する。第２の実施の形態は、各フィルタ処理手段１０のフィルタ処理能力の方が、各データ蓄積手段２０におけるキュー負荷の合計を上回るとき、フィルタ処理手段１０は適宜スリープ時間を入れながらデータ蓄積手段２０からデータ取得を行うことを特徴とする。これにより、フィルタ処理手段１０は、データ蓄積手段２０から必要以上にデータ取得を行うこと（空振り）が無くなるので、無駄な電気消費を抑制できる。さらに、フィルタ処理手段１０は必要以上にデータ蓄積手段２０へストリームデータの取得要求を行わないので、データ蓄積手段２０の負荷を低減し、データ蓄積手段２０はクライアント５０からの入力データ受信も滞りなく行うことができる。
【００６７】
第２の実施の形態のフィルタ処理手段１０は、スリープ情報追加部１２７を含み、このスリープ情報追加部１２７は、取得先リスト１４２ａにスリープ時間を設定する。スリープ時間とは、データ取得処理部１２２にデータ蓄積手段２０からのデータ取得を行わせない時間である。図６は、第２の実施の形態の取得先リストを概念的に説明した図である。図６に示すように、取得先リスト１４２ａは、データ取得処理部１２２の取得先となるデータ蓄積手段２０のＩＤ以外に、このデータ取得処理部１２２にデータ取得を行わせないスリープを示す情報（スリープ情報）を含む。このスリープ情報は、データ取得処理部１２２にデータ取得を行わせない時間（スリープ時間）を含む。選択処理部１２１において、このスリープ情報を読み取ったとき、スリープ時間が経過するまでデータ取得処理部１２２にデータ取得を指示しない。
【００６８】
例えば、このような取得先リスト１４２ａに基づき、選択処理部１２１はデータ取得処理部１２２にＩＤ「１」のデータ蓄積手段２０からのデータ取得→スリープ→再度ＩＤ「１」のデータ蓄積手段２０からのデータ取得→スリープ…というように指示を行う。なお、このスリープ時間の合計は、図６に示すように、データ蓄積手段２０の総負荷をＳ、フィルタ処理手段１０の総処理能力をＰとしたとき、各フィルタ処理手段１０の総負荷を（Ｐ−Ｓ）に相当する時間まで削減する値とする。例えば、このようなスリープによる負荷削減率（Ｐ−Ｓ）／Ｐ＝Ｒとしたとき、各フィルタ処理手段１０におけるスリープ時間は、{Ｒ／（１−Ｒ）}Ｓに相当する時間とする。
【００６９】
次に、図７を用いて、スリープ情報追加部１２７の動作手順を説明する。図７は、第２の実施の形態のスリープ情報追加部の動作手順を示したフローチャートである。図７に示すように、まず、スリープ情報追加部１２７は、キュー負荷テーブル１４１から、各データ蓄積手段２０の総負荷Ｓを計算する（Ｓ３０１）。そして、（各フィルタ処理手段１０の総処理能力Ｐ−総負荷Ｓ）が所定の閾値（Ｐ×ε）より大きいか否かを判断する（Ｓ３０２）。なお、各フィルタ処理手段１０の総処理能力Ｐの値は、例えば、フィルタ処理手段１０に記憶部１４に格納してあるものものとする。この総処理能力Ｐの値は、共有情報管理手段３０において管理し、この共有情報管理手段３０が各フィルタ処理手段１０の総処理能力Ｐの値が変化したときに、フィルタ処理手段１０へ送信するようにしてもよい。また、εの値は、例えば０．２〜０．３程度とする。
【００７０】
ここで、（各フィルタ処理手段１０の総処理能力Ｐ−総負荷Ｓ）が所定の閾値（Ｐ×ε）より大きいとき（Ｓ３０２のＹｅｓ）、スリープ情報追加部１２７は、負荷削減率Ｒ（＝（Ｐ−Ｓ）／Ｐ）を計算する（Ｓ３０３）。そして、スリープ情報追加部１２７は、{Ｒ／（１−Ｒ）}Ｓのスリープを取得先リスト１４２ａへ追加する（Ｓ３０４）。なお、ここで取得先リスト１４２ａに追加するスリープは、{Ｒ／（１−Ｒ）}Ｓに０．７〜０．８をかけた値とするのが好ましい。この理由については後記する。
【００７１】
一方、（各フィルタ処理手段１０の総処理能力Ｐ−総負荷Ｓ）が所定の閾値（Ｐ×ε）より小さいときは（Ｓ３０２のＮｏ）、取得先リスト１４２ａへのスリープ情報の追加を行わず、処理を終了する。
【００７２】
なお、（各フィルタ処理手段１０の総処理能力Ｐ−総負荷Ｓ）の値が所定の閾値（Ｐ×ε）より大きいときに、取得先リスト１４２にスリープ時間を設定することにしたのは、（各フィルタ処理手段１０の総処理能力Ｐ−総負荷Ｓ）が小さければ空振りは殆ど発生せず、スリープ時間の設定は不要だからである。
【００７３】
さらに、取得先リスト１４２ａに設定するスリープ時間は、{Ｒ／（１−Ｒ）}Ｓに０．７〜０．８をかけた値とするのが好ましいのは、これは以下の理由による。図８は、データ処理における待ち行列長の原理を説明するために引用したグラフである。図８に示すように、ρ＝到着率（総負荷）／平均サービス量（総処理能力）としたとき、待ち行列長はほぼ１／（１−ρ）に近い値になることが知られている。つまり、ρ＝１近傍、総処理能力Ｐ＝総負荷Ｓにすると、待ち行列長は激増する。一方、ρ＝０．７〜０．８程度の値であれば、データ蓄積手段２０における待ち行列長の増加は抑えられる。つまり、スリープ時間を{Ｒ／（１−Ｒ）}Ｓ×（０．７〜０．８）とした方が、各データ蓄積手段２０における待ち行列長の増加が抑制され、かつ空振りも低減できることが期待できるからである。
【００７４】
また、取得先リスト１４２ａへのスリープ情報の設定は、例えば、図６に示すように、取得先リスト１４２ａ全体にわたり、まんべんなく設定するのが好ましい。例えば、スリープ時間の合計値を、取得先リスト１４２に設定される要素数（データ蓄積手段２０へのデータ取得の回数）で割った値を、取得先であるデータ蓄積装置２０のＩＤそれぞれの後に均等に設定する。これは、データ取得処理部１２２が各データ蓄積手段２０から何回かデータ取得を行った後、まとめてスリープ時間をとるよりも、データ取得の後にこまめにスリープ時間をとる方が、データ蓄積手段２０におけるデータ行列長の増加を抑制できるからである。
【００７５】
次に、第２の実施の形態のフィルタ処理手段１０の動作手順を説明する。図９は、第２の実施の形態のフィルタ処理手段の動作手順を示したフローチャートである。図９のＳ４０１,４０２は、図４のＳ１０１,１０２と同様である。なお、Ｓ４０３における取得先リスト１４２ａの作成は、前記したとおりスリープ情報の設定処理を含む。Ｓ４０４〜Ｓ４０８は、図４のＳ１０４〜Ｓ１０８と同様であるが、Ｓ４０８で読み取ったＩＤが「スリープ」であった場合（Ｓ４０９のＹｅｓ）、選択処理部１２１は、スリープ情報に設定されたスリープ時間が経過するまでデータ取得処理部１２２にデータ取得を指示しない。つまりスリープ情報を読み取った場合、データ取得処理部１２２は指定時間スリープする（Ｓ４１０）。そして、このスリープ情報に設定されたスリープ時間経過後、選択処理部１２１は、位置番号カウンタＣをインクリメントし（Ｓ４１２）、Ｓ４０５へ戻る。一方、Ｓ４０８で読み取ったＩＤが「スリープ」でなかった場合（Ｓ４０９のＮｏ）、図４のＳ１０９と同様に、選択処理部１２１はデータ取得処理部１２２に、この読み取ったＩＤを持つデータ蓄積手段２０からのデータ取得を指示する。そして、フィルタ実行制御部１２３において取得したデータのフィルタ処理を実行する。
【００７６】
このようにすることでフィルタ処理手段１０は、データ蓄積手段２０から必要以上にデータ取得を行うこと（空振り）が無くなるので、無駄な電気消費を抑制できる。さらに、データ蓄積手段２０において、フィルタ処理手段１０から必要以上にストリームデータの取得を受け付けることで負荷が増大し、データ蓄積手段２０がクライアント５０からの入力データを受け付けられないという事態を防止できる。
【００７７】
≪実験結果≫
次に、本実施の形態のストリームデータ管理システム１を用いた実験結果を説明する。
【００７８】
図１０は、第２の実施の形態のストリームデータ管理システムを用いた実験結果を示したグラフである。本実験において、比較例のストリームデータ管理システムのフィルタ処理手段は取得先リスト１４２を持たず、処理すべきデータが無くなった時点でデータ取得先となるデータ蓄積手段を選択するものとした。なお、この比較例におけるデータ蓄積手段の選択方法はラウンドロビンとした。また、データ蓄積手段の数を、それぞれ５０個とした場合と、２００個とした場合の、両方について実験を行った。グラフの横軸は、各データ蓄積手段の入力スループットの合計値を示し、縦軸は、各データ蓄積手段に必要となるフィルタ処理手段の数を示す。ここで、フィルタ処理手段の１個あたりの処理能力（ＣＰＵコスト）は、０．１（ｓｅｃ）とした。また、各データ蓄積手段に必要となるフィルタ処理手段の数とは、いずれか１つのデータ蓄積手段においてキュー長が「６００」を超えた時点のフィルタ処理手段の数とした。取得先リスト１４２の大きさＭは、「５０」とした。
【００７９】
図１０に示すように、本実験において比較例のストリームデータ管理システムは、データ蓄積手段の数が５０個の場合、入力スループットが「２００」を超えたあたりで、必要となるフィルタ処理手段の数が５０個をはるかに超えてしまっている。また、データ蓄積手段の数が２００個の場合、既に、入力スループットが「５」を超えたあたりで、必要となるフィルタ処理手段の数が５０個をはるかに超えてしまっている。つまり、比較例の場合、データ蓄積手段の数の増加にともない、少ない入力スループットであっても必要となるフィルタ処理手段の数が爆発的に増加し、システム全体の性能が大きく劣化することが分かる。
【００８０】
一方、本実施の形態のストリームデータ管理システム１によれば、データ蓄積手段２０の入力スループットが増えた場合でも、必要となるフィルタ処理手段１０の数は５０個までゆるやかな伸びを示す。つまり、本実施の形態のストリームデータ管理システム１によれば、データ蓄積手段２０の数が増加しても、システム全体の処理能力があまり劣化しないことが分かる。
【００８１】
すなわち、本実施の形態のストリームデータ管理システム１におけるフィルタ処理手段１０のように、キュー負荷の値が大きいデータ蓄積手段２０ほど高確率で選択して、データ取得先となるデータ蓄積手段２０を示した取得先リスト１４２に設定しておき、この取得先リスト１４２に示される順にデータ蓄積手段２０からデータ取得を行った方が、データ蓄積手段２０の入力スループットが増加した場合でも、システム全体の処理能力はあまり劣化しないことが分かる。
【００８２】
図１１は、第２の実施の形態のストリームデータ管理システムを用いた実験結果を示したグラフである。本実験において、ストリームデータ管理システム１で計算されたスリープ時間を変化させ、各データ蓄積手段２０の平均待ち行列長と、各フィルタ処理手段１０の空振り率と計測した。ここでのスリープ時間は、計算されたスリープ時間{Ｒ／（１−Ｒ）}Ｓに対する乗数を０〜１まで変化させたものとした。また、データ蓄積手段２０の数は、「３０」とし、フィルタ処理手段１０のＣＰＵコストは０．１（ｓｅｃ）とした。図１１に示すように、計算されたスリープ時間に対する乗数が「１」の場合、つまり、計算された{Ｒ／（１−Ｒ）}Ｓそのものである場合、空振り率は限りなく「０」に近づくが、平均待ち行列長が激増する。一方、計算されたスリープ時間に対する乗数が「０」に近づくにつれて、空振り率は増加する。従って、フィルタ処理手段１０の空振り率を低減しつつ、データ蓄積手段２０の平均待ち行列長の増加も抑制するには、スリープ時間{Ｒ／（１−Ｒ）}Ｓに対する乗数として「０．７〜０．８」を採用するのがよいことが分かる。
【００８３】
本実施の形態に係るフィルタ処理手段１０、データ蓄積手段２０、共有情報管理手段３０は、前記したような処理を実行させるプログラムによって実現することができ、そのプログラムをコンピュータによる読み取り可能な記憶媒体（ＣＤ−ＲＯＭ等）に記憶して提供することが可能である。また、そのプログラムを、インターネット等のネットワークを通して提供することも可能である。
【図面の簡単な説明】
【００８４】
【図１】本実施の形態のストリームデータ管理システムを含むシステムの構成例を示した図である。
【図２】図１のデータ蓄積手段の詳細を示した機能ブロック図である。
【図３】図１のキュー負荷テーブルに基づく取得先リストの作成を概念的に説明した図である。
【図４】図１のフィルタ処理手段の動作を示したフローチャートである。
【図５】図１の取得先リストの作成手順を示すフローチャートである。
【図６】第２の実施の形態の取得先リストを概念的に説明した図である。
【図７】第２の実施の形態のスリープ情報追加部の動作手順を示したフローチャートである。
【図８】データ処理における待ち行列長の原理を説明するために引用したグラフである。
【図９】第２の実施の形態のフィルタ処理手段の動作手順を示したフローチャートである。
【図１０】第２の実施の形態のストリームデータ管理システムを用いた実験結果を示したグラフである。
【図１１】第２の実施の形態のストリームデータ管理システムを用いた実験結果を示したグラフである。
【符号の説明】
【００８５】
１      ストリームデータ管理システム
１０    フィルタ処理手段（フィルタ処理装置）
１１,２１  入力部
１２,２２  処理部
１３,２３  出力部
１４,２４  記憶部
２０    データ蓄積手段（データ蓄積装置）
３０    共有情報管理手段
４０    キュー割当手段
５０    クライアント
６０    出力先
６２    蓄積処理部
６３    検索処理部
７０    検索クライアント
１２１  選択処理部
１２２  データ取得処理部
１２３  フィルタ実行制御部
１２４  フィルタ処理部
１２５  取得先リスト作成部
１２６  キュー負荷テーブル更新部
１２７  スリープ情報追加部
１４１  キュー負荷テーブル
１４２,１４２ａ  取得先リスト
【図１】
0004714202.tif 000002
【図２】
0004714202.tif 000003
【図３】
0004714202.tif 000004
【図４】
0004714202.tif 000005
【図５】
0004714202.tif 000006
【図６】
0004714202.tif 000007
【図７】
0004714202.tif 000008
【図８】
0004714202.tif 000009
【図９】
0004714202.tif 000010
【図１０】
0004714202.tif 000011
【図１１】
0004714202.tif 000012