(57)【要約】【課題】データストリーム管理システムのスループットを向上させるデータストリーム複合制御システム、制御装置、データストリーム複合制御方法およびプログラムを提供する。【解決手段】ＤＳキュー３１は、逐次に継続して発生するデータストリームを受信した単位で記憶し、受信した順に出力する。ＤＳ複合部３２は記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合する。ＤＳ処理部３５は、複合したデータをまとめて所定の処理を行う。制御装置２の処理ネットワーク監視部４は、ノード３のそれぞれの状態を検知し、処理ネットワーク分析部６は、所定の条件に適合するように、ノード３それぞれのデータ複合方法を設定する。ＤＳ複合方法指示部８は、設定したデータ複合方法をノードのそれぞれに送信する。各ノード３は、データ複合方法を制御装置２から受信したデータ複合方法に変更する。【選択図】図１
2011082807.tif 000002
【特許請求の範囲】
【請求項１】
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードと、
前記複数の処理ノードのそれぞれの状態を検知するノード状態検知手段、所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定手段、および、前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信手段、を備える制御装置と、
を含み、
前記処理ノードは、前記データ複合手段のデータ複合方法を前記制御装置から受信したデータ複合方法に変更する、
ことを特徴とするデータストリーム複合制御システム。
【請求項２】
前記ノード状態検知手段は、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定手段は、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする請求項１に記載のデータストリーム複合制御システム。
【請求項３】
前記設定手段は、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合手段の精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合手段のデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合手段のデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする請求項１または２に記載のデータストリーム複合制御システム。
【請求項４】
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知手段と、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定手段と、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信手段と、
を備えることを特徴とする制御装置。
【請求項５】
前記ノード状態検知手段は、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定手段は、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする請求項４に記載の制御装置。
【請求項６】
前記設定手段は、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合手段の精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合手段のデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合手段のデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする請求項４または５に記載の制御装置。
【請求項７】
逐次に継続して発生するデータであるデータストリームを受信するデータ受信ステップ、前記データストリームを受信した順に受信した単位でデータストリームキューに記憶する記憶ステップ、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合ステップ、および、前記データ複合ステップで複合したデータをまとめて所定の処理を行うデータ処理ステップ、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知ステップと、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定ステップと、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信ステップと、
を備え、
前記処理ノードは、前記データ複合ステップのデータ複合方法を前記制御装置から受信したデータ複合方法に変更する複合方法変更ステップを備える、
ことを特徴とするデータストリーム複合制御方法。
【請求項８】
前記ノード状態検知ステップは、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定ステップは、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする請求項７に記載のデータストリーム複合制御方法。
【請求項９】
前記設定ステップは、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合ステップの精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合ステップのデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合ステップのデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする請求項７または８に記載のデータストリーム複合制御方法。
【請求項１０】
コンピュータに、
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知ステップと、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定ステップと、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信ステップと、
を実行させることを特徴とするプログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、逐次に継続して生成され続けるデータを、逐次的に分析するデータストリーム処理システムに関する。より詳しくは、処理ノード間を流通するデータの複合処理を制御するデータストリーム複合制御システム、制御装置、データストリーム複合制御方法およびプログラムに関する。
【背景技術】
【０００２】
データストリームとは、逐次に継続して生成され続けるデータである。一般に、大量のデータを処理するためのシステムとして、データベース管理システム（以下、ＤＢＭＳ：Data Base Management System ）が用いられる。しかし、データの追加・更新が限定的である静的なデータを処理することに特化したシステムであるＤＢＭＳでは、動的に生成され続けるデータストリームを効率よく処理することができない。そこで、このような動的に生成され続けるデータを処理するために、データストリーム管理システム（以下、ＤＳＭＳ：Data Stream Management System ）と呼ばれるシステムの技術が用いられる。例えば、非特許文献１および非特許文献２には、ＤＳＭＳに関連する技術として、連続クエリを実行する複数のノードから成る処理ネットワークによるストリーム処理技術が記載されている。
【０００３】
特許文献１は、テレビジョン装置の復号処理の技術であるが、復号処理の負荷が異なる複数の復号モードの中から、符号化データの記憶状態に応じて復号モードを選択することが記載されている。特許文献１の対象である符号化データは、連続的に受信するデータストリームである。
【０００４】
特許文献２は、複数のストリームデータを受信し復号化するための技術が記載されている。特許文献２のデータストリーム処理装置は、基準周波数を生成するために必要な基準時間情報を、符号化側と復号化側のどちらも外部の同一の基準時間局から取得し、同一の周波数で符号化処理、復号化処理を行う。
【０００５】
特許文献３は、リアルタイム性を要求されるコンテンツをネットワーク経由で転送する技術が記載されている。特許文献３のデータ転送方法は、送信側の装置および受信側の装置それぞれが、バッファリングによってデータが途切れないように処理を制御し、データが途切れそうになったら送信側および受信側の少なくとも一方の処理を中断させてバッファ蓄積量の回復を待ち、ある程度のデータ量が蓄積されたら処理を再開させる。
【０００６】
特許文献４は、ネットワークの構成情報をノードの監視情報とともに所定のノードから取得するネットワーク監視装置の技術が記載されている。特許文献４のネットワーク監視装置は、運用中の構成情報の変更を監視し、構成情報の切替時に新構成情報に切り替えが完了するまではネットワークの監視情報に基づいて作成された構成情報に非依存の情報を参照して旧構成情報を表示する。さらに新構成情報に切り替えた後、識別情報中に不当な構成情報を有するノードがネットワーク内に存在した場合、正当な構成情報をそのノードへ転送してその不当な構成情報と切り替えさせる。
【０００７】
特許文献５ないし特許文献７には、通信経路のトラフィックを制御する技術が記載されている。特許文献５には、例えば、中継サーバにアクセスするユーザの数（回線数）が多いときには、通信回線が混雑していると判断し、データの間引きを行い、データ量を減らして通信回線の混雑を回避することが記載されている。特許文献６の多重伝送装置は、多重伝送路のトラフィック量が所定値未満のときに各送信ノードが制御性を重視した第１のプライオリティを選択し、トラフィック量が所定値以上のときに各送信ノードが安全性を重視した第２のプライオリティを選択するように切り替える。
【０００８】
また、特許文献７のデータフロー制御システムは、複数のデータストリームが複数のパスを介してそれぞれ上流側ノードから下流側ノードへ送出される通信網において、それぞれのパス上のデータストリームの最大平均速度を指定するための下流側ノード用の指定手段を有する。指定手段は、バッファの使用状態と前回の速度変更を考慮した時点からの経過時間とに基づいて、パス上のデータストリームの最大平均速度の変更を考慮する時点を決定し、速度が変更されるべきデータパスを識別する。
【先行技術文献】
【特許文献】
【０００９】
【特許文献１】特開２００２−０１０２６２号公報
【特許文献２】特開２００３−２５９３６５号公報
【特許文献３】特開２００３−３４８１７５号公報
【特許文献４】特開平０９−２８９５０８号公報
【特許文献５】特開平１１−２３１８６６号公報
【特許文献６】特開平０４−３３９４３８号公報
【特許文献７】特開昭６２−１０４２４５号公報
【非特許文献】
【００１０】
【非特許文献１】The STREAM Group, "STREAM: The Stanford Stream Data Manager", IEEE Data Engineering Bulletin, 2003.
【非特許文献２】Daniel J. Abadi, et al., "Aurora: a new model and architecture for data stream management", VLDB journal, 2003.
【発明の概要】
【発明が解決しようとする課題】
【００１１】
図１０は、非特許文献１や非特許文献２に記載されているＤＳＭＳの構成例を示すブロック図である。ＤＳＭＳは、データストリーム処理の最小単位であるノードを複数備える処理ネットワークから構成される。また、各ノードは、それぞれ、ＤＳキューと、ＤＳ複合部と、ＤＳ処理部とを有する。
【００１２】
ＤＳキューは、ノードが入力したデータを一時的に蓄積するバッファである。ＤＳ複合部は、所定のデータ複合方法に基づいて、ＤＳキューから所定量のデータを取り出して１つの複合データに出力する。ＤＳ処理部は、複合データを統計処理する機能を備える。ＤＳキューは、外部または処理ネットワーク内の他のノードからデータを受信し、その受信したデータをキュー形式で格納する。
【００１３】
ＤＳ複合部は、ＤＳキューを監視し所定のＤＳ複合方法に基づいて、ＤＳキューから所定量のデータを取り出して１つの複合データを出力する。ＤＳ複合方法は、ＤＳキューに格納されたデータを所定条件で所定量の複合データに複合する方法である。ＤＳ複合部は例えば、所定時間毎（例えば、１分間毎）に取得した所定個数（例えば、１０個）のデータを所定グループ化方法（例えば、１０個のデータを１つにグループ化する）で複合する。また、ＤＳ複合部は、複合データをＤＳ処理部に転送する機能を備える。
【００１４】
ＤＳ処理部は、ＤＳ複合部から受け取ったデータを統計処理する。そして、ＤＳ処理部は、統計処理したデータを、外部または処理ネットワーク内の次のノードのＤＳキューに出力する。以上のように動作することで、ＤＳＭＳでは、ノードによって構成された処理ネットワーク間をデータが流れつつ、その経路上でデータが処理される。
【００１５】
しかしながら、非特許文献１や非特許文献２に記載されたＤＳＭＳでは、個々のノードが並列に処理を実行しているとともに、ノードによって各処理の負荷が異なる。そのため、ノード毎にデータのスループットが異なる。また、ＤＳＭＳでは、ノード間をデータが流れることによってデータが処理される。従って、ＤＳＭＳ全体のスループットは、最もスループットの低いノードのスループットに律速してしまうという問題がある。
【００１６】
本発明は、上記事情に鑑みてなされたものであり、データストリーム管理システム（ＤＳＭＳ）における各ノードのスループット格差を是正し、ＤＳＭＳ全体のスループットを向上させるデータストリーム複合制御システム、制御装置、データストリーム複合制御方法およびプログラムを提供することを目的とする。
【課題を解決するための手段】
【００１７】
本発明の第１の観点に係るデータストリーム複合制御システムは、
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードと、
前記複数の処理ノードのそれぞれの状態を検知するノード状態検知手段、所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定手段、および、前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信手段、を備える制御装置と、
を含み、
前記処理ノードは、前記データ複合手段のデータ複合方法を前記制御装置から受信したデータ複合方法に変更する、
ことを特徴とする。
【００１８】
本発明の第２の観点に係る制御装置は、
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知手段と、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定手段と、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信手段と、
を備えることを特徴とする。
【００１９】
本発明の第３の観点に係るデータストリーム複合制御方法は、
逐次に継続して発生するデータであるデータストリームを受信するデータ受信ステップ、前記データストリームを受信した順に受信した単位でデータストリームキューに記憶する記憶ステップ、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合ステップ、および、前記データ複合ステップで複合したデータをまとめて所定の処理を行うデータ処理ステップ、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知ステップと、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定ステップと、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信ステップと、
を備え、
前記処理ノードは、前記データ複合ステップのデータ複合方法を前記制御装置から受信したデータ複合方法に変更する複合方法変更ステップを備える、
ことを特徴とする。
【００２０】
本発明の第４の観点に係るプログラムは、コンピュータに、
逐次に継続して発生するデータであるデータストリームを受信するデータ受信手段、前記データストリームを受信した単位で記憶し、受信した順に出力するデータストリームキュー、前記データストリームキューに記憶したデータストリームの中から、連続する所定の数の受信した単位のデータのうち所定の数のデータを選択して複合するデータ複合手段、および、前記データ複合手段で複合したデータをまとめて所定の処理を行うデータ処理手段、を備える複数の処理ノードのそれぞれの状態を検知するノード状態検知ステップと、
所定の条件に適合するように、前記複数の処理ノードそれぞれのデータ複合方法を設定する設定ステップと、
前記設定したデータ複合方法を表す情報を前記処理ノードのそれぞれに送る送信ステップと、
を実行させることを特徴とする。
【発明の効果】
【００２１】
本発明によれば、データストリーム管理システムにおける各ノードのスループット格差を是正し、データストリーム管理システム全体のスループットを向上させることができる。
【図面の簡単な説明】
【００２２】
【図１】本発明の実施の形態に係るデータストリーム複合制御システムの構成例を示すブロック図である。
【図２】監視情報記憶部が格納する各ノードの監視結果のデータ構造例を示す図である。
【図３】選択ＤＳ複合方法一覧情報の一例を示す図である。
【図４】選択可能ＤＳ複合方法一覧情報の一例を示す図である。
【図５】ＤＳ制約情報の一例を示す図である。
【図６】実施の形態に係る制御装置の動作の一例を示すフローチャートである。
【図７】実施の形態に係るＤＳ複合方法の変更判定処理の一例を示すフローチャートである。
【図８】実施の形態に係るノードのデータ処理の動作の一例を示すフローチャートである。
【図９】本発明の実施の形態に係る制御装置のハードウェア構成の一例を示すブロック図である。
【図１０】ＤＳＭＳの構成例を示すブロック図である。
【発明を実施するための形態】
【００２３】
（実施の形態）
図１は、本発明の実施の形態に係るデータストリーム複合制御システムの構成例を示すブロック図である。なお、本実施の形態において、データストリーム管理システム（以下、ＤＳＭＳ）は、データストリーム複合制御方法を用いたデータストリーム複合制御システム１によって実現される。以下の説明では、データストリーム複合制御システム１とＤＳＭＳを同一視して、データストリーム複合制御システム１をＤＳＭＳということがある。また、データストリームを略してＤＳということがある。
【００２４】
図１に示すように、本実施の形態において、データストリーム複合制御システム１は、データストリーム処理を行う処理ネットワーク９と、処理ネットワーク９の複数のノード３を制御する制御装置２とから構成される。ノード３はデータストリーム処理の最小単位である。処理ネットワーク９は、複数のノード３を含む。なお、以下、各ノード３を包括的に表現する場合、または各ノード３のいずれかを指す場合に、単にノード３という。
【００２５】
制御装置２や各ノード３は、具体的には例えば、プログラムに従って動作するワークステーションやパーソナルコンピュータ等の情報処理装置によって実現される。制御装置２と処理ネットワーク９とは、１つの情報処理装置を用いて実現される場合がある。また、制御装置２を１つの情報処理装置を用いて実現するとともに、処理ネットワーク９を他の１つの情報処理装置を用いて実現してもよい。さらに、制御装置２を１つの情報処理装置を用いて実現するとともに、処理ネットワーク９に含まれる各ノード３を、それぞれ別々の情報処理装置を用いて実現してもよい。また、制御装置２と一部のノード３が１つの情報処理装置を用いて実現され、他のノード３がそれぞれ別々の情報処理装置を用いて実現される等、ＤＳＭＳの実現形態は問わない。
【００２６】
また、データストリーム複合制御システム１によって実現されるＤＳＭＳは、例えば、複数のＲＦＩＤタグを荷物等に付与し、各ＲＦＩＤタグからの位置情報をトレースすることによって荷物を管理する物流管理システムの用途に適用できる。また、データストリーム複合制御システム１は、例えば、ＧＰＳセンサ等を搭載した複数の車両（プローブカー）からの速度情報や位置情報を収集し解析することによって、渋滞情報案内や経路探索サービスを提供する交通管理システムの用途に適用できる。また、データストリーム複合制御システム１は、例えば、複数のサーバが生成するログ情報を収集して解析するシステム運用管理システムの用途に適用できる。
【００２７】
本実施の形態において、データストリームとは、時々刻々と逐次に継続して生成され続ける大量のデータである。例えば、物流管理システムの用途に適用する場合、データストリーム複合制御システム１は、逐次に継続して運ばれてくる荷物のＲＦＩＤタグから時々刻々と読み取った荷物の識別子や位置情報を、データストリームとして処理する。また、例えば、交通管理システムの用途に適用する場合、データストリーム複合制御システム１は、プローブとなる車両から継続的に時々刻々と収集する車両の識別子や位置情報、速度情報、車両状況情報（ワイパーの動作状態等）を、データストリームとして統計処理する。また、例えば、システム運用管理システムの用途に適用する場合、データストリーム複合制御システム１は、各サーバから収集するログ情報を、データストリームとして統計処理する。
【００２８】
ノード３は、それぞれ、ＤＳキュー３１と、ＤＳ複合部３２と、ＤＳ複合方法記憶部３３と、ＤＳ複合変更部３４と、ＤＳ処理部３５を含む。図１では、代表して１つのノード３にだけノード３に含まれる各部を記載しているが、処理ネットワーク９に含まれるノード３は、それぞれ処理の内容は異なるが同様の構成を備える。
【００２９】
ＤＳキュー３１は、具体的には、ハードディスク装置やメモリ等の記憶装置を用いて、リングバッファなどのデータ構造によって実現される。ＤＳキュー３１は、ノード３が受信したデータを受信した単位でキュー形式で一時的に蓄積するバッファである。ＤＳキュー３１は、蓄積したデータストリームを受信した順序でＤＳ複合部３２に出力する。
【００３０】
ＤＳ複合方法記憶部３３は、具体的には、ハードディスク装置やメモリ等の記憶装置によって実現される。ＤＳ複合方法記憶部３３は、複数のＤＳ複合方法を記憶する。ＤＳ複合方法は、ＤＳキュー３１に格納された連続する所定の数の受信した単位のデータのうち、所定の数のデータを選択して所定条件で所定量の複合データに複合する方法である。具体的には、所定時間毎（例えば、１分間毎）に取得した所定個数（例えば、１０個）の受信した単位のデータを所定グループ化方法（例えば、１０個の受信した単位のデータを１つにグループ化する）で複合する。
【００３１】
データ複合方法の所定時間や所定個数、所定グループ化方法によって、それぞれ複合データのデータ量やデータ精度は異なる。例えば、グループ化を行う所定の時間間隔が１分間毎と１０分間毎とでは、複合対象となるデータ量は異なる。また、複合する所定個数の違い（例えば、１０個、１００個など）によっても、複合データのデータ量は異なる。
【００３２】
また、所定グループ化方法によって、複合データのデータ精度を維持する場合と劣化する場合がある。例えば、１０個のデータをすべてグループ化するグループ化方法は複合データのデータ精度を維持するが、ランダムに抽出した５個のみのデータをグループ化するグループ化方法は複合データのデータ精度が５０％劣化する。また、データの桁数を落としてグループ化する方法も複合データのデータ精度を劣化する。例えば、倍精度型のデータを単精度型のデータに型変換して複合するグループ化方法は、データ精度が５０％劣化する。
【００３３】
ＤＳ複合部３２は、ＤＳ複合方法記憶部３３に記憶された１つのデータ複合方法に基づいて、ＤＳキュー３１から所定量のデータを取り出してひとまとまりのデータに複合してＤＳ処理部３５に転送する機能を備える。なお、複数のノード３のＤＳ複合部３２のすべてが同一のＤＳ複合方法に基づく制約はない。データ量が少ないデータ複合方法に基づくＤＳ複合部３２と、データ量が多いデータ複合方法に基づくＤＳ複合部３２が、１つの処理ネットワーク９に混在してもよい。
【００３４】
ＤＳ複合変更部３４は、制御装置２のＤＳ複合方法指示部８から受信したコマンドに応じて、ＤＳ複合部３２のデータ複合方法を変更する機能を備える。データ複合方法の変更は、例えば、ＤＳキュー３１から取り出す受信した単位のデータの連続する所定の数、取り出したデータから複合するために選択する所定の数と選択する方法、データを複合するときにデータ型を変換するかどうか、またその変換するデータ型などの項目を含む。
【００３５】
ＤＳ処理部３５は、ＤＳ複合部３２から転送された複合データを統計処理する機能を備える。例えば、複合データに含まれるデータの値が８０であるデータの個数ＳＵＭを求める場合、ＤＳ処理部３５は複合データ内のデータをすべてチェックして、値が８０であるデータの個数をカウントする。そして、ＤＳ処理部３５は、そのカウント値を解ＳＵＭとして求める。
【００３６】
ＤＳ処理部３５は、複合データのデータ量とデータ精度によって、それぞれ処理負荷と処理精度が異なる。データ１０個をデータ精度１００％でグループ化した複合データの場合、ＤＳ処理部３５の処理負荷（チェック回数）は１０で処理精度は１００％である。データ量１０個をデータ精度５０％でグループ化した複合データの場合、ＤＳ処理部３５の処理負荷（チェック回数）は１０で処理精度は５０％である。データ量１００個をデータ精度５０％でグループ化した複合データの場合、ＤＳ処理部３５の処理負荷（チェック回数）は１００で処理精度は５０％である。
【００３７】
制御装置２は、処理ネットワーク監視部４と、監視情報記憶部５と、処理ネットワーク分析部６と、ＤＳ処理要件記憶部７と、ＤＳ複合方法指示部８とを含む。
【００３８】
処理ネットワーク監視部４は、処理ネットワーク９に含まれる各ノード３の処理状態を監視する機能を備える。監視情報記憶部５は、具体的には、ハードディスク装置やメモリ等の記憶装置によって実現される。監視情報記憶部５は、処理ネットワーク監視部４による監視結果を監視情報として記憶する。
【００３９】
図２は、監視情報記憶部５が格納する各ノード３の監視結果のデータ構造例を示す。図２に示すように、監視情報記憶部５は、監視情報として、ノードＩＤ、監視項目、時刻および値を対応付けて記憶する。
【００４０】
ノードＩＤは、処理ネットワーク９内のノード３を識別するためのＩＤ（識別子）である。監視項目は、監視項目を識別するためのラベルである。時刻は、監視項目で示される値を監視し、その監視結果を取得した時刻である。値は、監視結果として取得した値である。
【００４１】
処理ネットワーク監視部４は、監視する項目として、各ノード３について、ＤＳキュー残サイズとＤＳ複合失敗率とＤＳ処理の処理精度を監視する。なお、これらの値に限られず、監視部は、他の項目で示される値を監視してもかまわない。
【００４２】
図２の監視項目の「ＤＳキュー残サイズ」とは、例えば、ＤＳキュー３１全体の容量に対する滞留している（未処理の）データストリームのデータ量の割合である。または、ＤＳキュー３１に滞留しているデータストリームのデータ量である。例えば、１ＧＢのメモリ領域のＤＳキュー３１が５００ＭＢの未処理のデータを格納している場合、ＤＳキュー残サイズは５０％または５００ＭＢである。ＤＳキュー残サイズは、ＤＳ処理部３５の処理待ち状態にあるデータ量であり、ノード３の処理負荷の混み具合を示す指標である。従って、ＤＳキュー残サイズの割合の値が大きい程、そのノード３が混み合っていることを示す。
【００４３】
図２の監視項目の「ＤＳ複合失敗率」とは、ＤＳ複合部３２が定期的にＤＳキュー３１を監視する回数に対する、ＤＳ複合方法を満たさずに複合データを出力できなかった回数の割合である。すなわち、ＤＳ複合失敗率とは、データ複合に失敗してＤＳ処理部３５に移行できない状態の割合であり、ノード３の処理負荷の空き具合を示す指標である。従って、ＤＳ複合失敗率の値が大きい程、そのノード３が空いていることを示す。
【００４４】
図２の監視項目の「ＤＳ処理の処理精度」は、ＤＳ処理部３５の処理対象とする複合データのデータ精度である。ＤＳ処理の処理精度は、値が大きい程、ノード３の処理精度が高いことを示す。
【００４５】
処理ネットワーク９と制御装置２とが１つの情報処理装置を用いて実現される場合、処理ネットワーク監視部４は、所定のタイミングで（例えば、所定時間毎に）、各ノード３のＤＳキュー３１中のデータ格納部分のサイズを求め、「ＤＳキュー残サイズ」を求めてもよい。また、処理ネットワーク監視部４は、ＤＳ複合部３２によるＤＳキュー３１の監視回数および複合データの出力失敗回数を求め、「ＤＳ複合失敗率」を求めてもよい。また、処理ネットワーク監視部４は、ＤＳ複合処理部の入力されたデータのデータ量と出力された複合データのデータ量から、「ＤＳ処理の処理精度」を求めてもよい。
【００４６】
また、処理ネットワーク９と制御装置２とが別々の情報処理装置を用いて実現される場合、処理ネットワーク監視部４は、「ＤＳキュー残サイズ」や「ＤＳ複合失敗率」、「ＤＳ処理の処理精度」を、ネットワークを介して各ノード３から受信してもよい。
【００４７】
図１の処理ネットワーク分析部６は、処理ネットワーク９内のどのノード３のＤＳ複合部３２のＤＳ複合方法を変更するかを分析する機能を備える。処理ネットワーク分析部６は、所定のタイミングで、監視情報記憶部５に集められた監視情報を取得する。そして、処理ネットワーク分析部６は、取得した監視結果と、ＤＳ処理要件記憶部７が記憶するＤＳ処理要件情報に基づいて、ＤＳ複合部３２のＤＳ複合方法を変更するノード３を分析する。また、処理ネットワーク分析部６は、ＤＳ複合部３２のＤＳ複合方法を変更すると分析したノード３について、どのＤＳ複合方法に変更するかを分析する。
【００４８】
ＤＳ処理要件記憶部７は、具体的には、ハードディスク装置やメモリ等の記憶装置を用いて、その中に形成されたデータ構造によって実現される。ＤＳ処理要件記憶部７は、ＤＳ複合方法を変更させるノード３を分析するために必要な情報であるＤＳ処理要件情報を記憶する。
【００４９】
図３、図４、図５は、ＤＳ処理要件記憶部７に格納されているＤＳ処理要件情報の構造の一例を示す説明図である。なお、ＤＳ処理要件情報は、例えば、予めシステム管理者によって作成され、ＤＳ処理要件記憶部７に格納されている。ＤＳ処理要件記憶部７は、ＤＳ処理要件情報として、選択ＤＳ複合方法一覧情報、選択可能ＤＳ複合方法一覧情報、およびＤＳ制約情報を記憶する。
【００５０】
図３は、選択ＤＳ複合方法一覧情報の一例を示す。選択ＤＳ複合方法一覧情報は、各ノード３のＤＳ複合部３２が現在選択しているＤＳ複合方法を示す情報である。図３に示すように、ＤＳ処理要件記憶部７は、選択ＤＳ複合方法一覧情報として、ノードＩＤと選択ＤＳ複合方法を対応付けて記憶する。ノードＩＤは、処理ネットワーク９内のノード３を識別するためのＩＤ（識別子）である。選択ＤＳ複合方法は、ノードＩＤで選別されるノード３のＤＳ複合部３２が現在選択しているＤＳ複合方法を識別するためのＩＤ（識別子）である。
【００５１】
図４は、選択可能ＤＳ複合方法一覧情報の一例を示す。選択可能ＤＳ複合方法一覧情報は、各ノード３のＤＳ複合方法記憶部３３が格納しているＤＳ複合方法を示す情報である。図４に示すように、ＤＳ処理要件記憶部７は、選択可能ＤＳ複合方法一覧情報として、ノードＩＤ、ＤＳ複合方法、レベル、データ量およびデータ精度を対応付けて記憶する。
【００５２】
ノードＩＤは、処理ネットワーク９内のノード３を識別するためのＩＤ（識別子）である。ＤＳ複合方法は、ノード３のＤＳ複合方法記憶部３３が格納しているＤＳ複合方法を識別するためのＩＤ（識別子）である。レベルは、ＤＳ複合方法を、データ量とデータ精度とを基準に順位付けした順序を示す情報である。すなわち、データ量とデータ精度との両方が大きい程、レベルの値が高くなる。データ量は、ＤＳ複合方法でグループ化されるデータの量を示す情報である。すなわち、データ量の値が大きい程、ＤＳ処理部３５が処理の実行に多くのシステムリソースを消費することを示す。データ精度は、ＤＳ複合方法でグループ化によって減少するデータの量の割合を示す情報である。すなわち、データ精度の値が大きい程、データを厳密に反映した複合データを出力するＤＳ複合部３２であり、この複合データを用いてＤＳ処理部３５は、より厳密な（精度の高い）統計処理が可能になることを示す。
【００５３】
図５は、ＤＳ制約情報の一例を示す。ＤＳ制約情報は、ノード３が満たさなくてはならない制約を示す情報、またはノード３同士が満たさなくてはならない制約を示す情報である。図５に示すように、ＤＳ処理要件記憶部７は、ＤＳ制約情報として、制約条件および対象ノードＩＤを対応付けて記憶する。図５の制約条件は、満たすべき制約条件の種類を示すラベルである。対象ノードＩＤは、制約条件の対象となるノード３を識別するためのＩＤ（識別子）である。制約条件には、以下のような種類の条件が考えられるが、これらに限定するものではない。
【００５４】
（１）同期：複数のノード３同士のスループットが同程度であるという制約を示す条件である。例えば、図１に示す例では、処理ネットワーク９において、２つのノード３が並列に配置されている。この並列に配置された２つのノード３の後段に接続されているノード３は、双方の処理結果を用いて処理を行うため、どちらか一方のノード３のスループットが高くても、他方のノード３のスループットが低ければ、処理待ちが発生する。このようなノード配置は、ノード３の混雑を発生させて、処理ネットワーク９全体のスループットを停滞させる可能性が高い。従って、このような２つのノード３が並列に配置されるノード配置の場合、これらのノード３のスループットが同程度であることが、制約条件として求められる。なお、同程度であるとは、例えば、２つのノード３のデータ量の差が、１０パーセント程度（好ましくは５パーセント程度）であることをいう。
【００５５】
（２）優先：対象ノード３のＤＳ複合方法のレベルが、他のノード３に比べて相対的に優先して高いという制約を示す条件である。例えば、図１に示す例では、処理ネットワーク９において、処理ネットワーク９の先頭に１つのノード３が配置されている。この処理ネットワーク９の先頭に配置されるノード３のデータ精度が極端に低いと、このノード３の後段のノード３のデータ精度も低くなり、処理ネットワーク９全体のデータ精度も劣化する。従って、このようなこの処理ネットワーク９の先頭に配置されるノード３は、他の後段のノード３よりも優先して高いレベル（高いデータ精度）であることが、制約条件として求められる。
【００５６】
（３）精度：対象ノード３のＤＳ複合部３２のデータ精度が一定以上でなければならないという制約を示す条件である。例えば、制約条件「優先」と同様の理由から、またはＤＳＭＳ自体に求められるサービスレベル（データ精度もしくは処理精度）を維持するために、一定以上のデータ精度が求められるという制約条件を示している。
【００５７】
（４）負荷：対象ノード３のＤＳ複合部３２のデータ量が、一定以下でなければならないという制約を示す条件である。例えば、ノード３のシステムリソースの制限、またはＤＳＭＳ自体に求められるサービスレベル（スループット）を維持するため、一定以下の処理負荷が求められるという制約条件を示している。
【００５８】
なお、制約条件として設定される精度制約や負荷制約は、ノード３単体に対する制約条件ではなく、複数のノード３からなるデータフロー経路に対して課される制約条件であってもかまわない。例えば、あるデータフロー経路上にあたるノード全てのデータ精度が一定以上でなければならないことを示す精度制約等を、予めＤＳ処理要件記憶部７に設定していてもよい。
【００５９】
処理ネットワーク分析部６は、ＤＳ複合方法の変更を分析方式として、様々な方式を用いることができるが、以下、処理ネットワーク分析部６による動作の一例を説明する。
【００６０】
処理ネットワーク分析部６は、１つの分析対象のノード３を選択する。この場合、処理ネットワーク分析部６は、例えば、選択方法として、単純ラウンドロビン（全てのノード３を順番に選択する）を用いて、ノード３を選択するのが最も容易である。なお、処理ネットワーク分析部６は、単純ラウンドロビンに限らず、重み付きラウンドロビンやランダム選択等の他の選択方法を用いて、ノード３を選択してもかまわない。処理ネットワーク分析部６は、選択したノード３に関する過去一定期間の監視情報を監視情報記憶部５から取得する。
【００６１】
処理ネットワーク分析部６は、取得した監視情報から、選択したノード３のＤＳキュー３１の過去一定期間の平均ＤＳキュー残サイズを計算する。そして、処理ネットワーク分析部６は、求めた平均ＤＳキュー残サイズが所定値以上か否かを判定する。
【００６２】
過去一定期間の平均ＤＳキュー残サイズが所定値以上であれば、処理ネットワーク分析部６は、選択したノード３が混雑している状態であると分析し、選択したノード３のＤＳ複合部３２のＤＳ複合方法をより低いレベルのＤＳ複合方法に変更指示すると分析する。この場合、処理ネットワーク分析部６は、選択したノード３が現在選択しているＤＳ複合方法の情報を、ＤＳ処理要件記憶部７が記憶する選択ＤＳ複合方法一覧情報から取得する。また、処理ネットワーク分析部６は、選択可能なＤＳ複合方法の一覧情報を、ＤＳ処理要件記憶部７が記憶する選択可能ＤＳ複合方法一覧情報から取得する。そして、処理ネットワーク分析部６は、取得したこれらの情報に基づいて、現在選択しているＤＳ複合方法よりレベルの低いＤＳ複合方法を選択したノード３が備えているか否かを判定する。
【００６３】
選択したノード３がよりレベルの低いＤＳ複合方法を備えていると判定すれば、処理ネットワーク分析部６は、次に、この低いレベルのＤＳ複合方法を仮に選択した場合に、制約条件を満たすか否かを判定する。この場合、処理ネットワーク分析部６は、選択したノード３が満たすべき制約を示す情報を、ＤＳ処理要件記憶部７が記憶する制約情報から取得する。そして、処理ネットワーク分析部６は、既に取得している低いレベルのＤＳ複合方法の情報等と照らし合わせて、取得した情報に示される制約条件を満たすか否かを判定する。
【００６４】
低いレベルのＤＳ複合方法を選択した場合に制約条件を満たすか否かを判定した結果、選択したノード３が満たすべき制約情報を満たしていると判定すれば、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を低いレベルのＤＳ複合方法に変更する指示すると判定し、判定結果をＤＳ複合方法指示部８に転送する。
【００６５】
一方、求めた過去一定期間の平均ＤＳキュー残サイズが所定値より小さければ、処理ネットワーク分析部６は、選択したノード３が混雑している状態ではないと分析する。
【００６６】
この場合、処理ネットワーク分析部６は、次に、監視情報記憶部５から取得した監視情報から、選択したノード３のＤＳ複合部３２の過去一定期間の平均ＤＳ複合失敗率を計算する。そして、処理ネットワーク分析部６は、求めた平均ＤＳ複合失敗率が所定値以上か否かを判定する。
【００６７】
過去一定期間の平均ＤＳ複合失敗率が所定値以上であれば、処理ネットワーク分析部６は、選択したノード３が空いている状態であると分析し、選択したノード３のＤＳ複合方法のレベルを上げるように切り替え指示すると判定する。処理ネットワーク分析部６は、選択したノード３が現在選択しているＤＳ複合方法の情報を、ＤＳ処理要件記憶部７が記憶する選択ＤＳ複合方法一覧情報から取得する。また、処理ネットワーク分析部６は、選択可能なＤＳ複合方法の一覧情報を、ＤＳ処理要件記憶部７が記憶する選択可能ＤＳ複合方法一覧情報から取得する。そして、処理ネットワーク分析部６は、取得したこれらの情報に基づいて、現在選択しているＤＳ複合方法よりレベルの高いＤＳ複合方法を選択したノード３が備えているか否かを判定する。
【００６８】
選択したノード３がよりレベルの高いＤＳ複合方法を備えていると判定すれば、処理ネットワーク分析部６は、次に、この高いレベルのＤＳ複合方法を仮に選択した場合に、制約条件を満たすか否かを判定する。この場合、処理ネットワーク分析部６は、選択したノード３が満たすべき制約条件を示す情報を、ＤＳ処理要件記憶部７が記憶する制約情報から取得する。そして、処理ネットワーク分析部６は、既に取得している高いレベルのＤＳ複合方法の情報等と照らし合わせて、取得した情報に示される制約条件を満たすか否かを判定する。
【００６９】
高いレベルのＤＳ複合方法を選択した場合に制約条件を満たすか否かを判定した結果、選択したノード３が満たすべき制約条件を満たしていると判定すれば、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を高いレベルのＤＳ複合方法に変更する指示すると判定し、判定結果をＤＳ複合方法指示部８に転送する。
【００７０】
一方、求めた過去一定期間の平均ＤＳ複合失敗率が所定値より小さければ、処理ネットワーク分析部６は、選択したノード３のＤＳスループットが混雑している状態ではないと分析する。この場合、処理ネットワーク分析部６は、次に、監視情報記憶部５から取得した監視情報から、選択したノード３のＤＳ処理部３５の過去一定期間の平均処理精度を計算する。そして、処理ネットワーク分析部６は、求めた平均処理精度が所定値以上か否かを判定する。
【００７１】
過去一定期間の平均処理精度が所定値以下であれば、処理ネットワーク分析部６は、選択したノード３のＤＳ処理精度が劣化している状態であると分析し、選択したノード３のＤＳ複合方法のレベルを上げるように切り替え指示すると判定する。処理ネットワーク分析部６は、選択したノード３が現在選択しているＤＳ複合方法の情報を、ＤＳ処理要件記憶部７が記憶する選択ＤＳ複合方法一覧情報から取得する。また、処理ネットワーク分析部６は、選択可能なＤＳ複合方法の一覧情報を、ＤＳ処理要件記憶部７が記憶する選択可能ＤＳ複合方法一覧情報から取得する。そして、処理ネットワーク分析部６は、取得したこれらの情報に基づいて、現在選択しているＤＳ複合方法よりレベルの高いＤＳ複合方法を選択したノード３が備えているか否かを判定する。
【００７２】
選択したノード３がよりレベルの高いＤＳ複合方法を備えていると判定すれば、処理ネットワーク分析部６は、次に、この高いレベルのＤＳ複合方法を仮に選択した場合に、制約条件を満たすか否かを判定する。この場合、処理ネットワーク分析部６は、選択したノード３が満たすべき制約条件を示す情報を、ＤＳ処理要件記憶部７が記憶する制約情報から取得する。そして、処理ネットワーク分析部６は、既に取得している高いレベルのＤＳ複合方法の情報等と照らし合わせて、取得した情報に示される制約条件を満たすか否かを判定する。
【００７３】
高いレベルのＤＳ複合方法を選択した場合に制約条件を満たすか否かを判定した結果、選択したノード３が満たすべき制約条件を満たしていると判定すれば、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を高いレベルのＤＳ複合方法に変更する指示すると判定し、判定結果をＤＳ複合方法指示部８に転送する。
【００７４】
一方、求めた過去一定期間の平均処理精度が所定値よりも大きければ、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。また、ＤＳ複合方法を選択した候補のレベルへと変更できない（選択したレベルのＤＳ複合方法を選択したノード３が備えていない）場合も、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。また、制約条件を満たさない場合も、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。
【００７５】
処理ネットワーク分析部６は、上述した処理を、すべてのノード３に対して繰り返し実行する。
【００７６】
図１のＤＳ複合方法指示部８は、処理ネットワーク分析部６の分析結果に基づいて、ＤＳ複合部３２のＤＳ複合方法の変更をノード３に指示する機能を備える。各ノード３のＤＳ複合変更部３４は、制御装置２のＤＳ複合方法指示部８から受信したコマンドに応じて、ＤＳ複合部３２のデータ複合方法を変更する。なお、この場合、制御装置とノード３とが別々の情報処理装置によって実現される場合には、ＤＳ複合変更部３４は、制御装置からネットワークを介して受信したコマンドに示される変更指示に従って、ＤＳ複合方法を変更する。
【００７７】
ノード３は、データの入力がある毎に、外部又は処理ネットワーク９内の他のノード３からデータを受信し、その受信したデータをキュー形式でＤＳキュー３１に格納する。ＤＳ複合部３２は、ＤＳキュー３１を監視し、ＤＳ複合変更部３４から指定されたデータ複合方法に基づいて、ＤＳキュー３１から所定量のデータを取り出して１つのデータに複合してＤＳ処理部３５に転送する。
【００７８】
ＤＳ処理部３５は、ＤＳ複合部３２から転送された複合データを統計処理する。そして、ＤＳ処理部３５は、統計処理した複合データを、外部又は処理ネットワーク９内の次のノード３のＤＳキュー３１に出力する。
【００７９】
図６は、本実施の形態に係る制御装置２の動作の一例を示すフローチャートである。制御装置２の処理ネットワーク監視部４は、各ノード３を所定のタイミングで繰り返し監視し、各ノード３の処理状態を監視情報として監視情報記憶部５に格納する（ステップＳ１１）。
【００８０】
処理ネットワーク監視部４は、ステップＳ１１の処理を、所定のタイミングで繰り返し実行する。これによって、監視情報記憶部５に、各ノード３の処理状態を監視した監視情報が蓄積される。
【００８１】
処理ネットワーク分析部６は、所定のタイミングで、監視情報記憶部５に集められた監視情報を取得する。そして、処理ネットワーク分析部６は、取得した監視結果と、ＤＳ処理要件記憶部７が記憶するＤＳ処理要件情報に基づいて、ＤＳ複合部３２のＤＳ複合方法を変更するノード３を分析する。また、処理ネットワーク分析部６は、ＤＳ複合部３２のＤＳ複合方法を変更すると分析したノード３について、どのＤＳ複合方法に変更するかを分析する（ステップＳ１２）。
【００８２】
ステップＳ１２において、処理ネットワーク分析部６は、各ノード３の処理状態を分析して、ＤＳ複合方法を変更するかどうかの判定を行う。処理ネットワーク分析部６によるＤＳ複合方法の変更可否定の分析処理の一例を以下に示す。
【００８３】
図７は、本実施の形態に係るＤＳ複合方法の変更判定処理の一例を示すフローチャートである。図７のフローチャートは、図６のステップＳ１２の詳細な内容を示す。処理ネットワーク分析部６は、番号変数ｉに１を代入して（ステップＳ２１）、分析対象として１番目のノード３を選択する。この場合、処理ネットワーク分析部６は、例えば、選択方法として、単純ラウンドロビン（全てのノード３を順番に選択する）を用いて、ノード３を選択するのが最も容易である。なお、処理ネットワーク分析部６は、単純ラウンドロビンに限らず、重み付きラウンドロビンやランダム選択等の他の選択方法を用いて、ノード３を選択してもかまわない。
【００８４】
処理ネットワーク分析部６は、選択したノード３に関する過去一定期間の監視情報を監視情報記憶部５から取得する（ステップＳ２２）。処理ネットワーク分析部６は、取得した監視情報から、選択したノード３のＤＳキュー３１の過去一定期間の平均ＤＳキュー残サイズＲを計算する（ステップＳ２３）。
【００８５】
処理ネットワーク分析部６は、求めた平均ＤＳキュー残サイズＲが所定値α以上か否かを判定する（ステップＳ２４）。平均ＤＳキュー残サイズＲが所定値α以上であれば（ステップＳ２４；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３が混雑していると判定し、現在選択しているＤＳ複合方法よりレベルの低いＤＳ複合方法を選択したノード３が備えているか否かを判定する（ステップＳ２５）。
【００８６】
選択したノード３がよりレベルの低いＤＳ複合方法を備えている場合（ステップＳ２５；ＹＥＳ）、処理ネットワーク分析部６は、次に、この低いレベルのＤＳ複合方法を仮に選択した場合に、制約条件を満たすか否かを判定する（ステップＳ２６）。選択した低いレベルのＤＳ複合方法が、選択したノード３が満たすべき制約条件を満たしている場合（ステップＳ２６；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を低いレベルのＤＳ複合方法に変更するように設定し、ＤＳ複合方法指示部８に転送する（ステップＳ２７）。選択したノード３がレベルの低い複合方法を備えていない場合（ステップＳ２５；ＮＯ）、またはレベルの低い複合方法が制約条件を満たさない場合は（ステップＳ２６；ＮＯ）、選択したノード３のＤＳ複合方法を変更しない。
【００８７】
一方、求めた過去一定期間の平均ＤＳキュー残サイズが所定値αより小さければ（ステップＳ２４；ＮＯ）、処理ネットワーク分析部６は、監視情報から、選択したノード３のＤＳ複合部３２の過去一定期間の平均ＤＳ複合失敗率Ｆを計算する（ステップＳ２８）。そして、処理ネットワーク分析部６は、求めた平均ＤＳ複合失敗率Ｆが所定値β以上か否かを判定する（ステップＳ２９）。
【００８８】
過去一定期間の平均ＤＳ複合失敗率Ｆが所定値β以上であれば（ステップＳ２９；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３が空いている状態であると判定し、現在選択しているＤＳ複合方法よりレベルの高いＤＳ複合方法を選択したノード３が備えているか否かを判定する（ステップＳ３０）。選択したノード３がよりレベルの高いＤＳ複合方法を備えている場合（ステップＳ３０；ＹＥＳ）、処理ネットワーク分析部６は、次に、この高いレベルのＤＳ複合方法を仮に選択した場合に、制約条件を満たすか否かを判定する（ステップＳ３１）。
【００８９】
高いレベルのＤＳ複合方法が、選択したノード３が満たすべき制約条件を満たしている場合（ステップＳ３１；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を高いレベルのＤＳ複合方法に変更すると設定し、ＤＳ複合方法指示部８に転送する（ステップＳ３２）。選択したノード３がレベルの高い複合方法を備えていない場合（ステップＳ３０；ＮＯ）、またはレベルの高い複合方法が制約条件を満たさない場合は（ステップＳ３１；ＮＯ）、選択したノード３のＤＳ複合方法を変更しない。
【００９０】
一方、求めた過去一定期間の平均ＤＳ複合失敗率Ｆが所定値βより小さければ（ステップＳ２９；ＮＯ）、処理ネットワーク分析部６は、次に、監視情報記憶部５から取得した監視情報から、選択したノード３のＤＳ処理部３５の過去一定期間の平均処理精度Ｐを計算する（ステップＳ３３）。そして、処理ネットワーク分析部６は、求めた平均処理精度Ｐが所定値γ以上か否かを判定する（ステップＳ３４）。
【００９１】
過去一定期間の平均処理精度が所定値γ以下であれば（ステップＳ３４；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３のＤＳ処理精度が劣化している状態であると判定し、現在選択しているＤＳ複合方法よりレベルの高いＤＳ複合方法を選択したノード３が備えているか否かを判定する（ステップＳ３５）。
【００９２】
選択したノード３がよりレベルの高いＤＳ複合方法を備えている場合（ステップＳ３５；ＹＥＳ）、処理ネットワーク分析部６は、次に、この高いレベルのＤＳ複合方法が制約条件を満たすか否かを判定する（ステップＳ３６）。高いレベルのＤＳ複合方法が、制約情報を満たしている場合（ステップＳ３６；ＹＥＳ）、処理ネットワーク分析部６は、選択したノード３のＤＳ複合部３２のＤＳ複合方法を高いレベルのＤＳ複合方法に変更すると設定し、判定結果をＤＳ複合方法指示部８に転送する（ステップＳ３７）。
【００９３】
一方、求めた過去一定期間の平均処理精度Ｐが所定値γよりも大きければ（ステップＳ３４；ＮＯ）、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。また、ＤＳ複合方法を意図するレベルへと変更できない（意図するレベルのＤＳ複合方法を選択したノード３が備えていない）場合も（ステップＳ３５；ＮＯ）、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。また、制約条件を満たさない場合も（ステップＳ３６；ＮＯ）、処理ネットワーク分析部６は、ＤＳ複合方法のレベルの変更指示を行わない。
【００９４】
処理ネットワーク分析部６は、番号変数ｉに１を加算して（ステップＳ３８）、番号変数ｉが処理ネットワーク９のノード３の数ｎを超えていなければ（ステップＳ３９；ＮＯ）、ステップＳ２２に戻って、上記に示した処理をi番のノード３に対して実行する。番号変数ｉがノード３の数ｎを超えていれば（ステップＳ３９；ＹＥＳ）、図６のフローチャートに戻る。
【００９５】
図６のフローチャートに戻って、ＤＳ複合方法指示部８は、処理ネットワーク分析部６の分析結果に従い、ＤＳ複合部３２のＤＳ複合方法の変更を指示すると判定したノード３に対して、ＤＳ複合方法の変更を指示する（ステップＳ１３）。なお、制御装置２とノード３とが別々の情報処理装置によって実現される場合には、ＤＳ複合方法指示部８は、ＤＳ複合方法の変更を指示するコマンドを、ネットワークを介してノード３に送信する。
【００９６】
また、ＤＳ複合方法指示部８は、同時に、ＤＳ処理要件記憶部７が記憶する選択ＤＳ複合方法一覧情報のうち、変更対象となるノード３のノードＩＤに対する選択ＤＳ複合方法を、変更したＤＳ複合方法のＩＤに書き換える。
【００９７】
図８は、本実施の形態に係るノード３のデータ処理の動作の一例を示すフローチャートである。ノード３は、データの入力がある毎に、外部または処理ネットワーク９内の他のノード３からデータを受信し、その受信したデータをキュー形式でＤＳキュー３１に格納する（ステップＳ４１）。図８の例では、所定の時間を経過するまで、ステップＳ４１を繰り返す（ステップＳ４２；ＮＯ）。
【００９８】
所定の時間を経過したら（ステップＳ４２；ＹＥＳ）、制御装置２からＤＳ複合部３２のＤＳ複合方法の変更を指示するコマンドを受けているかどうか調べる（ステップＳ４３）。変更指示を受けた場合には（ステップＳ４３；ＹＥＳ）、制御装置２からの変更指示に従って、ＤＳ複合方法を変更する（ステップＳ４４）。複合方法の変更指示がなければ（ステップＳ４３；ＮＯ）、ＤＳ複合方法を変更しない。
【００９９】
ＤＳ複合部３２は、ＤＳキュー３１を監視し、ＤＳ複合変更部３４から指定されたデータ複合方法に基づいて、ＤＳキュー３１から所定量のデータを取り出して１つのデータに複合してＤＳ処理部３５に転送する（ステップＳ４５）。
【０１００】
ＤＳ処理部３５は、ＤＳ複合部３２から転送された複合データを統計処理する（ステップＳ４６）。そして、ＤＳ処理部３５は、統計処理した複合データを、外部または処理ネットワーク９内の次のノード３のＤＳキュー３１に出力する（ステップＳ４７）。
【０１０１】
以上の処理によって、各ノード３の処理精度と処理負荷とを制御し、その制御の結果、各ノード３の消費システムリソース量を明示的に変更することなく、スループットを制御することができる。
【０１０２】
本実施の形態のデータストリーム複合制御システム１では、処理ネットワーク９に含まれる各ノード３は、逐次に継続して入力される大量の情報であるデータを複合するデータ複合方法を複数備えるＤＳ複合部３２を含む。そして、各ノード３の処理状態に応じて、ＤＳ複合部３２に用いるＤＳ複合方法を変更する処理を実行する。これにより、各ノード３の処理精度と処理負荷とを制御することができる。また、各ノード３の処理精度と処理負荷とを制御することで、ノード３の消費システムリソース量を明示的に変更することなく、スループットを制御することができる。また、各ノード３の監視結果に応じて、各ノード３の処理精度と処理負荷とを繰り返し制御することによって、処理ネットワーク９全体のスループットを向上させることができる。
【０１０３】
また、本実施の形態によれば、上記のような処理を実行することによって、スループットの低いノード３の処理精度を下げる代わりに処理負荷を下げることによって、スループットを向上させることができる。
【０１０４】
例えば、非特許文献１や非特許文献２に記載されたＤＳＭＳを用いた場合、スループットの高いノード３の後段にスループットの低いノード３が配置されると、スループットの低いノード３のＤＳキュー３１内のデータの消費速度より、前段のスループットの高いノード３によるデータの出力速度が上回る状態が発生してしまう。この状態が継続すると、ＤＳキュー３１内のデータは増え続け、やがてオーバーフローを起こし、ＤＳＭＳは異常停止してしまうという問題があった。
【０１０５】
ＤＳＭＳ全体のスループット律束や、ＤＳキュー３１のオーバーフローが、一部のノード３のスループット格差に基づくものであれば、スループットの高いノード３に割り当てているシステムリソース（メモリやＣＰＵパワー等）を少なくし、スループットの低いノード３に多くのシステムリソースを割り当てることにより、スループット格差を是正し、防ぐことが可能な場合もある。しかし、割り当てるシステムリソースを適切に配分するのは困難であるという問題がある。
【０１０６】
これに対して、本実施の形態によれば、スループットの高いノード３のスループットを下げることで、ＤＳＭＳにおける各ノード３のスループット格差を是正し、ＤＳＭＳ全体のスループットを向上させることができる。また、各ノード３のＤＳキュー３１のオーバーフローを防ぎ、ＤＳＭＳの異常停止を防ぐことができる。
【０１０７】
また、例えば、非特許文献１や非特許文献２に記載されたＤＳＭＳを用いた場合、外部からの入力データが時々刻々と変化するため、時には非常に大量のデータが入力される場合もある。この場合、入力データのデータ量がＤＳＭＳの全システムリソースを用いても処理しきれない量の場合には、処理ネットワーク９のどこかで上記したようなＤＳキュー３１のオーバーフローが発生してしまうおそれがある。さらに、システムリソースに余裕がないため、そのオーバーフローを解消できないおそれがある。そのため、ＤＳＭＳが異常停止してしまうという問題があった。
【０１０８】
これに対して、本実施の形態によれば、システム全体のノード３の処理精度を下げる代わりに処理負荷を下げて、システム全体の消費システムリソースを変更することなく、システム全体のスループットを向上させる。そのため、大量のデータが入力されても、ＤＳＭＳの異常停止を防ぐことができる。
【０１０９】
以上のように、本実施の形態のデータストリーム複合制御システム１によれば、処理ネットワーク９における各ノード３の処理状態に応じて、混雑しているノード３のＤＳ複合方法を変更することで、処理精度と処理負荷とを制御することを基本として、各ノード３のスループット格差を是正し、ＤＳＭＳ全体のスループットを向上することができる。
【０１１０】
図９は、本発明の実施の形態に係る制御装置２のハードウェア構成の一例を示すブロック図である。制御装置２は、図９に示すように、制御部１１、主記憶部１２、外部記憶部１３、操作部１４、表示部１５および送受信部１６を備える。主記憶部１２、外部記憶部１３、操作部１４、表示部１５および送受信部１６はいずれも内部バス１０を介して制御部１１に接続されている。
【０１１１】
制御部１１はＣＰＵ（Central Processing Unit）等から構成され、外部記憶部１３に記憶されている制御プログラム１７に従って、制御装置２の処理を実行する。
【０１１２】
主記憶部１２はＲＡＭ（Random-Access Memory）等から構成され、外部記憶部１３に記憶されている制御プログラム１７をロードし、制御部１１の作業領域として用いられる。
【０１１３】
外部記憶部１３は、フラッシュメモリ、ハードディスク、ＤＶＤ−ＲＡＭ（Digital Versatile Disc Random-Access Memory）、ＤＶＤ−ＲＷ（Digital Versatile Disc ReWritable）等の不揮発性メモリから構成され、上述の処理を制御部１１に行わせるための制御プログラム１７を予め記憶し、また、制御部１１の指示に従って、制御プログラム１７が記憶するデータを制御部１１に供給し、制御部１１から供給されたデータを記憶する。図１の監視情報記憶部５およびＤＳ処理要件記憶部７は、外部記憶部１３に構成される。
【０１１４】
操作部１４はキーボードおよびマウスなどのポインティングデバイス等と、キーボードおよびポインティングデバイス等を内部バス１０に接続するインタフェース装置から構成されている。操作部１４を介して、処理ネットワーク９の構成、各ノード３の監視項目と監視時間間隔、各ノード３のＤＳ複合方法を変更するための判定条件、または、各ノード３もしくは処理ネットワーク９の制約条件などが入力され、制御部１１に供給される。
【０１１５】
表示部１５は、ＣＲＴ（Cathode Ray Tube）またはＬＣＤ（Liquid Crystal Display）などから構成され、処理ネットワーク９の構成、各ノード３の監視項目と監視時間間隔、各ノード３のＤＳ複合方法を変更するための判定条件、または、各ノード３もしくは処理ネットワーク９の制約条件、あるいは各ノード３の監視情報などを表示する。
【０１１６】
送受信部１６は、通信装置、およびそれらと接続するシリアルインタフェースまたはＬＡＮ（Local Area Network）インタフェースから構成されている。送受信部１６は、ノード３から監視情報を受信し、ノード３へＤＳ複合方法の変更指示を送信する。
【０１１７】
図１の処理ネットワーク監視部４、処理ネットワーク分析部６およびＤＳ複合方法指示部８の処理は、制御プログラム１７が、制御部１１、主記憶部１２、外部記憶部１３、操作部１４、表示部１５、および送受信部１６などを資源として用いて処理することによって実行する。また、制御装置２は、ノード３の一部または全部を含んでコンピュータで構成することができる。
【０１１８】
本発明の好適な変形として、以下の構成が含まれる。
【０１１９】
本発明の第１の観点に係るデータストリーム複合制御システムについて、
好ましくは、前記ノード状態検知手段は、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定手段は、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする。
【０１２０】
好ましくは、前記設定手段は、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合手段の精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合手段のデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合手段のデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする。
【０１２１】
本発明の第２の観点に係る制御装置について、
好ましくは、前記ノード状態検知手段は、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定手段は、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする。
【０１２２】
好ましくは、前記設定手段は、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合手段の精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合手段のデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合手段のデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする。
【０１２３】
本発明の第３の観点に係るデータストリーム複合制御方法について、
好ましくは、前記ノード状態検知ステップは、前記処理ノードのデータストリームキューに滞留しているデータ量である残サイズ、および／または、前記データストリームキューから連続する所定の数の受信した単位のデータのうち複合する所定の数のデータを選択できなかった率を表すＤＳ複合失敗率、を検出し、
前記設定ステップは、前記残サイズおよび／または前記ＤＳ複合失敗率が所定の値になるように、前記連続する所定の数と前記複合する所定の数、および／または、前記複合した単位データをまとめて処理する精度を、前記データ複合方法として設定する、
ことを特徴とする。
【０１２４】
好ましくは、前記設定ステップは、前記複数の処理ノード同士のスループットの差が所定の範囲内であることを表す同期条件、前記複数の処理ノードのうち着目する対象の処理ノードのデータ複合ステップの精度が、他の処理ノードに比べて相対的に高いことを表す優先条件、前記処理ノードのデータ複合ステップのデータ精度が一定以上であることを表す精度条件、または、対象の処理ノードのデータ複合ステップのデータ量が一定以下であることを表す負荷条件、を含む前記処理ノードごとに定められた条件の範囲で、前記処理ノードの複合方法を設定する、ことを特徴とする。
【０１２５】
その他、前記のハードウェア構成やフローチャートは一例であり、任意に変更および修正が可能である。
【０１２６】
制御部１１、主記憶部１２、外部記憶部１３、送受信部１６、内部バス１０などから構成される制御装置のための処理を行う中心となる部分は、専用のシステムによらず、通常のコンピュータシステムを用いて実現可能である。たとえば、前記の動作を実行するためのコンピュータプログラムを、コンピュータが読み取り可能な記録媒体（フレキシブルディスク、ＣＤ−ＲＯＭ、ＤＶＤ−ＲＯＭ等）に格納して配布し、当該コンピュータプログラムをコンピュータにインストールすることにより、前記の処理を実行する制御装置を構成してもよい。また、インターネット等の通信ネットワーク上のサーバ装置が有する記憶装置に当該コンピュータプログラムを格納しておき、通常のコンピュータシステムがダウンロード等することで制御装置を構成してもよい。
【０１２７】
また、制御装置の機能を、ＯＳ（オペレーティングシステム）とアプリケーションプログラムの分担、またはＯＳとアプリケーションプログラムとの協働により実現する場合などには、アプリケーションプログラム部分のみを記録媒体や記憶装置に格納してもよい。
【０１２８】
また、搬送波にコンピュータプログラムを重畳し、通信ネットワークを介して配信することも可能である。たとえば、通信ネットワーク上の掲示板(BBS, Bulletin Board System)に前記コンピュータプログラムを掲示し、ネットワークを介して前記コンピュータプログラムを配信してもよい。そして、このコンピュータプログラムを起動し、ＯＳの制御下で、他のアプリケーションプログラムと同様に実行することにより、前記の処理を実行できるように構成してもよい。
【符号の説明】
【０１２９】
１  データストリーム複合制御システム
２  制御装置
３  ノード
４  処理ネットワーク監視部
５  監視情報記憶部
６  処理ネットワーク分析部
７  ＤＳ処理要件記憶部
８  ＤＳ複合方法指示部
９  処理ネットワーク
１１  制御部
１２  主記憶部
１３  外部記憶部
１４  操作部
１５  表示部
１６  送受信部
１７  制御プログラム
３１  ＤＳキュー
３２  ＤＳ複合部
３３  ＤＳ複合方法記憶部
３４  ＤＳ複合変更部
３５  ＤＳ処理部
【図１】
2011082807.tif 000003
【図２】
2011082807.tif 000004
【図３】
2011082807.tif 000005
【図４】
2011082807.tif 000006
【図５】
2011082807.tif 000007
【図６】
2011082807.tif 000008
【図７】
2011082807.tif 000009
【図８】
2011082807.tif 000010
【図９】
2011082807.tif 000011
【図１０】
2011082807.tif 000012