(57)【特許請求の範囲】
【請求項１】
（１）複数のクライアントから送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）前記データ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記クライアントが送信する前記ストリームデータの送信先となる前記データ蓄積装置を割り当てるキュー割当装置と、を備えるストリームデータ処理システムであって、
前記データ蓄積装置は、
前記ストリームデータをキューとして記憶するデータキュー、および前記フィルタ処理装置からのデータ取得要求をキューとして記憶するフィルタ情報キューを蓄積する記憶部と、
前記クライアントから前記ストリームデータを受け付けるデータ受付処理部と、
前記データ受付処理部が受け付けたストリームデータを、前記データキューに記憶するデータキュー管理部と、
前記複数のフィルタ処理装置から前記ストリームデータのデータ取得要求を受信し、前記データ取得要求を受信した時点において、前記フィルタ情報キューに記憶されている前記データ取得要求の数が所定のキュー最大長以下か否かを判定し、前記データ取得要求の数が所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、前記ストリームデータを送信するためのリクエスト処理部を起動させ、前記データ取得要求の数が所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、当該データ取得要求を送信した前記フィルタ処理装置にリクエスト拒絶通知を送信するフィルタ受付判定部と、
前記フィルタ受付判定部が受け付けたデータ取得要求を、前記フィルタ情報キューにキューとして記憶するフィルタ情報キュー管理部と、
前記フィルタ情報キューの先頭から前記データ取得要求を取得し、当該データ取得要求に示されるフィルタ処理装置に、前記データキューから前記ストリームデータを取得して送信する前記リクエスト処理部と、を備え、
前記フィルタ処理装置は、
前記複数のデータ蓄積装置のうち、前記ストリームデータを蓄積しているデータ蓄積装置を示す取得先情報が記憶される記憶部と、
前記キュー割当装置が割り当てた前記ストリームデータの送信先となる前記データ蓄積装置を示す割当先情報通知を前記キュー割当装置から受信し、前記取得先情報に記憶する取得先情報管理部と、
前記取得先情報に示される、前記ストリームデータを蓄積している前記データ蓄積装置の中から、前記ストリームデータの取得先となる前記データ蓄積装置を選択し、前記データ蓄積装置から前記リクエスト拒絶通知を受信した場合に、前記リクエスト拒絶通知を送信した前記データ蓄積装置以外の他のデータ蓄積装置を前記取得先情報から選択する選択処理部と、
前記選択処理部により選択されたデータ蓄積装置の前記フィルタ受付判定部に、前記データ取得要求を送信し、前記リクエスト処理部から前記ストリームデータを取得するデータ取得処理部と、
前記データ取得処理部が取得した前記ストリームデータにフィルタ処理を実行するフィルタ処理部と、
を備えることを特徴とするストリームデータ処理システム。
【請求項２】
前記データ蓄積装置の記憶部に、前記ストリームデータの前記データキューへの入力時刻および出力時刻を示す入出力データ情報をさらに蓄積し、
前記データキュー管理部は、前記ストリームデータを前記データキューに入出力する毎に、前記入出力データ情報に前記入力時刻および前記出力時刻を記憶し、
前記フィルタ情報キュー管理部は、所定時間毎に前記データキューへの前記所定時間内における前記ストリームデータの入力数（Ｘ）および出力数（Ｙ）を前記入出力データ情報から抽出し、
入力数（Ｘ）＜出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝１ … （式１）
として設定し、
入力数（Ｘ）≧出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝（Ｘ／Ｙ）＊α（αは定数） … （式２）
として設定し、
前記設定したフィルタ情報キューのキュー最大長と比較して、前記フィルタ情報キューに記憶されている前記データ取得要求の数が多い場合に、前記フィルタ情報キューに記憶されている前記データ取得要求の数が、前記設定したフィルタ情報キューのキュー最大長になるまで、当該フィルタ情報キューの先頭または後方から前記データ取得要求を取得し、前記取得したデータ取得要求を送信した前記フィルタ処理装置に、前記リクエスト拒絶通知を送信すること、
を特徴とする請求項１に記載のストリームデータ処理システム。
【請求項３】
（１）複数のクライアントから送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）前記データ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記クライアントが送信する前記ストリームデータの送信先となる前記データ蓄積装置を割り当てるキュー割当装置と、を備えるストリームデータ処理システムに用いられるストリームデータ処理方法であって、
前記データ蓄積装置は、
前記ストリームデータをキューとして記憶するデータキュー、および前記フィルタ処理装置からのデータ取得要求をキューとして記憶するフィルタ情報キューを蓄積する記憶部を備え、
前記クライアントから前記ストリームデータを受け付けるステップと、
前記受け付けたストリームデータを、前記データキューに記憶するステップと、
前記複数のフィルタ処理装置から前記ストリームデータのデータ取得要求を受信し、前記データ取得要求を受信した時点において、前記フィルタ情報キューに記憶されている前記データ取得要求の数が所定のキュー最大長以下か否かを判定し、前記データ取得要求の数が所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、前記フィルタ情報キューにキューとして記憶するステップと、
前記データ取得要求の数が所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、当該データ取得要求を送信した前記フィルタ処理装置にリクエスト拒絶通知を送信するステップと、
前記フィルタ情報キューの先頭から前記データ取得要求を取得し、当該データ取得要求に示されるフィルタ処理装置に、前記データキューから前記ストリームデータを取得して送信するステップと、を実行し、
前記フィルタ処理装置は、
前記複数のデータ蓄積装置のうち、前記ストリームデータを蓄積しているデータ蓄積装置を示す取得先情報が記憶される記憶部を備え、
前記キュー割当装置が割り当てた前記ストリームデータの送信先となる前記データ蓄積装置を示す割当先情報通知を前記キュー割当装置から受信し、前記取得先情報に記憶するステップと、
前記取得先情報に示される、前記ストリームデータを蓄積している前記データ蓄積装置の中から、前記ストリームデータの取得先となる前記データ蓄積装置を選択するステップと、
前記データ蓄積装置から前記リクエスト拒絶通知を受信した場合に、前記リクエスト拒絶通知を送信した前記データ蓄積装置以外の他のデータ蓄積装置を前記取得先情報から選択するステップと、
前記選択したデータ蓄積装置に、前記データ取得要求を送信し、前記ストリームデータを取得するステップと、
前記取得した前記ストリームデータにフィルタ処理を実行するステップと、
を実行することを特徴とするストリームデータ処理方法。
【請求項４】
前記データ蓄積装置の記憶部に、前記ストリームデータの前記データキューへの入力時刻および出力時刻を示す入出力データ情報をさらに蓄積し、
前記データ蓄積装置は、
前記ストリームデータを前記データキューに入出力する毎に、前記入出力データ情報に前記入力時刻および前記出力時刻を記憶するステップと、
所定時間毎に前記データキューへの前記所定時間内における前記ストリームデータの入力数（Ｘ）および出力数（Ｙ）を前記入出力データ情報から抽出し、
入力数（Ｘ）＜出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝１ … （式１）
として設定し、
入力数（Ｘ）≧出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝（Ｘ／Ｙ）＊α（αは定数） … （式２）
として設定し、
前記設定したフィルタ情報キューのキュー最大長と比較して、前記フィルタ情報キューに記憶されている前記データ取得要求の数が多い場合に、前記フィルタ情報キューに記憶されている前記データ取得要求の数が、前記設定したフィルタ情報キューのキュー最大長になるまで、当該フィルタ情報キューの先頭または後方から前記データ取得要求を取得し、前記取得したデータ取得要求を送信した前記フィルタ処理装置に、前記リクエスト拒絶通知を送信するステップと、
を実行することを特徴とする請求項３に記載のストリームデータ処理方法。
【請求項５】
（１）複数のクライアントから送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）前記データ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、を備えるストリームデータ処理システムに用いられる前記データ蓄積装置であって、
前記ストリームデータをキューとして記憶するデータキュー、および前記フィルタ処理装置からのデータ取得要求をキューとして記憶するフィルタ情報キューを蓄積する記憶部と、
前記クライアントから前記ストリームデータを受け付けるデータ受付処理部と、
前記データ受付処理部が受け付けたストリームデータを、前記データキューに記憶するデータキュー管理部と、
前記複数のフィルタ処理装置から前記ストリームデータのデータ取得要求を受信し、前記データ取得要求を受信した時点において、前記フィルタ情報キューに記憶されている前記データ取得要求の数が所定のキュー最大長以下か否かを判定し、前記データ取得要求の数が所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、前記ストリームデータを送信するためのリクエスト処理部を起動させ、前記データ取得要求の数が所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、当該データ取得要求を送信した前記フィルタ処理装置にリクエスト拒絶通知を送信するフィルタ受付判定部と、
前記フィルタ受付判定部が受け付けたデータ取得要求を、前記フィルタ情報キューにキューとして記憶するフィルタ情報キュー管理部と、
前記フィルタ情報キューの先頭から前記データ取得要求を取得し、当該データ取得要求に示されるフィルタ処理装置に、前記データキューから前記ストリームデータを取得して送信する前記リクエスト処理部と、を備え、
前記データ蓄積装置の記憶部に、前記ストリームデータの前記データキューへの入力時刻および出力時刻を示す入出力データ情報をさらに蓄積し、
前記データキュー管理部は、前記ストリームデータを前記データキューに入出力する毎に、前記入出力データ情報に前記入力時刻および前記出力時刻を記憶し、
前記フィルタ情報キュー管理部は、所定時間毎に前記データキューへの前記所定時間内における前記ストリームデータの入力数（Ｘ）および出力数（Ｙ）を前記入出力データ情報から抽出し、
入力数（Ｘ）＜出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝１ … （式１）
として設定し、
入力数（Ｘ）≧出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、
Ｍ＝（Ｘ／Ｙ）＊α（αは定数） … （式２）
として設定し、
前記設定したフィルタ情報キューのキュー最大長と比較して、前記フィルタ情報キューに記憶されている前記データ取得要求の数が多い場合に、前記フィルタ情報キューに記憶されている前記データ取得要求の数が、前記設定したフィルタ情報キューのキュー最大長になるまで、当該フィルタ情報キューの先頭または後方から前記データ取得要求を取得し、前記取得したデータ取得要求を送信した前記フィルタ処理装置に、前記リクエスト拒絶通知を送信すること、
を特徴とするデータ蓄積装置。
【請求項６】
請求項３または請求項４に記載のストリームデータ処理方法を、前記データ蓄積装置と、前記フィルタ処理装置と、前記キュー割当装置と、を備えるストリームデータ処理システムに実行させるためのプログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ストリームデータ処理システム、ストリームデータ処理方法、データ蓄積装置、およびプログラムに関する。
【背景技術】
【０００２】
従来、各種センサやＩＣタグ、携帯電話等の情報源で計測された温度情報、位置情報、画像情報等のデータをストリームデータとして受信し、そのストリームデータに加工、変換、検出等のフィルタ処理を行い、所定の利用先へ流通させるストリームデータ処理技術がある。このようなストリームデータ処理技術として、telegraphCQ（非特許文献１参照）が提案されている。
【０００３】
このtelegraphCQは、入力されるデータの性質変化に合わせてストリームデータ処理の最適な実行スケジューリングを行うが、処理を実行するのは、単一サーバのみである。telegraphCQにおいては、複数のサーバでストリームデータを分散処理する構成は考慮されていないため、ストリームデータ処理システムとしてスケールアウトしないという限界がある。
【０００４】
また、図６に示すような、単体のフィルタ処理装置を並べて、外部装置（クライアント）からのストリームデータを振り分ける接続振り分け装置を用いて、各フィルタ処理装置にデータ処理させることもできるが、この構成では、入力データのデータ量や、その入力データに対する処理時間が一定でない場合等に、各フィルタ処理装置の負荷に偏りが起きるという問題があった。
【０００５】
これに対し、図７に示すように、外部装置（クライアント）とフィルタ処理装置との間に、バッファ領域としてデータ蓄積装置を備え、データ蓄積装置が外部装置（クライアント）から受け取ったストリームデータをキューとして蓄積し、フィルタ処理装置が自律的に、データ取得先となるデータ蓄積装置を選択し、データを取得してフィルタ処理するストリームデータ処理システムが提案されている（特許文献１参照）。
【０００６】
図８は、図７に示した従来技術におけるストリームデータ処理システムを含むシステムのより詳細な構成を示す図である。
図８のストリームデータ処理システム５は、計測データやログデータ等の時系列で作成されるデータを情報源から取得して、そのデータにサービスＩＤ（ＳＩＤ）を付し、ストリームデータとして送信するクライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）と、受信したストリームデータのフィルタ処理を行うストリームデータ処理システム５と、フィルタ処理されたデータの出力先であるデータベース装置６０（６０Ａ，…，６０Ｎ）と、このデータベース装置６０に記憶されたデータの検索を行ったり、データベース装置６０からデータの配信を受けたりする参照クライアント７０（７０Ａ，…，７０Ｎ）とを含んで構成される。
【０００７】
ストリームデータ処理システム５は、クライアント５０からの要求により、ストリームデータの送信先となるデータ蓄積装置２０を割り当てるキュー割当装置４０と、クライアント５０から送信されたストリームデータを蓄積するデータ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）と、ストリームデータのフィルタ処理を行うフィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）と、このフィルタ処理装置１０におけるフィルタ処理部１１４の機能を実現するファイルデータ等を記憶する共有情報管理装置３０とを含んで構成される。
【０００８】
ここで、図８に示した従来のシステム５００の動作概要を説明する。まず、クライアント５０は、様々な情報源からデータを取得し、そのデータに応じたＳＩＤを付す。そして、クライアント５０は、ストリームデータ処理システム５のキュー割当装置４０に、送信先となるデータ蓄積装置２０の割当要求を送信する。キュー割当装置４０は、データの送信先となるデータ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）のうちの１つを割り当て、その割当先情報をクライアント５０に返信するとともに、割当先情報通知として、割り当てたデータ蓄積装置２０のアドレス情報を、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）へ送信する。この割当先情報通知をフィルタ処理装置１０が受信し、取得先情報管理部１１１が、取得先情報１２０を更新する。この取得先情報１２０は、データ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）のうち、ストリームデータが現時点で蓄積されているデータ蓄積装置２０であることを示す情報である。フィルタ処理装置１０の選択処理部１１２は、この取得先情報１２０を参照することにより、ストリームデータが蓄積されているデータ蓄積装置２０の中から、データの取得先を選択することができる。
【０００９】
キュー割当装置４０からデータ蓄積装置２０の割当先情報を受信したクライアント５０は、キュー割当装置４０が割り当てたデータ蓄積装置２０へ、ＳＩＤを付したストリームデータを送信する。なお、データ蓄積装置２０のデータキュー管理部２１２は、自身のキュー負荷が所定の閾値以上になった場合に、キュー負荷が増加したことを示す負荷状態変更通知を各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に通知する。この負荷状態変更通知を各フィルタ処理装置１０が受信すると、負荷状態変更通知を送信したデータ蓄積装置２０の処理負荷を減らすため、フィルタ処理装置１０は、優先してそのデータ蓄積装置２０を取得先として選択しデータを取得する。
【００１０】
続いて、フィルタ処理装置１０の選択処理部１１２が、取得先情報１２０を参照し、現時点でストリームデータを蓄積しているデータ蓄積装置２０の中から、データを取得しにいくデータ蓄積装置２０を選択する。このように、現時点でストリームデータを蓄積しているデータ蓄積装置２０を示す取得先情報１２０を参照することで、データを受け取っていないデータ蓄積装置２０へアクセスすることをなくすことができる。つまり、ストリームデータを蓄積していないデータ蓄積装置２０に対し、無駄にアクセスすることをなくすことができる。そして、フィルタ処理装置１０のデータ取得処理部１１３は、選択処理部１１２が選択したデータ蓄積装置２０に、データ取得要求を送信する。
【００１１】
フィルタ処理装置１０からのデータ取得要求を受け付けたデータ蓄積装置２０のフィルタ受付部２１３は、リクエスト処理部２１４を起動させ、データキュー２２１からデータを取得し、データ取得要求を送信したフィルタ処理装置１０に対しストリームデータを送信する。ここで、フィルタ受付部２１３は、各フィルタ処理装置１０からデータ取得要求を受け付ける毎に、そのデータ取得要求に対応するリクエスト処理部２１４を起動させる。つまり、例えば、１０個のフィルタ処理装置１０からデータ取得要求をフィルタ受付部２１３が受け付けた場合には、１０個のリクエスト処理部２１４が起動される。よって、このリクエスト処理部２１４は、フィルタ受付部２１３の指示により、１つのデータ取得要求に対応して１つが起動され、起動数には制限がない。
【００１２】
次に、データ蓄積装置２０からフィルタ処理装置１０のデータ取得処理部１１３がデータを受信し、フィルタ処理部１１４によるフィルタ処理が行われる。フィルタ処理部１１４は、各ストリームデータに対して行うべきフィルタ処理を、そのストリームデータに付されたＳＩＤに基づき判断する。なお、ここで、このストリームデータに必要なフィルタ処理機能が自身のフィルタ処理部１１４になかった場合、フィルタ処理装置１０は、共有情報管理装置３０からフィルタデータを取得して、フィルタ処理機能の追加を行うようにすることができる。
【００１３】
フィルタ処理装置１０は、フィルタ処理後、ストリームデータを所定のデータベース装置６０に出力する。そしてデータベース装置６０においてフィルタ処理後のデータがデータベースに記憶され、そのデータベースのデータが、参照クライアント７０により検索可能となる。また、データベース装置６０は、予め設定された参照クライアント７０に、フィルタ処理装置１０から受信したデータを配信する。
【００１４】
このような図８に示す従来技術におけるストリームデータ処理システム５によれば、以下のような効果を得ることができる。
【００１５】
（効果Ａ）クライアント５０から入力されるストリームデータ数の増加に対応して、フィルタ処理装置１０やデータ蓄積装置２０のスケールアウトが可能となる。また、ストリームデータをデータ蓄積装置２０にキューとして蓄積し、フィルタ処理装置１０がその蓄積されたデータをデータ蓄積装置２０に取得しにいく構成とすることにより、フィルタ処理装置１０の負荷を分散した上で、並列処理することが可能となる。そのため、システム全体としてのスループットが向上する。
【００１６】
（効果Ｂ）フィルタ処理装置１０の高可用化が可能となる。具体的には、フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）のうちのあるフィルタ処理装置１０が故障した場合であっても、クライアント５０から受信したストリームデータは、データ蓄積装置２０に蓄積されているため、故障したフィルタ処理装置１０以外のフィルタ処理装置１０が、故障したフィルタ処理装置１０の代わりに、データを取得しフィルタ処理することが可能となる。
【００１７】
（効果Ｃ）ストリームデータ処理システム５は、フィルタ処理部１１４によるフィルタ処理の機能を実現するためのファイルデータが記憶された共有情報管理装置３０を備える。この構成により、あるフィルタ処理装置１０のフィルタ処理部１１４に、データ蓄積装置２０から取得したストリームデータに必要なフィルタ処理機能が備わっていない場合に、共有情報管理装置３０からファイルデータを取得することにより、フィルタ処理機能を追加したり、該当するフィルタ処理のバージョンアップをしたりすること等が可能となる。そして、フィルタ処理装置１０は、自律的にデータ蓄積装置２０を選択してデータを取得するため、１つデータのフィルタ処理を終了し次のデータのフィルタ処理を行う間に、共有情報管理装置３０からフィルタ処理機能の追加を行えばよく、システム全体を停止させる必要がない。
【先行技術文献】
【非特許文献】
【００１８】
【非特許文献１】Sirish Chandrasekaran, Owen Cooper, Amol Deshpande, Michael J. Franklin, Joseph M. Hellerstein, Wei Hong, Sailesh Krishnamurthy, Samuel Madden, Vijayshankar Raman, Frederick Reiss, Mehul A. Shah: “ TelegraphCQ: Continuous Dataflow Processing for an Uncertain World. ”, CIDR(2003)
【特許文献】
【００１９】
【特許文献１】特開２００９−１０４５１３号公報
【発明の概要】
【発明が解決しようとする課題】
【００２０】
しかしながら、多数のフィルタ処理装置１０と多数のデータ蓄積装置２０とから構成されるストリームデータ処理システム５においては、以下に示すケースの場合に、ある特定のデータ蓄積装置２０に対して、フィルタ処理装置１０からのデータ取得要求が増加するという問題がある。
【００２１】
（ケース１）クライアント５０からのストリームデータの送信数が減少したため、ストリームデータを受け付けているデータ蓄積装置２０の数が減少した時。
大量のストリームデータを受け付けることを前提に、多数のフィルタ処理装置１０と多数のデータ蓄積装置２０とで構成されるストリームデータ処理システム５の場合、必ずしも、ストリームデータを常時一定量受け付けるとは限らない。停電や故障等の何らかの原因により、一時的にストリームデータ処理システム５が受け付けるストリームデータの数が、減少することがある。その場合、ストリームデータを蓄積していないデータ蓄積装置２０に対しては、フィルタ処理装置１０はアクセスしないため、ストリームデータを受け付けた数少ないデータ蓄積装置２０に対して、多数のフィルタ処理装置１０からのデータ取得要求が集中して送信される。
【００２２】
（ケース２）データ蓄積装置２０のデータキュー管理部２１２が、自身のキュー負荷が増加したことを示す負荷状態変更通知を各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に通知した時。
データ蓄積装置２０のデータキュー管理部２１２は、キューの負荷状態を監視しており、例えば、入力スループットや、フィルタ処理装置１０におけるデータ１個あたりの平均処理時間等が、所定の閾値に達したときに、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に負荷状態変更通知を送信する。この負荷状態変更通知を受信したフィルタ処理装置１０は、負荷状態変更通知を送信したデータ蓄積装置２０の処理負荷を減らす目的で、優先的にそのデータ蓄積装置２０にデータ取得要求を送信する。その結果、負荷状態変更通知を送信したデータ蓄積装置２０に、多数のフィルタ処理装置１０からのデータ取得要求が集中して送信される。
【００２３】
（ケース３）システム起動後、最初のストリームデータを受信した時。
ストリームデータ処理システム５の起動直後に、最初のストリームデータを、あるデータ蓄積装置２０が受信すると、他のデータ蓄積装置２０には、まだストリームデータが蓄積されていないため、起動した多数のフィルタ処理装置１０から、そのデータ蓄積装置２０に集中してデータ取得要求が送信される。
【００２４】
このようなケース１〜３等の場合に、従来のストリームデータ処理システム５においては、データ蓄積装置２０が、フィルタ処理装置１０からの多数のデータ取得要求を受信すると、フィルタ受付部２１３が、受信したデータ取得要求の数に対応するリクエスト処理部２１４を起動させてしまう。そして、リクエスト処理部２１４が大量に起動されることで、データ蓄積装置２０のリソースが消費され、さらに、起動されたリクエスト処理部２１４が、データキュー２２１からデータを取得するためデータキュー管理部２１２にアクセスすると、データキュー２２１の入出力レスポンスが著しく低下する。その結果、データ受信待ちのフィルタ処理装置１０が増加し、ストリームデータ処理システム５全体のスループットが低下する。
【００２５】
このような問題に鑑みて本発明がなされたのであり、本発明は、特定のデータ蓄積装置に、多数のフィルタ処理装置からのデータ取得要求が集中することによる、系全体のスループットの低下を防止することを課題とする。
【課題を解決するための手段】
【００２６】
前記課題を解決するため、請求項１に記載の発明は、（１）複数のクライアントから送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）前記データ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記クライアントが送信する前記ストリームデータの送信先となる前記データ蓄積装置を割り当てるキュー割当装置と、を備えるストリームデータ処理システムであって、前記データ蓄積装置が、前記ストリームデータをキューとして記憶するデータキュー、および前記フィルタ処理装置からのデータ取得要求をキューとして記憶するフィルタ情報キューを蓄積する記憶部と、前記クライアントから前記ストリームデータを受け付けるデータ受付処理部と、前記データ受付処理部が受け付けたストリームデータを、前記データキューに記憶するデータキュー管理部と、前記複数のフィルタ処理装置から前記ストリームデータのデータ取得要求を受信し、前記データ取得要求を受信した時点において、前記フィルタ情報キューに記憶されている前記データ取得要求の数が所定のキュー最大長以下か否かを判定し、前記データ取得要求の数が所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、前記ストリームデータを送信するためのリクエスト処理部を起動させ、前記データ取得要求の数が所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、当該データ取得要求を送信した前記フィルタ処理装置にリクエスト拒絶通知を送信するフィルタ受付判定部と、前記フィルタ受付判定部が受け付けたデータ取得要求を、前記フィルタ情報キューにキューとして記憶するフィルタ情報キュー管理部と、前記フィルタ情報キューの先頭から前記データ取得要求を取得し、当該データ取得要求に示されるフィルタ処理装置に、前記データキューから前記ストリームデータを取得して送信する前記リクエスト処理部と、を備え、前記フィルタ処理装置が、前記複数のデータ蓄積装置のうち、前記ストリームデータを蓄積しているデータ蓄積装置を示す取得先情報が記憶される記憶部と、前記キュー割当装置が割り当てた前記ストリームデータの送信先となる前記データ蓄積装置を示す割当先情報通知を前記キュー割当装置から受信し、前記取得先情報に記憶する取得先情報管理部と、前記取得先情報に示される、前記ストリームデータを蓄積している前記データ蓄積装置の中から、前記ストリームデータの取得先となる前記データ蓄積装置を選択し、前記データ蓄積装置から前記リクエスト拒絶通知を受信した場合に、前記リクエスト拒絶通知を送信した前記データ蓄積装置以外の他のデータ蓄積装置を前記取得先情報から選択する選択処理部と、前記選択処理部により選択されたデータ蓄積装置の前記フィルタ受付判定部に、前記データ取得要求を送信し、前記リクエスト処理部から前記ストリームデータを取得するデータ取得処理部と、前記データ取得処理部が取得した前記ストリームデータにフィルタ処理を実行するフィルタ処理部と、を備えることを特徴とするストリームデータ処理システムとした。
【００２７】
また、請求項３に記載の発明は、（１）複数のクライアントから送信されたストリームデータを蓄積する複数のデータ蓄積装置と、（２）前記データ蓄積装置から取得した前記ストリームデータに対しフィルタ処理を実行する複数のフィルタ処理装置と、（３）前記クライアントが送信する前記ストリームデータの送信先となる前記データ蓄積装置を割り当てるキュー割当装置と、を備えるストリームデータ処理システムに用いられるストリームデータ処理方法であって、前記データ蓄積装置が、前記ストリームデータをキューとして記憶するデータキュー、および前記フィルタ処理装置からのデータ取得要求をキューとして記憶するフィルタ情報キューを蓄積する記憶部を備え、前記クライアントから前記ストリームデータを受け付けるステップと、前記受け付けたストリームデータを、前記データキューに記憶するステップと、前記複数のフィルタ処理装置から前記ストリームデータのデータ取得要求を受信し、前記データ取得要求を受信した時点において、前記フィルタ情報キューに記憶されている前記データ取得要求の数が所定のキュー最大長以下か否かを判定し、前記データ取得要求の数が所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、前記フィルタ情報キューにキューとして記憶するステップと、前記データ取得要求の数が所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、当該データ取得要求を送信した前記フィルタ処理装置にリクエスト拒絶通知を送信するステップと、前記フィルタ情報キューの先頭から前記データ取得要求を取得し、当該データ取得要求に示されるフィルタ処理装置に、前記データキューから前記ストリームデータを取得して送信するステップと、を実行し、前記フィルタ処理装置が、前記複数のデータ蓄積装置のうち、前記ストリームデータを蓄積しているデータ蓄積装置を示す取得先情報が記憶される記憶部を備え、前記キュー割当装置が割り当てた前記ストリームデータの送信先となる前記データ蓄積装置を示す割当先情報通知を前記キュー割当装置から受信し、前記取得先情報に記憶するステップと、前記取得先情報に示される、前記ストリームデータを蓄積している前記データ蓄積装置の中から、前記ストリームデータの取得先となる前記データ蓄積装置を選択するステップと、前記データ蓄積装置から前記リクエスト拒絶通知を受信した場合に、前記リクエスト拒絶通知を送信した前記データ蓄積装置以外の他のデータ蓄積装置を前記取得先情報から選択するステップと、前記選択したデータ蓄積装置に、前記データ取得要求を送信し、前記ストリームデータを取得するステップと、前記取得した前記ストリームデータにフィルタ処理を実行するステップと、を実行することを特徴とするストリームデータ処理方法とした。
【００２８】
このようにすることで、ストリームデータ処理システムのデータ蓄積装置は、フィルタ処理装置からデータ取得要求を受信すると、フィルタ情報キューに記憶されているデータ取得要求の数が所定のキュー最大長以下か否かを判定し、所定のキュー最大長を超えた場合に、受信したデータ取得要求を受け付けず、そのデータ取得要求を送信したフィルタ処理装置に、リクエスト拒絶通知を送信する。そして、リクエスト拒絶通知を受信したフィルタ処理装置は、リクエスト拒絶通知を送信したデータ蓄積装置以外の他のデータ蓄積装置に、データ取得要求を送信する。
【００２９】
よって、本発明によれば、特定のデータ蓄積装置に、フィルタ処理装置からのデータ取得要求が集中した場合であっても、そのデータ蓄積装置は、フィルタ情報キューにおける所定のキュー最大長を超えるデータ取得要求を受け付けないため、データキューの入出力レスポンスを低下させることがなく、データ受信待ちのフィルタ処理装置の増加を防ぐことができる。その結果、システム全体のスループットの低下を防ぐことができる。
【００３０】
請求項２に記載の発明は、前記データ蓄積装置の記憶部に、前記ストリームデータの前記データキューへの入力時刻および出力時刻を示す入出力データ情報をさらに蓄積し、前記データキュー管理部が、前記ストリームデータを前記データキューに入出力する毎に、前記入出力データ情報に前記入力時刻および前記出力時刻を記憶し、前記フィルタ情報キュー管理部は、所定時間毎に前記データキューへの前記所定時間内における前記ストリームデータの入力数（Ｘ）および出力数（Ｙ）を前記入出力データ情報から抽出し、入力数（Ｘ）＜出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、Ｍ＝１…（式１）として設定し、入力数（Ｘ）≧出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、Ｍ＝（Ｘ／Ｙ）＊α（αは定数）…（式２）として設定し、前記設定したフィルタ情報キューのキュー最大長と比較して、前記フィルタ情報キューに記憶されている前記データ取得要求の数が多い場合に、前記フィルタ情報キューに記憶されている前記データ取得要求の数が、前記設定したフィルタ情報キューのキュー最大長になるまで、当該フィルタ情報キューの先頭または後方から前記データ取得要求を取得し、前記取得したデータ取得要求を送信した前記フィルタ処理装置に、前記リクエスト拒絶通知を送信すること、を特徴とする請求項１に記載のストリームデータ処理システムとした。
【００３１】
また、請求項４に記載の発明は、前記データ蓄積装置の記憶部に、前記ストリームデータの前記データキューへの入力時刻および出力時刻を示す入出力データ情報をさらに蓄積し、前記データ蓄積装置が、前記ストリームデータを前記データキューに入出力する毎に、前記入出力データ情報に前記入力時刻および前記出力時刻を記憶するステップと、所定時間毎に前記データキューへの前記所定時間内における前記ストリームデータの入力数（Ｘ）および出力数（Ｙ）を前記入出力データ情報から抽出し、入力数（Ｘ）＜出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、Ｍ＝１…（式１）として設定し、入力数（Ｘ）≧出力数（Ｙ）のときに、前記フィルタ情報キューのキュー最大長（Ｍ）を、Ｍ＝（Ｘ／Ｙ）＊α（αは定数）…（式２）として設定し、前記設定したフィルタ情報キューのキュー最大長と比較して、前記フィルタ情報キューに記憶されている前記データ取得要求の数が多い場合に、前記フィルタ情報キューに記憶されている前記データ取得要求の数が、前記設定したフィルタ情報キューのキュー最大長になるまで、当該フィルタ情報キューの先頭または後方から前記データ取得要求を取得し、前記取得したデータ取得要求を送信した前記フィルタ処理装置に、前記リクエスト拒絶通知を送信するステップと、を実行することを特徴とする請求項３に記載のストリームデータ処理方法とした。
【００３２】
このようにすることで、データ蓄積装置は、データキューへのストリームデータの入力数および出力数に応じて、フィルタ情報キューのキュー最大長を設定することができ、フィルタ情報キューに記憶された状態で、待ち状態にあるデータ取得要求の数を低減させることができる。その結果、データ受信待ちのフィルタ処理装置を減らすことができる。
【００３３】
また、請求項５に記載の発明によれば、特定のデータ蓄積装置に、フィルタ処理装置からのデータ取得要求が集中した場合であっても、フィルタ情報キューにおける所定のキュー最大長を超えるデータ取得要求を受け付けず、データキューの入出力レスポンスの低下を防ぐことができるデータ蓄積装置を提供することができる。
さらに、データ蓄積装置が、データキューへのストリームデータの入力数および出力数に応じて、フィルタ情報キューのキュー最大長を設定することができ、フィルタ情報キューに記憶された状態で、待ち状態にあるデータ取得要求の数を低減させることができるデータ蓄積装置を提供することができる。
【００３５】
請求項６に記載の発明は、請求項３または請求項４に記載のストリームデータ処理方法を、前記データ蓄積装置と、前記フィルタ処理装置と、前記キュー割当装置と、を備えるストリームデータ処理システムに実行させるためのプログラムとした。
【００３６】
このようなプログラムによれば、請求項３または請求項４に記載のストリームデータ処理方法を、ストリームデータ処理システムにおいて実現させることができる。
【発明の効果】
【００３７】
本発明によれば、特定のデータ蓄積装置に、多数のフィルタ処理装置からのデータ取得要求が集中することによる、系全体のスループットの低下を防止する、ストリームデータ処理システム、ストリームデータ処理方法、データ蓄積装置、およびプログラムを提供することができる。
【図面の簡単な説明】
【００３８】
【図１】本実施形態に係るストリームデータ処理システムの構成例を示す機能ブロック図である。
【図２】本実施形態に係る入出力データ情報のデータ構成の一例を示す図である。
【図３】本実施形態に係るストリームデータ処理の流れを示すシーケンス図である。
【図４】本実施形態に係るキュー最大長更新処理の流れを示すフローチャートである。
【図５】本実施形態に係るフィルタ情報キュー管理部によるキュー最大長の設定処理を説明するための図である。
【図６】単体のフィルタ処理装置を並列させて、ストリームデータを処理するシステムを示す図である。
【図７】従来技術におけるストリームデータ処理システムの概略を示す図である。
【図８】従来技術におけるストリームデータ処理システムの構成例を示す機能ブロック図である。
【発明を実施するための形態】
【００３９】
次に、本発明を実施するための形態（「実施形態」という）について、適宜図面を参照しながら詳細に説明する。
【００４０】
図１は、本実施形態に係るストリームデータ処理システム１を含むシステム１００の構成例を示す図である。図１に示すように、システム１００は、図８で示した従来技術におけるシステム５００の構成と同様に、情報源から取得したデータにＳＩＤを付し、ストリームデータとして送信するクライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）と、受信したストリームデータのフィルタ処理を行うストリームデータ処理システム１と、データベース装置６０（６０Ａ，…，６０Ｎ）と、参照クライアント７０（７０Ａ，…，７０Ｎ）とを含んで構成される。なお、図８に示した構成と同様の処理を実行する構成には、同一の符号と名称を付し、本実施形態に特有な処理を行う構成を除いて説明を省略する。
【００４１】
また、ストリームデータ処理システム１は、図８に示したストリームデータ処理システム５と同様に、ストリームデータの送信先となるデータ蓄積装置２０を割り当てるキュー割当装置４０と、クライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）から送信されたストリームデータを蓄積するデータ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）と、ストリームデータのフィルタ処理を行うフィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）と、このフィルタ処理装置１０におけるフィルタ処理部１１４（詳細は後記）の機能を実現するファイルデータ等を記憶する共有情報管理装置３０とを含んで構成される。なお、図８に示すストリームデータ処理システム５との違いは、各装置の詳細において説明する。
【００４２】
なお、（１）クライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）と、キュー割当装置４０およびデータ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）との間、（２）キュー割当装置４０と、データ蓄積装置２０（２０Ａ，２０Ｂ，…、２０Ｎ）およびフィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）との間、（３）データ蓄積装置２０（２０Ａ，２０Ｂ，…２０Ｎ）と、フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）との間、（４）フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）とデータベース装置６０（６０Ａ，…，６０Ｎ）との間、（５）データベース装置６０（６０Ａ，…，６０Ｎ）と、参照クライアント７０（７０Ａ，…，７０Ｎ）との間はそれぞれ、図示しない通信ネットワークにより接続される。さらに、クライアント５０、データ蓄積装置２０、フィルタ処理装置１０、データベース装置６０、参照クライアント７０は、それぞれＮ個としたが、これに限定されない。
【００４３】
≪動作概要≫
ここで、図１のシステム１００の動作概要を説明する。
クライアント５０によりＳＩＤが付されたストリームデータを、データ蓄積装置２０が受け付けると、キューとしてデータキュー２２１に保存する。一方、フィルタ処理装置１０の選択処理部１１２は、取得先情報１２０を参照し、ストリームデータを蓄積しているデータ蓄積装置２０の中から１つを選択し、データ取得処理部１１３が、その選択したデータ蓄積装置２０のフィルタ受付判定部２１５にデータ取得要求を送信する。
【００４４】
データ蓄積装置２０のフィルタ受付判定部２１５は、フィルタ処理装置１０からデータ取得要求を受信した時点において、フィルタ情報キュー２２２に記憶されているデータ取得要求の数が所定のキュー最大長以下か否かを判定する。そして、フィルタ受付判定部２１５は、そのデータ取得要求の数がフィルタ情報キュー２２２における所定のキュー最大長以下と判定した場合に、当該受信したデータ取得要求を受け付けて、リクエスト処理部２１４を起動させる。また、フィルタ情報キュー管理部２１６を介して、そのデータ取得要求をフィルタ情報キュー２２２に記憶する。一方、フィルタ受付判定部２１５は、そのデータ取得要求の数がフィルタ情報キュー２２２における所定のキュー最大長を超えると判定した場合に、当該受信したデータ取得要求を受け付けず、そのデータ所得要求を送信したフィルタ処理装置１０にリクエスト拒絶通知を送信する。
【００４５】
データ取得要求がフィルタ受付判定部２１５により受け付けられると、起動されたリクエスト処理部２１４は、フィルタ情報キュー２２２の先頭からデータ取得要求を取得し、そのデータ取得要求に示されるフィルタ処理装置１０に対して、データキュー２２１からストリームデータを取得して送信する。そして、フィルタ処理装置１０によりストリームデータがフィルタ処理される。
【００４６】
一方、データ蓄積装置２０のフィルタ受付判定部２１５からリクエスト拒絶通知を受信したフィルタ処理装置１０は、取得先情報１２０を参照して、そのリクエスト拒絶通知を送信したデータ蓄積装置２０以外の他のデータ蓄積装置２０を選択し、データ取得要求を送信する。
【００４７】
このようにすることで、データ蓄積装置２０に、フィルタ処理装置１０からのデータ取得要求が集中した場合であっても、フィルタ情報キュー２２２における所定のキュー最大長を超えるデータ取得要求を受け付けず、リクエスト拒絶通知を返信することで、他のデータ蓄積装置２０にデータ取得要求の送信先を変更させることができる。その結果、データ受信待ちのフィルタ処理装置１０の増加を防ぎ、システム全体としてのスループットの低下を防ぐことができる。
【００４８】
また、フィルタ情報キュー２２２のキュー最大長は、フィルタ情報キュー管理部２１６が、所定時間内のストリームデータのデータキュー２２１への入出力数に応じて設定する。そして、設定したキュー最大長と比較して、フィルタ情報キュー２２２に記憶されているデータ取得要求の数が多い場合に、設定したキュー最大長を超える分のデータ取得要求をフィルタ情報キュー２２２から取得し、そのデータ取得要求に示されるフィルタ処理装置１０にリクエスト拒絶通知を送信する。
【００４９】
このようにすることで、フィルタ情報キュー２２２において、待ち状態にあるデータ取得要求の数を低減させることができ、データ受信待ちのフィルタ処理装置１０を減らすことができる。
【００５０】
≪システム構成≫
次に、本実施形態に係るストリームデータ処理システム１を構成する各装置について、図１を用いて、具体的に説明する。
【００５１】
＜キュー割当装置＞
キュー割当装置４０は、クライアント５０（５０Ａ，５０Ｂ，…，５０Ｎ）の接続先となるデータ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）の割当てを行う装置であり、接続先割当部４１と、割当先通知部４２とを含んで構成される。なお、ここでは、図示を省略しているが、キュー割当装置４０は、各種データの入出力を司り入出力インタフェースにより実現される入出力部、ＣＰＵ（Central Processing Unit）、ＲＡＭ（Random Access Memory）やハードディスク等により実現される記憶部、通信インタフェース等を備えるコンピュータによる実現される。
【００５２】
接続先割当部４１は、クライアント５０からストリームデータの送信先となるデータ蓄積装置２０の割当要求を受信すると、所定の割当方法で割当先のデータ蓄積装置２０を決定し、割当要求を送信したクライアント５０に割当先情報として返信する。ここで、接続先割当部４１の割当方法としては、例えば、ラウンドロビンを用いてもよいし、各データ蓄積装置２０の負荷状態に関する情報を取得して、より負荷の少ない割当先を選択するようにしてもよい。
【００５３】
割当先通知部４２は、接続先割当部４１が割り当てたストリームデータの送信先となるデータ蓄積装置２０のアドレス情報を、割当先情報通知として、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に送信する。
【００５４】
＜データ蓄積装置＞
データ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）は、クライアント５０から送信されたストリームデータを一時的に蓄積する。このデータ蓄積装置２０は、ストリームデータの受付けや、フィルタ処理装置１０からのデータ取得要求に応じて行うデータ処理の全般を司る処理部２１と、ストリームデータ等をキューとして記憶する記憶部２２と、不図示の入出力部とを含んで構成される。この入出力部は、クライアント５０やフィルタ処理装置１０との間でデータの入出力が可能な入出力インタフェースにより実現される。なお、図８に示すデータ蓄積装置２０との違いは、処理部２１内に、図８のフィルタ受付部２１３の代わりにフィルタ受付判定部２１５を備え、さらに、フィルタ情報キュー管理部２１６を備えること、そして、記憶部２２内に、フィルタ情報キュー２２２と入出力データ情報２２３とを備えることである（詳細は後記）。
【００５５】
処理部２１は、ＣＰＵによる記憶部２２に記憶されるプログラムの実行処理や、専用回路等により実現され、データ受付処理部２１１と、データキュー管理部２１２と、リクエスト処理部２１４と、フィルタ受付判定部２１５と、フィルタ情報キュー管理部２１６とを含んで構成される。
【００５６】
データ受付処理部２１１は、クライアント５０から、不図示の入出力部経由でストリームデータを受信する。そして、受信したストリームデータをデータキュー管理部２１２に出力する。
【００５７】
データキュー管理部２１２は、データ受付処理部２１１から出力されたデータ（ストリームデータ）を、データキュー２２１に記憶する。そして、リクエスト処理部２１４からのデータ取得要求があったとき、データキュー２２１に記憶されたストリームデータをリクエスト処理部２１４に出力する。
また、データキュー管理部２１２は、ストリームデータをデータキュー２２１から入出力する毎に、その入力時刻と出力時刻を、記憶部２２に入出力データ情報２２３として記憶する。
【００５８】
なお、このデータキュー管理部２１２は、データキュー２２１の負荷状態を監視し、例えば、入力スループットや、フィルタ処理装置１０におけるデータ１個あたりの平均処理時間等が、所定の閾値に達したときに、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に負荷状態変更通知を送信するようにしてもよい。なお、負荷状態変更通知を受信したフィルタ処理装置１０は、負荷状態変更通知を送信したデータ蓄積装置２０を優先的に選択して、キュー負荷を減らすことができる。
また、データキュー管理部２１２は、所定時間、データキュー２２１に、ストリームデータが蓄積されていない状態が続いた場合に、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に対して、取得先情報更新通知を送信する。この取得先情報更新通知は、そのデータ蓄積装置２０のデータキュー２２１に、ストリームデータが蓄積されていないことを示す情報である。
【００５９】
フィルタ受付判定部２１５は、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）から、ストリームデータのデータ取得要求を受信し、そのデータ取得要求を受信した時点において、後記するフィルタ情報キュー２２２に記憶されているデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長以下か否かを判定する。そして、フィルタ受付判定部２１５は、フィルタ情報キュー２２２に記憶されたデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長以下と判定した場合に、受信したデータ取得要求を受け付けて、フィルタ情報キュー管理部２１６に出力する。そして、ストリームデータをフィルタ処理装置１０に送信するためのリクエスト処理部２１４を起動させる。また、フィルタ受付判定部２１５は、フィルタ情報キュー２２２に記憶されたデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長を超えていると判定した場合に、その受信したデータ取得要求を受け付けず、そのデータ取得要求を送信したフィルタ処理装置１０にリクエスト拒絶通知を送信する。なお、このフィルタ情報キュー２２２のキュー最大長は、記憶部２２内に予め記憶しておくものとしてもよいし、後記するように、フィルタ情報キュー管理部２１６が、ストリームデータのデータキュー２２１への入力数および出力数を記憶した入出力データ情報２２３を用いて、所定時間毎に設定するものとしてもよい。
【００６０】
フィルタ情報キュー管理部２１６は、フィルタ受付判定部２１５から出力されたデータ取得要求を、フィルタ情報キュー２２２に記憶する。そして、リクエスト処理部２１４の指示により、フィルタ情報キュー２２２に記憶されたデータ取得要求をそのリクエスト処理部２１４に出力する。
【００６１】
また、フィルタ情報キュー管理部２１６は、入出力データ情報２２３から、所定時間毎のストリームデータの入力数（Ｘ）および出力数（Ｙ）を抽出し、フィルタ情報キュー２２２のキュー最大長（Ｍ）を設定する。
【００６２】
具体的には、入力数（Ｘ）＜出力数（Ｙ）のときに、フィルタ情報キュー２２２のキュー最大長（Ｍ）を、
Ｍ＝１  …  （式１）
として設定し、
入力数（Ｘ）≧出力数（Ｙ）のときに、フィルタ情報キュー２２２のキュー最大長（Ｍ）を、
Ｍ＝（Ｘ／Ｙ）＊α（αは定数）  …  （式２）
として設定する。
【００６３】
そして、フィルタ情報キュー管理部２１６は、（式１）または（式２）を用いて設定したフィルタ情報キュー２２２のキュー最大長と比較して、その時点のフィルタ情報キュー２２２に記憶されているデータ取得要求の数が大きい場合は、フィルタ情報キュー２２２に記憶されているデータ取得要求の数が、設定したフィルタ情報キュー２２２のキュー最大長になるまで、フィルタ情報キュー２２２の先頭または後方からデータ取得要求を取得する。そして、フィルタ情報キュー管理部２１６は、その取得したデータ取得要求に示されるフィルタ処理装置１０に、リクエスト拒絶通知を送信する。
【００６４】
リクエスト処理部２１４は、フィルタ情報キュー２２２の先頭から、フィルタ情報キュー管理部２１６を介して、データ取得要求を取得し、そのデータ取得要求に示されるフィルタ処理装置１０に対して、データキュー２２１からストリームデータを取得して送信する。
また、リクエスト処理部２１４は、フィルタ処理装置１０から、ストリームデータのフィルタ処理が完了したことを示すフィルタ処理完了通知を受信すると、データキュー管理部２１２を介して、そのフィルタ処理完了通知に示されるフィルタ処理済みのストリームデータをデータキュー２２１から削除する。
【００６５】
次に、記憶部２２は、ＲＡＭやハードディスク等により実現される記憶手段であり、データキュー２２１と、フィルタ情報キュー２２２と、入出力データ情報２２３とを含んで構成される。
【００６６】
データキュー２２１は、データキュー管理部２１２から出力されたストリームデータをキューとして蓄積する。
【００６７】
フィルタ情報キュー２２２は、フィルタ情報キュー管理部２１６から出力されたデータ取得要求をキューとして蓄積する。
【００６８】
入出力データ情報２２３は、データキュー管理部２１２がストリームデータをデータキュー２２１に入出力する毎に、所定の時間間隔毎に、その入力時刻および出力時刻が記憶される。
【００６９】
図２は、本実施形態に係る入出力データ情報２２３のデータ構成の一例を示す図である。図２では、所定の時間間隔として１０秒毎に、ストリームデータが入出力された時刻がレコード（rec）として記憶される。この入出力データ情報２２３により、ある１０秒間において、入力されたストリームデータの数が何レコードあり、出力されたストリームデータの数が何レコードあるかが示される。
【００７０】
＜フィルタ処理装置＞
図１に戻り、フィルタ処理装置１０は、データ蓄積装置２０（２０Ａ，２０Ｂ，…，２０Ｎ）から、ストリームデータを取得し、その取得したストリームデータに対しフィルタ処理を実行する。このフィルタ処理装置１０は、フィルタ処理の全体を制御する処理部１１と、記憶部１２と、不図示の入出力部とを含んで構成される。なお、入出力部は、フィルタ処理装置１０や、キュー割当装置４０との間でデータの入出力が可能な入出力インタフェースにより実現される。
【００７１】
処理部１１は、ＣＰＵによる記憶部１２に記憶されるプログラムの実行処理や、専用回路等により実現され、取得先情報管理部１１１と、選択処理部１１２と、データ取得処理部１１３と、フィルタ処理部１１４とを含んで構成される。
【００７２】
取得先情報管理部１１１は、記憶部１２内の取得先情報１２０の更新を行う。具体的には、取得先情報管理部１１１は、キュー割当装置４０の割当先通知部４２から、ストリームデータの送信先となるデータ蓄積装置２０を示す割当先情報通知を受信し、後記する記憶部１２内の取得先情報１２０に記憶する。また、データ蓄積装置２０のデータキュー管理部２１２から、データキュー２２１にストリームデータが記憶されていないことを示す取得先情報更新通知を受信し、その取得先情報更新通知を送信したデータ蓄積装置２０を取得先情報１２０から削除する。
【００７３】
選択処理部１１２は、取得先情報１２０に示される、ストリームデータを蓄積しているデータ蓄積装置２０の中から、ストリームデータの取得先となるデータ蓄積装置２０を選択し、データ取得処理部１１３に出力する。選択処理部１１２による取得先情報１２０に基づくデータ蓄積装置２０の選択方法は、例えば、ラウンドロビンでもよいし、データ蓄積装置２０のデータキュー管理部２１２から、各データ蓄積装置２０の負荷状態変更通知（入力スループット等を示した情報）を取得し、キュー負荷の大きいデータ蓄積装置２０から優先的に選択するようにしてもよい。
また、この選択処理部１１２は、データ蓄積装置２０から、リクエスト拒絶通知を受信した場合は、そのリクエスト拒絶通知を送信したデータ蓄積装置２０を除いた他のデータ蓄積装置２０を取得先情報１２０の中から選択する。
【００７４】
データ取得処理部１１３は、選択処理部１１２により選択されたデータ蓄積装置２０に対し、データ取得要求を送信し、そのデータ蓄積装置２０からストリームデータを取得する。なお、このデータ取得要求を送信するタイミングは、例えば、自身のフィルタ処理装置１０において処理しているデータがなくなったときに実行される。つまり、フィルタ処理装置１０が自律的に、自身の処理負荷が少ない適切なタイミングで、データ蓄積装置２０に、データ取得要求を送信することができる。
【００７５】
フィルタ処理部１１４は、データ取得処理部１１３から取得したストリームデータに付されたＳＩＤに基づき、フィルタ処理を実行する。なお、このフィルタ処理部１１４は、このストリームデータに対し実行すべきフィルタ処理機能が自身のフィルタ処理装置１０にないときは、このフィルタ処理機能を実現するためのファイルデータを、共有情報管理装置３０から取得し、機能追加を行うことで、フィルタ処理を実行することができる。そして、このフィルタ処理部１１４は、入出力部経由で、フィルタ処理後のストリームデータをデータベース装置６０に送信する。また、フィルタ処理部１１４は、ストリームデータのフィルタ処理を完了すると、フィルタ処理完了通知を、そのストリームデータを送信したデータ蓄積装置２０に送信する。
【００７６】
＜共有情報管理装置＞
共有情報管理装置３０は、フィルタ処理装置１０が各種データをフィルタ処理する機能を実現するためのフィルタデータを記憶する。そして、フィルタ処理装置１０からの要求に基づき、その要求を送信したフィルタ処理装置１０にフィルタデータを送信する。この共有情報管理装置３０は、各種データの入出力を司り、入出力インタフェースにより実現される入出力部と、ファイルデータ等を記憶する記憶部と、フィルタ処理装置１０からの要求を受付け、ファイルデータ等の送信の制御を行う処理部と含んだコンピュータ装置により実現される。
【００７７】
≪動作手順≫
次に、本実施形態に係るストリームデータ処理システム１の全体の処理の流れについて、図１を参照しつつ、図３を用いて詳細に説明する。
【００７８】
まず、クライアント５０は、情報源から計測データやログデータ等のストリームデータを取得して、そのデータにＳＩＤを付与する。そして、クライアント５０は、そのストリームデータの送信先となるデータ蓄積装置２０の割当要求を、キュー割当装置４０に送信する（ステップＳ１０１）。
【００７９】
次に、キュー割当装置４０の接続先割当部４１は、割当要求を受信すると、ラウンドロビン等の所定の方法で、割当先となるデータ蓄積装置２０を決定し（ステップＳ１０２）、その割当要求を送信したクライアント５０に、決定したデータ蓄積装置２０を割当先情報として返信する（ステップＳ１０３）。また、キュー割当装置４０の割当先通知部４２が、接続先割当部４１が割り当てたストリームデータの送信先となるデータ蓄積装置２０を、割当先情報通知として、各フィルタ処理装置１０（１０Ａ，１０Ｂ，…，１０Ｎ）に送信する（ステップＳ１０４）。
【００８０】
そして、割当先情報を受信したクライアント５０は、割当先情報に示されるデータ蓄積装置２０に、ＳＩＤを付したストリームデータを送信する（ステップＳ１０５）。
【００８１】
データ蓄積装置２０のデータ受付処理部２１１は、クライアント５０から、ストリームデータを受け付けたか否かを判定する（ステップＳ１０６）。ここで、ストリームデータを受け付けていなければ（ステップＳ１０６→Ｎｏ）、ストリームデータがクライアント５０から送信されるまで待つ。一方、ストリームデータを受け付けた場合（ステップＳ１０６→Ｙｅｓ）、データ受付処理部２１１は、受け付けたストリームデータをデータキュー管理部２１２へ出力する。
【００８２】
データキュー管理部２１２は、データ受付処理部２１１から受け取ったストリームデータを、データキュー２２１に記憶する（ステップＳ１０７）。
【００８３】
一方、フィルタ処理装置１０の取得先情報管理部１１１は、キュー割当装置４０からの割当先情報通知を受信したか否かを判定する（ステップＳ１０８）。ここで、割当先情報通知を受信していなければ（ステップＳ１０８→Ｎｏ）、キュー割当装置４０から送信されるまで待つ。一方、割当先情報通知を受信した場合（ステップＳ１０８→Ｙｅｓ）、取得先情報管理部１１１は、割当先情報通知に示される、ストリームデータを蓄積しているデータ蓄積装置２０を、取得先情報１２０に記憶する（ステップＳ１０９）。
【００８４】
そして、フィルタ処理装置１０の選択処理部１１２は、例えば、これから処理対象となるストリームデータの前に取得していたストリームデータのフィルタ処理が終了したことを契機として、取得先情報１２０に示されるデータ蓄積装置２０の中から、ストリームデータの取得先となるデータ蓄積装置２０を選択し（ステップＳ１１０）、データ取得処理部１１３に出力する。
【００８５】
続いて、データ取得処理部１１３は、選択処理部１１２により選択されたデータ蓄積装置２０に対し、データ取得要求を送信する（ステップＳ１１１）。
【００８６】
次に、データ蓄積装置２０のフィルタ受付判定部２１５は、フィルタ処理装置１０からストリームデータのデータ取得要求を受信し、そのデータ取得要求を受信した時点において、フィルタ情報キュー２２２に記憶されているデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長以下か否かを判定する（ステップＳ１１２）。
【００８７】
そして、フィルタ情報キュー２２２のデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長を超えている場合に（ステップＳ１１２→Ｎｏ）、フィルタ受付判定部２１５は、その受信したデータ取得要求を受け付けず、そのデータ取得要求を送信したフィルタ処理装置１０にリクエスト拒絶通知を送信する（ステップＳ１１３）。
【００８８】
続いて、フィルタ処理装置１０の選択処理部１１２は、リクエスト拒絶通知を受信したか否かを判定する（ステップＳ１１４）。ここで、リクエスト拒絶通知を受信していない場合は（ステップＳ１１４→Ｎｏ）、リクエスト拒絶通知を受信するまで待つ。一方、リクエスト拒絶通知を受信している場合には（ステップＳ１１４→Ｙｅｓ）、ステップＳ１１０に戻り、選択処理部１１２が、取得先情報１２０の中からそのリクエスト拒絶通知を送信したデータ蓄積装置２０を除いた他のデータ蓄積装置２０を選択し（ステップＳ１１０）、その新たに選択したデータ蓄積装置２０にデータ取得要求を送信する（ステップＳ１１１）。
【００８９】
これに対し、ステップＳ１１２において、フィルタ情報キュー２２２のデータ取得要求の数が、フィルタ情報キュー２２２における所定のキュー最大長以下の場合に（ステップＳ１１２→Ｙｅｓ）、フィルタ受付判定部２１５は、受信したデータ取得要求を受け付けて、フィルタ情報キュー管理部２１６に出力する。そして、フィルタ情報キュー管理部２１６は、受け取ったデータ取得要求を、フィルタ情報キュー２２２に記憶する（ステップＳ１１５）。続いて、フィルタ受付判定部２１５は、リクエスト処理部２１４を起動させる（ステップＳ１１６）。
【００９０】
次に、リクエスト処理部２１４は、フィルタ情報キュー２２２の先頭から、フィルタ情報キュー管理部２１６を介して、データ取得要求を取得する（ステップＳ１１７）。そして、リクエスト処理部２１４は、データキュー２２１からデータキュー管理部２１２を介してストリームデータを取得し、取得したストリームデータを、ステップＳ１１７で取得したデータ取得要求に示されるフィルタ処理装置１０に対して送信する（ステップＳ１１８）。
【００９１】
続いて、フィルタ処理装置１０のデータ取得処理部１１３は、ストリームデータを受信し、そのストリームデータをフィルタ処理部１１４に出力する。そして、フィルタ処理部１１４は、ストリームデータに付与されたＳＩＤに基づき、そのストリームデータに対して行うべきフィルタ処理を判断して実行する（ステップＳ１１９）。そして、フィルタ処理部１１４は、フィルタ処理されたストリームデータを、所定のデータベース装置６０に出力する。
【００９２】
続いて、フィルタ処理部１１４は、フィルタ処理を終えると、そのストリームデータを送信したデータ蓄積装置２０のリクエスト処理部２１４に、フィルタ処理完了通知を送信する（ステップＳ１２０）。
【００９３】
次に、フィルタ処理完了通知を受信したデータ蓄積装置２０のリクエスト処理部２１４は、データキュー管理部２１２を介して、フィルタ処理装置１０でフィルタ処理済みのストリームデータをデータキュー２２１から削除する（ステップＳ１２１）。
【００９４】
そして、データ蓄積装置２０のデータキュー管理部２１２は、データキュー２２１に１つ以上のストリームデータが蓄積されているか否かを判定する（ステップＳ１２２）。具体的には、データキュー管理部２１２は、所定時間、データキュー２２１に、ストリームデータが蓄積されていない状態が続いているか否かを判定する。ここで、データキュー２２１にストリームデータが蓄積されている場合は（ステップＳ１２２→Ｙｅｓ）、ステップＳ１０６に戻る。一方、データキュー２２１にストリームデータが所定時間、蓄積されていない状態が続いた場合は（ステップＳ１２２→Ｎｏ）、取得先情報更新通知をフィルタ処理装置１０の取得先情報管理部１１１に送信する（ステップＳ１２３）。
【００９５】
フィルタ処理装置１０の取得先情報管理部１１１は、取得先情報更新通知を受信すると、その取得先情報更新通知を送信したデータ蓄積装置２０を、取得先情報１２０から削除する更新を行う（ステップＳ１２４）。そして、ステップＳ１０８に戻る。
【００９６】
このようにすることで、本実施形態に係るストリームデータ処理システム、ストリームデータ処理方法、データ蓄積装置、およびプログラムによれば、特定のデータ蓄積装置２０に、フィルタ処理装置１０からのデータ取得要求が集中した場合であっても、フィルタ情報キュー２２２における所定のキュー最大長を超えるデータ取得要求を受け付けないため、データ蓄積装置２０でのデータキューの入出力レスポンスを低下させることがなく、データ受信待ちのフィルタ処理装置１０の増加を防ぐことができる。その結果、システム全体のスループットの低下を防ぐことができる。
【００９７】
次に、図４を用いて、データ蓄積装置２０のフィルタ情報キュー管理部２１６による、フィルタ情報キュー２２２のキュー最大長更新処理について説明する。図４は、本実施形態に係るフィルタ情報キュー２２２のキュー最大長更新処理の流れを示すフローチャートである。
【００９８】
ここで、データ蓄積装置２０のデータキュー管理部２１２は、データ受付処理部２１１からストリームデータを受け取り、データキュー２２１にデータを記憶する毎に、そして、リクエスト処理部２１４からの指示により、データキュー２２１からストリームデータを出力する毎に、その入力時刻と出力時刻とを入出力データ情報２２３に記憶する。そして、この入出力データ情報２２３を用いて、フィルタ情報キュー管理部２１６は、以下に示すフィルタ情報キュー２２２のキュー最大長更新処理を行い、フィルタ情報キュー２２２に記憶するデータ取得要求のキュー最大長を所定の時間間隔毎に設定する。
【００９９】
まず、フィルタ情報キュー管理部２１６は、記憶部２２内の入出力データ情報２２３を参照し、所定時間内のストリームデータの入力数および出力数を抽出する（ステップＳ２０１）。
【０１００】
続いて、フィルタ情報キュー管理部２１６は、抽出した入力数および出力数に基づき、前記した（式１）または（式２）を用いてフィルタ情報キュー２２２のキュー最大長（Ｍ）を設定する（ステップＳ２０２）。
【０１０１】
具体的には、フィルタ情報キュー管理部２１６は、入力数（Ｘ）＜出力数（Ｙ）の場合には、フィルタ情報キュー２２２のキュー最大長（Ｍ）を、前記した（式１）を用いて算出する。
【０１０２】
また、フィルタ情報キュー管理部２１６は、入力数（Ｘ）≧出力数（Ｙ）の場合には、
フィルタ情報キュー２２２のキュー最大長（Ｍ）を、前記した（式２）を用いて算出する。
【０１０３】
図５は、本実施形態に係るフィルタ情報キュー管理部２１６によるフィルタ情報キュー２２２のキュー最大長（Ｍ）の算出を説明するための図である。図５に例示するように、入出力データ情報２２３は、所定の時間間隔として１０秒毎に、データキュー２２１へのストリームデータの入力時刻と出力時刻とをレコード（rec）として記憶している。そして、所定の時間間隔として設定された１０秒間のうち、定数αが２の場合、例えば、入力レコード数が「５」、出力レコード数が「５」のときは、（式２）を用いて、フィルタ情報キュー管理部２１６は、フィルタ情報キュー２２２のキュー最大長（Ｍ）＝（５／５）＊２＝２を算出する。入力レコード数が「１２」、出力レコード数が「４」のときは、（式２）を用いて、フィルタ情報キュー２２２のキュー最大長（Ｍ）＝（１２／４）＊２＝６を算出する。入力レコード数が「３」、出力レコード数が「６」のときは、（式１）を用いて、フィルタ情報キュー２２２のキュー最大長（Ｍ）＝１を算出する。そして、フィルタ情報キュー管理部２１６は、算出したキュー最大長（Ｍ）を、フィルタ情報キュー２２２の新たなキュー最大長として設定する。
【０１０４】
図４に戻り、ステップＳ２０３において、フィルタ情報キュー管理部２１６は、（式１）または（式２）を用いて新たに設定したフィルタ情報キュー２２２のキュー最大長と、その時点においてフィルタ情報キュー２２２に記憶されているデータ取得要求の数とを比較して、フィルタ情報キュー２２２に記憶されているデータ取得要求の数の方が大きいか否かを判定する。ここで、フィルタ情報キュー２２２のデータ取得要求の数が、新たに設定したキュー最大長以下の場合は（ステップＳ２０３→Ｎｏ）、特に処理を行わず、ステップＳ２０１へ戻る。
【０１０５】
一方、フィルタ情報キュー２２２のデータ取得要求の数が、新たに設定したキュー最大長より大きい場合は（ステップＳ２０３→Ｙｅｓ）、フィルタ情報キュー管理部２１６は、フィルタ情報キュー２２２に記憶されているデータ取得要求の数が、ステップＳ２０２で設定したキュー最大長になるまで、フィルタ情報キュー２２２の先頭または後方からデータ取得要求を取得する。そして、その取得したデータ取得要求に示されるフィルタ処理装置１０に、リクエスト拒絶通知を送信する（ステップＳ２０４）。
【０１０６】
なお、リクエスト拒絶通知を受信したフィルタ処理装置１０は、選択処理部１１２が、リクエスト拒絶通知を送信したデータ蓄積装置２０以外の他のデータ蓄積装置２０を、取得先情報１２０から選択する。そして、データ取得処理部１１３により、他のデータ蓄積装置２０にデータ取得要求が送信される。
【０１０７】
このようにすることで、データ蓄積装置２０は、データキュー２２１へのストリームデータの入力数および出力数に応じて、フィルタ情報キュー２２２のキュー最大長を設定することができ、フィルタ情報キュー２２２に記憶された状態で、待ち状態にあるデータ取得要求の数を低減させることができる。その結果、データ受信待ちのフィルタ処理装置１０を減らすことができる。
【０１０８】
なお、本実施形態に係るストリームデータ処理システム、ストリームデータ処理方法、データ蓄積装置、およびプログラムは、例えば、以下のようなログ監視システムに利用することができる。
【０１０９】
メールシステム、Ｗｅｂシステム、ＤＢシステム、監視システム、物流システム、勘定システム等の各種情報処理システムにおいて、そのシステムの異常をログファイルに出力し、そのログファイルを運用監視システムで監視することが日常的に行われている。また、複数のシステム間やシステム内の各サーバ間、またはクライアントとシステム間に流れるパケット情報をモニタリングして、そのシステムの異常を監視することも行われている。このようなログを集中管理するシステムでは、ログの非定常状態や異常状態等をいち早く発見し、管理者に通知する必要がある。このシステムに本発明に係るストリームデータ処理システム１を適用した場合、監視対象システムは、逐次、ログをクライアントを介してストリームデータ処理システム１に送信する。そして、そのログデータを受け取ったストリームデータ処理システム１のデータ蓄積装置２０は、データキュー２２１にデータを記憶し、フィルタ処理装置１０が、データ蓄積装置２０からデータを取得し、フィルタ処理することで、非定常状態や異常状態等があれば、アラートをシステムの管理者に送信することができる。
【０１１０】
このとき、何らかの原因で、特定のデータ蓄積装置２０に対して、フィルタ処理装置１０からのデータ処理要求が増加した場合であっても、データ蓄積装置２０は、フィルタ情報キュー２２２における所定のキュー最大長を超えるデータ取得要求を受け付けないため、データキュー２２１の入出力のレスポンスを低下させることがない。そして、データ処理待ちのフィルタ処理装置１０の増加を防ぐことができるため、システム全体としてのスループットの低下を防ぐことができる。したがって、ログ監視において、非定常状態や異常状態等を管理者へのアラートを、システム内で停滞させることなく送信することができる。
【符号の説明】
【０１１１】
１，５  ストリームデータ処理システム
１０    フィルタ処理装置
１１，２１  処理部
１２，２２  記憶部
２０    データ蓄積装置
３０    共有情報管理装置
４０    キュー割当装置
４１    接続先割当部
４２    割当先通知部
５０    クライアント
６０    データベース装置
７０    参照クライアント
１００，５００  システム
１１１  取得先情報管理部
１１２  選択処理部
１１３  データ取得処理部
１１４  フィルタ処理部
１２０  取得先情報
２１１  データ受付処理部
２１２  データキュー管理部
２１３  フィルタ受付部
２１４  リクエスト処理部
２１５  フィルタ受付判定部
２１６  フィルタ情報キュー管理部
２２１  データキュー
２２２  フィルタ情報キュー
２２３  入出力データ情報
【図１】
0005385811.tif 000002
【図２】
0005385811.tif 000003
【図３】
0005385811.tif 000004
【図４】
0005385811.tif 000005
【図５】
0005385811.tif 000006
【図６】
0005385811.tif 000007
【図７】
0005385811.tif 000008
【図８】
0005385811.tif 000009