(57)【要約】【課題】  ネットワーク上のデータを収集・蓄積し、利用者の目的に応じた処理を行う際に処理遅延を最小にする。【解決手段】  本発明は、処理手段において、追記手段からデータを取得できなかった際の空振り率を空振り率記憶手段に記録し、追記手段からのデータ取得時に、当該追記手段に当該追記手段が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得し、空振り率に基づいて、取得したチケットのグループに所属するか否かを判定し、所属しないと判定した場合には、追記手段に対して脱退要求を送出する。追記手段において、全ての処理手段から取得した空振り率の集計結果を計算し、自身の入力レートと自身の出力レートから、チケットを生成し、チケットを取得した処理手段をグループリストに登録し、処理手段から脱退要求を取得した場合は、グループリストから削除する。【選択図】  図１
2013117867.tif 000002
【特許請求の範囲】
【請求項１】
異種情報源を複数有し、データを処理する分散データ処理装置であって、
前記異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置にキューイングする複数の追記手段と、
前記追記手段からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理手段と、
を有し、
前記追記手段は、
前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO順にキューイングし、当該データに対する前記処理手段による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、
前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、
前記データ記憶装置から前記処理手段によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、
前記処理手段から自身に対する空振り率を取得する空振り率取得手段と、
を具備し、
前記処理手段は、
空振り率を格納する空振り率記憶装置と、
ネットワーク内に存在するすべての追記手段の中から追記手段を1つ選択し、選択した追記手段のデータ記憶装置からFIFO順で情報源IDとともにデータを取得（Pull）するデータ取得手段と、
取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理手段と、
前記追記手段からデータ取得する度に直近N（Nは追記手段毎に予め指定する整数）回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、
を具備することを特徴とする分散データ処理装置。
【請求項２】
前記処理手段は、
前記データ取得手段が前記追記手段からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、
前記追記手段からのデータ取得時に、当該追記手段に当該追記手段が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、
前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記手段に対して脱退要求を送出する脱退要求手段と、
を含み、
前記追記手段は、
全ての処理手段から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理手段を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、
前記チケットを取得した処理手段を当該グループに所属させ、該グループに所属する処理手段のデータ取得対象を自追記手段だけとなるようにグループリストに登録するグループ化手段と、
前記処理手段から前記脱退要求を取得した場合は、前記グループに所属する処理手段を前記グループリストから削除するグループ解除手段と、
を含む請求項１記載の分散データ処理装置。
【請求項３】
前記チケット生成手段は、
生成するチケット数が0以上で、いずれのグループにも所属していない処理手段の数よりも少ない範囲内でチケットを生成し、
前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記手段が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む
請求項１または２記載の分散データ処理装置。
【請求項４】
前記処理手段の前記脱退要求手段は、
前記空振り率記憶装置の当該追記手段に対する空振り率が予め設定した定数γよりも大きければ、前記グループ解除手段によりグループを離脱していずれのグループにも所属しないものとし、前記処理手段に前記脱退要求を送信する
請求項１または２記載の分散データ処理装置。
【請求項５】
複数の異種情報源と、該異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置（キュー）にキューイングする複数の追記装置と、該追記装置からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理装置と、を有し、データを分散して処理するデータ分散システムであって、
前記追記装置は、
前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO順にキューイングし、当該データに対する前記処理装置による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、
前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、
前記データ記憶装置から前記処理装置によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、
前記処理装置から自身に対する空振り率を取得する空振り率取得手段と、
を具備し、
前記処理装置は、
空振り率を格納する空振り率記憶装置と、
ネットワーク内に存在するすべての追記装置の中から追記装置を1つ選択し、選択した追記装置のデータ記憶装置からFIFO順で情報源IDとともにデータを取得するデータ取得手段と、
取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理装置と、
前記追記装置からデータ取得する度に直近N（Nは追記装置毎に予め指定する整数）回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、
を具備することを特徴とする分散データ処理システム。
【請求項６】
前記処理装置は、
前記データ取得手段が前記追記装置からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、
前記追記装置からのデータ取得時に、当該追記装置に当該追記装置が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、
前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記装置に対して脱退要求を送出する脱退要求手段と、
を含み、
前記追記装置は、
全ての処理装置から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理装置を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、
前記チケットを取得した処理装置を当該グループに所属させ、該グループに所属する処理装置のデータ取得対象を自追記装置だけとなるようにグループリストに登録するグループ化手段と、
前記処理装置から前記脱退要求を取得した場合は、前記グループに所属する処理装置を前記グループリストから削除するグループ解除手段と、
を含む請求項５記載の分散データ処理システム。
【請求項７】
前記チケット生成手段は、
生成するチケット数が0以上で、いずれのグループにも所属していない処理装置の数よりも少ない範囲内でチケットを生成し、
前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記装置が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む
請求項５または６記載の分散データ処理システム。
【請求項８】
コンピュータを、
請求項１乃至４のいずれか１項に記載の分散データ処理装置の各手段として機能させるための分散データ処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、分散データ処理装置及びシステム及びプログラムに係り、特に、大量のセンサーデータや各種情報システムのログ情報、Webやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種データストリームが時々刻々と大量に生成されるネットワーク上で、データを収集・蓄積し、利用者の目的に応じた処理を行う分散データ処理装置及びシステム及びプログラムに関する。
【背景技術】
【０００２】
負荷分散やスケールアウト、可用性の向上を目的とした分散データ処理方式として、クライアントから入力されたデータ（レコード）を受け付ける追記部と、追記部が受け付けたレコードを処理するフィルタ部とが、それぞれ別プロセスとしてネットワーク上に分散配置され、ネットワーク内のプロセス構成を管理する情報部がネットワーク内のプロセス構成を記録したプロセス構成情報をマルチキャストでネットワーク内の全プロセスに配信することでプロセスの生存状況や新規プロセスの追加状況を確認し、最新のプロセス構成に更新したプロセス構成情報を配信することを定期的に行い、追記部ではクライアントから入力されたレコードに対して管理用のレコードIDおよび時刻印を付与した後でキューにレコードを蓄積し、フィルタ部では情報部からのマルチキャストで受信したプロセス構成情報の中から接続可能な追記部の一覧を取得し、ラウンドロビンアルゴリズムなどによって追記部に接続し、キューから情報源IDとともにレコードを取得（Pull）し、情報源IDに応じて予め決められた処理を行うステップを繰り返す技術がある(例えば、非特許文献1参照)。このように非特許文献1の技術では、追記部がフィルタ部に対してレコードを配信（Push）する方式ではなく、フィルタ部が主体的に追記部に蓄積されたレコードを取得（Pull）する方式を採用することで、フィルタ部が過負荷になることを防ぎ、フィルタ部が自らのタイミングで処理の制御（処理の実行、停止、再起動）を行えるようにしている。また、すべてのフィルタ部は機能等質に構成されているため、どの追記部から取得したレコードでも処理ができるようになっている。
【先行技術文献】
【非特許文献】
【０００３】
【非特許文献１】「追記・参照型データ管理システムの設計と評価」，赤間 他，情報処理学会論文誌 Vol.49 No.2，2008年2月，pp749-764．
【発明の概要】
【発明が解決しようとする課題】
【０００４】
一般に各情報源から送信されるデータの入力レートは時間とともに変化する。更に送信されたデータのサイズや内容の違いによってフィルタ部で実行される処理の処理負荷（処理時間）も一定ではないことが知られている。
【０００５】
前述の非特許文献1の技術では機能等質なフィルタ部がすべての追記部をデータ取得対象としているため、情報源から送信されるデータの入力レートが大きく異なる場合や情報源から送信されるデータのサイズや内容の違いによってフィルタ部での処理負荷（処理時間）が大きく異なるデータ処理が混在する場合に特定の情報源から送信されたデータ処理ばかりにフィルタ部が占有されてしまい、その他の情報源から送信されたデータ処理の遅延が大きくなるという問題がある。
【０００６】
本発明は、上記の点に鑑みなされたもので、大量のセンサーデータや各種情報システムのログ情報、Webやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種データストリームが時々刻々と大量に生成されるネットワーク上で、データを収集・蓄積し、利用者の目的に応じた処理を行う際に処理遅延を最小にする分散データ処理装置及びシステム及びプログラムを実現することを目的とする。
【課題を解決するための手段】
【０００７】
上記の課題を解決するため、本発明（請求項１）は、ネットワーク上の複数の異種情報源から取得したデータを分散処理する分散データ処理装置であって、
前記異種情報源から時々刻々と生成される時系列データを受け付けて自身のデータ記憶装置にキューイングする複数の追記手段と、
前記追記手段からデータを取得し、取得したデータに対して情報源毎に予め定義された処理を実行する複数の処理手段と、
を有し、
前記追記手段は、
前記情報源からデータを受け付け、情報源IDとデータ受け付け時の時刻印を管理情報として該データに付与し、前記データ記憶装置にFIFO(First In First Out)順にキューイングし、当該データに対する前記処理手段による処理が完了するまで自身のデータ記憶装置にデータを保持し続けるデータ受付手段と、
前記データ記憶装置に入力されたデータの単位時間当たりの入力レートを計算する入力レート計算手段と、
前記データ記憶装置から前記処理手段によって取り出されたデータの単位時間当たりの出力レートを計算する出力レート計算手段と、
前記処理手段から自身に対する空振り率を取得する空振り率取得手段と、
を具備し、
前記処理手段は、
空振り率を格納する空振り率記憶装置と、
ネットワーク内に存在するすべての追記手段の中から追記手段を1つ選択し、選択した追記手段のデータ記憶装置からFIFO順で情報源IDとともにデータを取得（Pull）するデータ取得手段と、
取得したデータに対して情報源ID毎に予め定義された処理を実行するデータ処理手段と、
前記追記手段からデータ取得する度に直近N（Nは追記手段毎に予め指定する整数）回分の空振り率を前記空振り率記憶装置に記録する空振り率記録手段と、を具備する。
【０００８】
また、本発明（請求項２）は、前記処理手段において、
前記データ取得手段が前記追記手段からデータを取得できなかった際の空振り率を空振り率記憶手段に記録する手段と、
前記追記手段からのデータ取得時に、当該追記手段に当該追記手段が自身が所属するグループに参加させるためのチケットが存在すればチケットを取得するチケット取得手段と、
前記空振り率に基づいて、取得した前記チケットのグループに所属するか否かを判定し、所属しないと判定した場合には、前記追記手段に対して脱退要求を送出する脱退要求手段と、
を含み、
前記追記手段において、
全ての処理手段から取得した空振り率の集計結果を計算し、前記入力レート計算手段により取得した自身の入力レートと前記出力レート計算手段により取得した自身の出力レートから、データを取得した処理手段を自身が所属するグループに参加させるためのチケットを生成するチケット生成手段を具備し、
前記チケットを取得した処理手段を当該グループに所属させ、該グループに所属する処理手段のデータ取得対象を自追記手段だけとなるようにグループリストに登録するグループ化手段と、
前記処理手段から前記脱退要求を取得した場合は、前記グループに所属する処理手段を前記グループリストから削除するグループ解除手段と、を含む。
【０００９】
また、本発明（請求項３）は、前記チケット生成手段において、
生成するチケット数が0以上で、いずれのグループにも所属していない処理手段の数よりも少ない範囲内でチケットを生成し、
前記空振り率の集計結果が予め設定した定数αよりも大きければ、前記追記手段が生成するチケット数の範囲内で定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置の入力レートと出力レートの比率が予め設定した定数βよりも大きければ、チケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす手段を含む。
【００１０】
また、本発明（請求項４）は、前記処理手段の前記脱退要求手段において、
前記空振り率記憶装置の当該追記手段に対する空振り率が予め設定した定数γよりも大きければ、前記グループ解除手段によりグループを離脱していずれのグループにも所属しないものとし、前記処理手段に前記脱退要求を送信する。
【発明の効果】
【００１１】
本発明によれば大量のセンサーデータや各種情報システムのログ情報、Webやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種データストリームが時々刻々と大量に生成されるネットワーク上で、追記部がデータを収集・蓄積し、各情報源から送信されるデータの入力レートの変化に応じて処理部がデータ取得対象となる追記部を変化させながら利用者の目的に応じた処理を行うことで処理遅延を最小にする分散データ処理方式を実現することができる。
【図面の簡単な説明】
【００１２】
【図１】本発明の一実施の形態における分散データ処理システムの構成図である。
【図２】本発明の一実施の形態における概要動作のフローチャートである。
【図３】本発明の一実施の形態における追記部の動作のフローチャートである。
【図４】本発明の一実施の形態における処理部の動作のフローチャートである。
【発明を実施するための形態】
【００１３】
以下図面と共に、本発明の実施の形態を説明する。
【００１４】
図１は、本発明の一実施形態の分散データ処理システムのシステム構成図である。
【００１５】
同図に示す分散データ処理システムは、複数の追記部１００１〜１００ｎと処理部２００１〜２００mから構成され、各追記部１００では、データの量、データサイズが均一とは限らない異種情報源からデータを取得する。
【００１６】
各追記部１００は、データ記憶装置(キュー)１１０、データ受付部１２０、入力レート計算部１３０、出力レート計算部１４０、チケット生成部１５０、空振り率取得部１６０、グループ化部１７０、グループ解除部１８０から構成される。
【００１７】
各処理部２００は、データ取得部２１０、データ処理部２２０、空振り率記録部２３０、チケット取得部２４０、空振り率記憶装置２５０から構成される。
【００１８】
図２は、本発明の一実施の形態における概要動作のシーケンスチャートである。
【００１９】
本実施形態の分散データ処理システムは、追記部１００のデータ受付部１２０において情報源から送信されたデータに情報源IDとデータ受け付け時の時刻印を付与してデータ記憶装置（キュー）１１０にFIFO順にキューイングし、当該データに対する処理部２００による処理が完了するまでデータ記憶装置（キュー）１１０に保持し続ける（ステップ１）。なお、情報源ＩＤは、取得したデータに予め付与されている場合は、当該ＩＤを用い、付与されていない場合には、予め追記部１００内に情報源とそのＩＤを対応付けたテーブルを用意しておくことにより情報源ＩＤを付与するものとする。
【００２０】
追記部１００の入力レート計算部１３０では、取得したデータに関する入力レートを計算する（ステップ２）。次に、チケット生成部１５０は、処理部２００から空振り率を取得し（ステップ３）、後述する図３のS4の方法によりチケット生成を行う（ステップ４）。なお、当該チケット生成処理を行うことにより、ある追記部１００が多くの枚数のチケットを発行すれば、多くの処理部２００が当該追記部１００のみからデータを取得するようになるため、当該追記部１００に入力されたデータの処理遅延を最小にすることができる。
【００２１】
処理部２００のデータ取得部２１０は、ネットワーク内に存在するすべての追記部の中から（例えば、ラウンドロビンアルゴリズムや追記部の入力レートの高さなどから）追記部１００を1つ選択し、選択した追記部１００のデータ記憶装置（キュー）１１０からFIFO順で情報源IDとともにデータを取得（Pull）し（ステップ５）、また、チケットがある場合には、チケットも取得する(ステップ６)。取得したデータに対して情報源IDごとに予め定義された処理を実行する(ステップ７)。当該処理が終了すると、追記部１００に対してデータ処理完了通知を送出する。データを送出した追記部１００の出力レート計算部１４０は、当該データ処理完了通知を受け取ると、処理部２００に送出したデータに関する出力レートの計算を行う(ステップ８)。
【００２２】
グループ化部１７０は、ステップ２により求められた入力レートとステップ６により求められた出力レート及び、処理部２００から取得した空振り率に基づいて、グループ化処理を行う(ステップ９)。
【００２３】
処理部２００の空振り率記録部２３０は、追記部１００からデータ取得する度に直近N（Nは追記部ごとに予め指定する整数）回分の空振り率を自身の空振り率記憶装置（空振り率カウンタ）２５０に記録する（空振り率の値は追記部１００ごとに記録する）(ステップ１０)。なお、ここで、空振り率とは、ある処理部２００がある追記部１００に対してデータ取得要求した際に、当該追記部１００のデータ記憶装置（キュー）にデータが存在しない場合に、当該処理部２００のデータ取得要求が失敗した（これを「空振りが発生した」いう）割合である。直近Ｎ回分の空振り率とは、ある処理部２００が追記部１００にデータ取得要求をした直近Ｎ回のうち、空振りが発生した割合を指し、
データ取得を試みた直近Ｎ回のうち空振りが発生した回数／Ｎ
により求める。
【００２４】
処理部２００のチケット取得部２４０は、追記部１００から取得したチケットについて、空振り率記憶装置２５０の空振り率に基づいて、グループを離脱するか否かを判定する(ステップ１１)。脱退する場合は、追記部１００に対して脱退要求を送信し、すべての追記部をデータ取得対象とする(ステップ１３)。一方、脱退しない場合はデータを取得した追記部１００のみをデータ取得対象とする（ステップ１４）。
【００２５】
尚、本実施の形態のデータ記憶装置（キュー）１１０は追記部１００と同じサーバ上に配置されているが、他のサーバ上に配置することも可能である。更に、空振り率記憶装置（空振り率カウンター）２５０は、処理部２００と同じサーバ上に配置されているが、他のサーバ上に配置することも可能である。また、本実施の形態の追記部１００と処理部２００の比率はn：m（nおよびmは自然数）としているが、情報源の増加や減少に応じて追記部１００の数を増加させたり減少させることが可能であり、処理量（処理負荷）の増加や減少に応じて処理部の数を増加させたり減少させることが可能である。
【００２６】
図３は、本発明の一実施の形態における追記部の動作のフローチャートである。
【００２７】
S1から始まる処理は、追記部１００がデータ受付部１２０により情報源からデータを受け付ける際の動作を示している。
【００２８】
ステップ１０１）  データ受付部１２０は、情報源からのデータ送信があるまで待機し、データを受信した場合は、ステップ１０２に移行する。
【００２９】
ステップ１０２）  データを受け付けて、情報源ＩＤと時刻印を付与する。
【００３０】
ステップ１０３）  データをデータ記憶装置１１０にＦＩＦＯ順にキューイングする。
【００３１】
ステップ１０４）  入力レート計算部１３０の入力レート計算のため、メモリ（図示せず)内の入力カウンタを１増加させ、ステップ１０１に戻る。
【００３２】
S2から始まる処理は、処理部２００からのデータ取得要求に対応し、空振り率取得部１６０が、処理部２００から自身に対する空振り率を取得する際の動作を示している。なお、当該空振り率は、前回の処理において、処理部２００の空振り率記録部２３０によりデータ処理の結果に基づいて空振り率が求められ、空振り率記憶装置２５０に格納されているものとする。
【００３３】
ステップ２０１）  空振り率取得部１６０は、処理部２００からデータ取得要求があるまで待機し、要求があった場合にステップ２０２に移行する。
【００３４】
ステップ２０２）  処理部２００からのデータ取得要求を受け付ける。
【００３５】
ステップ２０３）  データ記憶装置１１０の当該データのステータスを「取得済み」「処理完了待ち」にする。
【００３６】
ステップ２０４）  処理部２００の空振り率記憶装置（空振り率カウンタ）２５０から空振り率を取得し、メモリ(図示せず)に格納し、ステップ２０１移行する。
【００３７】
S3から始まる処理は、処理部２００からデータの処理完了通知を受けた際の動作を示している。
【００３８】
ステップ３０１）  出力レート計算部１４０は、処理部２００から処理完了通知があったらステップ３０２に移行する。
【００３９】
ステップ３０２）  データ記憶装置１１０から当該処理完了通知に対応するデータを削除する。
【００４０】
ステップ３０３）  出力レート計算部１４０での出力レート計算のためのメモリ(図示せず)内の出力カウンタを１つ増加させ、ステップ３０１に移行する。
【００４１】
S4から始まる処理は、予め指定した時間が経過するごとに入力レート計算部１３０によりデータの入力レートを計算し、出力レート計算部１４０によりデータの出力レートを計算し、すべての処理部２００から取得した空振り率から空振り率の集計結果Ａを計算し、自身の入力レートと出力レートの比率Ｂを計算し、チケット生成部２４０によりチケットを生成する際の動作を示している。
【００４２】
ステップ４０１）  入力レート計算部１３０は、予め指定した時間が経過したかを判定し、経過した場合には、ステップ４０２に移行する。
【００４３】
ステップ４０２）  入力レート計算部１３０は、メモリ(図示せず)から入力カウンタ値を取得し、データ記憶装置１１０からデータを取得して、入力されたデータの単位時間辺りの入力レートを計算する。入力レートの算出例として、例えば、
【００４４】
【数１】
2013117867.tif 000003
により求めることができる。当該方法以外にも、入力カウンタに直近Ｔ単位時間（Ｔは予め指定する整数）の各単位時間における入力データ数を入力データ数記憶装置に記憶しておき、それら入力データ数の移動平均を算出して入力レートする方法がある。
【００４５】
ステップ４０３）  出力レート計算部１４０は、メモリ（図示せず）から出力カウンタ値を取得し、データ記憶装置１１０からデータを取得して、当該データの単位時間当たりの出力レートを計算する。出力レートの算出例として、例えば、
【００４６】
【数２】
2013117867.tif 000004
により求めることができる。なお、入力レートの場合と同様に、上記の算出方法に限定されるものではない。
【００４７】
ステップ４０４）  チケット生成部１５０は、全ての処理部２００から取得したメモリ(図示せず)内の空振り率から空振り率の集計結果Ａを計算する。
【００４８】
ステップ４０５）  チケット生成部１５０は、自身の入力レートと出力レートの比率Ｂを計算する。
【００４９】
ステップ４０４）  チケット生成部１５０は、生成するチケット数が0以上、かつ、いずれのグループにも所属していない処理部の数−１以下となる範囲内(追記部１００が生成するチケット数は０以上、かつ、いずれのグループにも所属していない処理部の数−１以下となる範囲)で、上記のＡ，Ｂの計算結果に応じた枚数のチケットを生成する。このとき、空振り率の集計結果Ａが予め設定した定数αよりも大きければ、生成するチケット数の範囲内で、定数αからの大きさに応じて生成するチケット数を減らし、自身のデータ記憶装置１１０の入力レートと出力レートの比率Ｂが予め設定した定数βよりも大きければ、生成するチケット数の範囲内で定数βからの大きさに応じて生成するチケット数を増やす。
【００５０】
次に、処理部２００の動作を説明する。
【００５１】
図４は、本発明の一実施の形態における処理部の動作のフローチャートである。
【００５２】
S5から始まる処理は、データ取得部２１０により取得対象の追記部１００の中から追記部を1つ選択して(ステップ５０１)データを取得し(ステップ５０２)、空振り率記録部１６０によって当該追記部１００に対する直近N回分の空振り率を更新して空振り率記憶装置（空振り率カウンター）２５０に記録する(ステップ５０３)。チケットが存在すれば(ステップ５０４、Ｙｅｓ)チケット取得２４０によってチケットを取得し(ステップ５０５)、
追記部１００からデータを取得できた（空振りではない）場合は(ステップ５０６、Ｙｅｓ)、データ処理部２２０により情報源IDごとに予め定義された処理を行い（ステップ５０９）、その後で追記部１００に処理完了を通知する(ステップ５１０)。追記部１００からチケットを取得できた場合は(ステップ５１１、Ｙｅｓ）、追記部１００のグループ化部１７０は、当該処理部２００のデータ取得対象を当該追記部（チケットを生成した追記部）だけとなるようにグループリストを更新し(ステップ５１２)、チケットを取得できなかった場合は(ステップ５１１、Ｎｏ)、ステップ５０１に戻る。
【００５３】
一方、追記部１００からデータを取得できなかった（空振りした）場合は(ステップ５０６、Ｎｏ)、当該追記部１００に対する空振り率が予め設定した定数γより大きい場合に(ステップ５０７、Ｙｅｓ)、追記部１００のグループ解除部１８０対して、脱退要求を送出する。これにより、グループ解除部１８０は、グループリストから当該処理部２００を削除し(ステップ５０８)、ステップ５０１に戻る。これにより当該処理部２００は、再びすべての追記部１００をデータ取得対象とすることになる。
【００５４】
なお、上記の図１の追記部１００、処理部２００の各構成要素の動作をプログラムとして構築し、分散データ処理装置として利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【００５５】
本発明は、上記の実施の形態に限定されることなく、特許請求の範囲内において、種々変更・応用が可能である。
【符号の説明】
【００５６】
１００  追記部（追記装置）
１１０  データ記憶装置(キュー)
１２０  入力レート計算部
１３０  出力レート計算部
１４０  出力レート計算部
１５０  チケット生成部
１６０  空振り率取得部
１７０  グループ化部
１８０  グループ解除部
２００  処理部（処理装置）
２１０  データ取得部
２２０  データ処理部
２３０  空振り率記録部
２４０  チケット取得部
２５０  空振り率記憶装置（空振り率カウンタ）
【図１】
2013117867.tif 000005
【図２】
2013117867.tif 000006
【図３】
2013117867.tif 000007
【図４】
2013117867.tif 000008