"異種情報源を複数有し、データを処理する分散データ処理装置であって、前記異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置にキューイングする複数の追記手段と、前記追記手段からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理手段と、を有し、前記追記手段は、前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO順にキューイングし、当該データに対する前記処理手段による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、前記データ記憶装置から前記処理手段によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、前記処理手段から自身に対する空振り率を取得する空振り率取得手段と、を具備し、前記処理手段は、空振り率を格納する空振り率記憶装置と、ネットワーク内に存在するすべての追記手段の中から追記手段を1つ選択し、選択した追記手段のデータ記憶装置からFIFO順で情報源IDとともにデータを取得Pullするデータ取得手段と、取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理手段と、前記追記手段からデータ取得する度に直近NNは追記手段毎に予め指定する整数回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、を具備することを特徴とする分散データ処理装置。前記処理手段は、前記データ取得手段が前記追記手段からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、前記追記手段からのデータ取得時に、当該追記手段に当該追記手段が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記手段に対して脱退要求を送出する脱退要求手段と、を含み、前記追記手段は、全ての処理手段から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理手段を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、前記チケットを取得した処理手段を当該グループに所属させ、該グループに所属する処理手段のデータ取得対象を自追記手段だけとなるようにグループリストに登録するグループ化手段と、前記処理手段から前記脱退要求を取得した場合は、前記グループに所属する処理手段を前記グループリストから削除するグループ解除手段と、を含む請求項1記載の分散データ処理装置。前記チケット生成手段は、生成するチケット数が0以上で、いずれのグループにも所属していない処理手段の数よりも少ない範囲内でチケットを生成し、前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記手段が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む請求項1または2記載の分散データ処理装置。前記処理手段の前記脱退要求手段は、前記空振り率記憶装置の当該追記手段に対する空振り率が予め設定した定数γよりも大きければ、前記グループ解除手段によりグループを離脱していずれのグループにも所属しないものとし、前記処理手段に前記脱退要求を送信する請求項1または2記載の分散データ処理装置。複数の異種情報源と、該異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置キューにキューイングする複数の追記装置と、該追記装置からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理装置と、を有し、データを分散して処理するデータ分散システムであって、前記追記装置は、前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO順にキューイングし、当該データに対する前記処理装置による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、前記データ記憶装置から前記処理装置によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、前記処理装置から自身に対する空振り率を取得する空振り率取得手段と、を具備し、前記処理装置は、空振り率を格納する空振り率記憶装置と、ネットワーク内に存在するすべての追記装置の中から追記装置を1つ選択し、選択した追記装置のデータ記憶装置からFIFO順で情報源IDとともにデータを取得するデータ取得手段と、取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理装置と、前記追記装置からデータ取得する度に直近NNは追記装置毎に予め指定する整数回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、を具備することを特徴とする分散データ処理システム。前記処理装置は、前記データ取得手段が前記追記装置からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、前記追記装置からのデータ取得時に、当該追記装置に当該追記装置が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記装置に対して脱退要求を送出する脱退要求手段と、を含み、前記追記装置は、全ての処理装置から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理装置を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、前記チケットを取得した処理装置を当該グループに所属させ、該グループに所属する処理装置のデータ取得対象を自追記装置だけとなるようにグループリストに登録するグループ化手段と、前記処理装置から前記脱退要求を取得した場合は、前記グループに所属する処理装置を前記グループリストから削除するグループ解除手段と、を含む請求項5記載の分散データ処理システム。前記チケット生成手段は、生成するチケット数が0以上で、いずれのグループにも所属していない処理装置の数よりも少ない範囲内でチケットを生成し、前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記装置が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む請求項5または6記載の分散データ処理システム。コンピュータを、請求項1乃至4のいずれか1項に記載の分散データ処理装置の各手段として機能させるための分散データ処理プログラム。一般に各情報源から送信されるデータの入力レートは時間とともに変化する。更に送信されたデータのサイズや内容の違いによってフィルタ部で実行される処理の処理負荷処理時間も一定ではないことが知られている。前述の非特許文献1の技術では機能等質なフィルタ部がすべての追記部をデータ取得対象としているため、情報源から送信されるデータの入力レートが大きく異なる場合や情報源から送信されるデータのサイズや内容の違いによってフィルタ部での処理負荷処理時間が大きく異なるデータ処理が混在する場合に特定の情報源から送信されたデータ処理ばかりにフィルタ部が占有されてしまい、その他の情報源から送信されたデータ処理の遅延が大きくなるという問題がある。本発明は、上記の点に鑑みなされたもので、大量のセンサーデータや各種情報システムのログ情報、Webやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種データストリームが時々刻々と大量に生成されるネットワーク上で、データを収集蓄積し、利用者の目的に応じた処理を行う際に処理遅延を最小にする分散データ処理装置及びシステム及びプログラムを実現することを目的とする。上記の課題を解決するため、本発明請求項1は、ネットワーク上の複数の異種情報源から取得したデータを分散処理する分散データ処理装置であって、前記異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置にキューイングする複数の追記手段と、前記追記手段からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理手段と、を有し、前記追記手段は、前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO順にキューイングし、当該データに対する前記処理手段による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、前記データ記憶装置から前記処理手段によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、前記処理手段から自身に対する空振り率を取得する空振り率取得手段と、を具備し、前記処理手段は、空振り率を格納する空振り率記憶装置と、ネットワーク内に存在するすべての追記手段の中から追記手段を1つ選択し、選択した追記手段のデータ記憶装置からFIFO順で情報源IDとともにデータを取得Pullするデータ取得手段と、取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理手段と、前記追記手段からデータ取得する度に直近NNは追記手段毎に予め指定する整数回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、を具備する。また、本発明請求項2は、前記処理手段において、前記データ取得手段が前記追記手段からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、前記追記手段からのデータ取得時に、当該追記手段に当該追記手段が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記手段に対して脱退要求を送出する脱退要求手段と、を含み、前記追記手段において、全ての処理手段から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理手段を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、前記チケットを取得した処理手段を当該グループに所属させ、該グループに所属する処理手段のデータ取得対象を自追記手段だけとなるようにグループリストに登録するグループ化手段と、前記処理手段から前記脱退要求を取得した場合は、前記グループに所属する処理手段を前記グループリストから削除するグループ解除手段と、を含む。また、本発明請求項3は、前記チケット生成手段において、生成するチケット数が0以上で、いずれのグループにも所属していない処理手段の数よりも少ない範囲内でチケットを生成し、前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記手段が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む。また、本発明請求項4は、前記処理手段の前記脱退要求手段において、前記空振り率記憶装置の当該追記手段に対する空振り率が予め設定した定数γよりも大きければ、前記グループ解除手段によりグループを離脱していずれのグループにも所属しないものとし、前記処理手段に前記脱退要求を送信する。本発明によれば大量のセンサーデータや各種情報システムのログ情報、Webやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種データストリームが時々刻々と大量に生成されるネットワーク上で、追記部がデータを収集蓄積し、各情報源から送信されるデータの入力レートの変化に応じて処理部がデータ取得対象となる追記部を変化させながら利用者の目的に応じた処理を行うことで処理遅延を最小にする分散データ処理方式を実現することができる。以下図面と共に、本発明の実施の形態を説明する。図1は、本発明の一実施形態の分散データ処理システムのシステム構成図である。同図に示す分散データ処理システムは、複数の追記部1001~100nと処理部2001~200mから構成され、各追記部100では、データの量、データサイズが均一とは限らない異種情報源からデータを取得する。各追記部100は、データ記憶装置110、データ受付部120、入力レート計算部130、出力レート計算部140、チケット生成部150、空振り率取得部160、グループ化部170、グループ解除部180から構成される。各処理部200は、データ取得部210、データ処理部220、空振り率記録部230、チケット取得部240、空振り率記憶装置250から構成される。図2は、本発明の一実施の形態における概要動作のシーケンスチャートである。本実施形態の分散データ処理システムは、追記部100のデータ受付部120において情報源から送信されたデータに情報源IDとデータ受け付け時の時刻印を付与してデータ記憶装置キュー110にFIFO順にキューイングし、当該データに対する処理部200による処理が完了するまでデータ記憶装置キュー110に保持し続けるステップ1。なお、情報源IDは、取得したデータに予め付与されている場合は、当該IDを用い、付与されていない場合には、予め追記部100内に情報源とそのIDを対応付けたテーブルを用意しておくことにより情報源IDを付与するものとする。追記部100の入力レート計算部130では、取得したデータに関する入力レートを計算するステップ2。次に、チケット生成部150は、処理部200から空振り率を取得しステップ3、後述する図3のS4の方法によりチケット生成を行うステップ4。なお、当該チケット生成処理を行うことにより、ある追記部100が多くの枚数のチケットを発行すれば、多くの処理部200が当該追記部100のみからデータを取得するようになるため、当該追記部100に入力されたデータの処理遅延を最小にすることができる。処理部200のデータ取得部210は、ネットワーク内に存在するすべての追記部の中から例えば、ラウンドロビンアルゴリズムや追記部の入力レートの高さなどから追記部100を1つ選択し、選択した追記部100のデータ記憶装置キュー110からFIFO順で情報源IDとともにデータを取得Pullしステップ5、また、チケットがある場合には、チケットも取得する。取得したデータに対して情報源IDごとに予め定義された処理を実行する。当該処理が終了すると、追記部100に対してデータ処理完了通知を送出する。データを送出した追記部100の出力レート計算部140は、当該データ処理完了通知を受け取ると、処理部200に送出したデータに関する出力レートの計算を行う。グループ化部170は、ステップ2により求められた入力レートとステップ6により求められた出力レート及び、処理部200から取得した空振り率に基づいて、グループ化処理を行う。処理部200の空振り率記録部230は、追記部100からデータ取得する度に直近NNは追記部ごとに予め指定する整数回分の空振り率を自身の空振り率記憶装置空振り率カウンタ250に記録する空振り率の値は追記部100ごとに記録する。なお、ここで、空振り率とは、ある処理部200がある追記部100に対してデータ取得要求した際に、当該追記部100のデータ記憶装置キューにデータが存在しない場合に、当該処理部200のデータ取得要求が失敗したこれを空振りが発生したいう割合である。直近N回分の空振り率とは、ある処理部200が追記部100にデータ取得要求をした直近N回のうち、空振りが発生した割合を指し、データ取得を試みた直近N回のうち空振りが発生した回数/Nにより求める。処理部200のチケット取得部240は、追記部100から取得したチケットについて、空振り率記憶装置250の空振り率に基づいて、グループを離脱するか否かを判定する。脱退する場合は、追記部100に対して脱退要求を送信し、すべての追記部をデータ取得対象とする。一方、脱退しない場合はデータを取得した追記部100のみをデータ取得対象とするステップ14。尚、本実施の形態のデータ記憶装置キュー110は追記部100と同じサーバ上に配置されているが、他のサーバ上に配置することも可能である。更に、空振り率記憶装置空振り率カウンター250は、処理部200と同じサーバ上に配置されているが、他のサーバ上に配置することも可能である。また、本実施の形態の追記部100と処理部200の比率はn:mnおよびmは自然数としているが、情報源の増加や減少に応じて追記部100の数を増加させたり減少させることが可能であり、処理量処理負荷の増加や減少に応じて処理部の数を増加させたり減少させることが可能である。図3は、本発明の一実施の形態における追記部の動作のフローチャートである。S1から始まる処理は、追記部100がデータ受付部120により情報源からデータを受け付ける際の動作を示している。ステップ101データ受付部120は、情報源からのデータ送信があるまで待機し、データを受信した場合は、ステップ102に移行する。ステップ102データを受け付けて、情報源IDと時刻印を付与する。ステップ103データをデータ記憶装置110にFIFO順にキューイングする。ステップ104入力レート計算部130の入力レート計算のため、メモリ図示せず内の入力カウンタを1増加させ、ステップ101に戻る。S2から始まる処理は、処理部200からのデータ取得要求に対応し、空振り率取得部160が、処理部200から自身に対する空振り率を取得する際の動作を示している。なお、当該空振り率は、前回の処理において、処理部200の空振り率記録部230によりデータ処理の結果に基づいて空振り率が求められ、空振り率記憶装置250に格納されているものとする。ステップ201空振り率取得部160は、処理部200からデータ取得要求があるまで待機し、要求があった場合にステップ202に移行する。ステップ202処理部200からのデータ取得要求を受け付ける。ステップ203データ記憶装置110の当該データのステータスを取得済み処理完了待ちにする。ステップ204処理部200の空振り率記憶装置空振り率カウンタ250から空振り率を取得し、メモリに格納し、ステップ201移行する。S3から始まる処理は、処理部200からデータの処理完了通知を受けた際の動作を示している。ステップ301出力レート計算部140は、処理部200から処理完了通知があったらステップ302に移行する。ステップ302データ記憶装置110から当該処理完了通知に対応するデータを削除する。ステップ303出力レート計算部140での出力レート計算のためのメモリ内の出力カウンタを1つ増加させ、ステップ301に移行する。S4から始まる処理は、予め指定した時間が経過するごとに入力レート計算部130によりデータの入力レートを計算し、出力レート計算部140によりデータの出力レートを計算し、すべての処理部200から取得した空振り率から空振り率の集計結果Aを計算し、自身の入力レートと出力レートの比率Bを計算し、チケット生成部240によりチケットを生成する際の動作を示している。ステップ401入力レート計算部130は、予め指定した時間が経過したかを判定し、経過した場合には、ステップ402に移行する。ステップ402入力レート計算部130は、メモリから入力カウンタ値を取得し、データ記憶装置110からデータを取得して、入力されたデータの単位時間辺りの入力レートを計算する。入力レートの算出例として、例えば、により求めることができる。当該方法以外にも、入力カウンタに直近T単位時間Tは予め指定する整数の各単位時間における入力データ数を入力データ数記憶装置に記憶しておき、それら入力データ数の移動平均を算出して入力レートする方法がある。ステップ403出力レート計算部140は、メモリ図示せずから出力カウンタ値を取得し、データ記憶装置110からデータを取得して、当該データの単位時間当たりの出力レートを計算する。出力レートの算出例として、例えば、により求めることができる。なお、入力レートの場合と同様に、上記の算出方法に限定されるものではない。ステップ404チケット生成部150は、全ての処理部200から取得したメモリ内の空振り率から空振り率の集計結果Aを計算する。ステップ405チケット生成部150は、自身の入力レートと出力レートの比率Bを計算する。ステップ404チケット生成部150は、生成するチケット数が0以上、かつ、いずれのグループにも所属していない処理部の数-1以下となる範囲内で、上記のA,Bの計算結果に応じた枚数のチケットを生成する。このとき、空振り率の集計結果Aが予め設定した定数αよりも大きければ、生成するチケット数の範囲内で、定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置110の入力レートと出力レートの比率Bが予め設定した定数βよりも大きければ、生成するチケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす。次に、処理部200の動作を説明する。図4は、本発明の一実施の形態における処理部の動作のフローチャートである。S5から始まる処理は、データ取得部210により取得対象の追記部100の中から追記部を1つ選択してデータを取得し、空振り率記録部160によって当該追記部100に対する直近N回分の空振り率を更新して空振り率記憶装置空振り率カウンター250に記録する。チケットが存在すればチケット取得240によってチケットを取得し、追記部100からデータを取得できた空振りではない場合は、データ処理部220により情報源IDごとに予め定義された処理を行いステップ509、その後で追記部100に処理完了を通知する。追記部100からチケットを取得できた場合は、チケットを取得できなかった場合は、ステップ501に戻る。一方、追記部100からデータを取得できなかった空振りした場合は、当該追記部100に対する空振り率が予め設定した定数γより大きい場合に、追記部100のグループ解除部180対して、脱退要求を送出する。これにより、グループ解除部180は、グループリストから当該処理部200を削除し、ステップ501に戻る。これにより当該処理部200は、再びすべての追記部100をデータ取得対象とすることになる。なお、上記の図1の追記部100、処理部200の各構成要素の動作をプログラムとして構築し、分散データ処理装置として利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。本発明は、上記の実施の形態に限定されることなく、特許請求の範囲内において、種々変更応用が可能である。"