JP5245711B2 - 分散データ処理システム、分散データ処理方法および分散データ処理用プログラム - Google Patents

Description

本発明は、分散データ処理システム、データ処理装置、分散データ処理方法、および分散データ処理用プログラムに関し、特に、処理時間が速く、ネットワークの物理的配線やアドレス体系に依存しない分散データ処理システム、データ処理装置、分散データ処理方法、および分散データ処理用プログラムに関する。

分散システムにおける分散データ処理システムの一例が、非特許文献１〜３に記載されている。これらの分散データ処理システムでは、複数のデータセンタが存在し、個々のデータセンタは複数のラックを含み、各ラックはサーバを含んでいる。各々のサーバは、ルータまたはスイッチといったネットワーク機器により接続されている。また、この物理的な構成に応じて、サーバの割り当てられるＩＰアドレスが決定される。すなわち、＊を任意のアドレスとして、あるデータセンタのアドレスが１０．１.＊.＊というアドレス体系を持つとすると、そのデータセンタ内のラックは例えば１０．１.１.＊といったアドレス体系を持ち、さらにそのラック内のサーバは１０.１．１.１といったアドレスとなる。すなわち、同じデータセンタに属する各ラックのアドレスの上位部分は共通であり、同じラックに属する各サーバのアドレスの上位部分は共通である。

これらのサーバ上で動作するプログラムは複数種類存在する。そして、チャンクサーバ、ファイルシステムマスタ、クライアント、ワーカ、マスタ、ワークキューとして動作するサーバが存在する。

チャンクサーバは、分散ファイルシステムを構成するサーバであり、ファイルデータを格納する役割を担う。ファイルシステムマスタは、分散ファイルシステムにおけるファイル名と、そのファイルを格納するチャンクサーバを対応づける役割を担うサーバである。クライアントは、この分散データ処理システムのユーザプログラムが動作するサーバであり、分散データ処理システムに対して指定した処理を依頼する。ワーカは、依頼された処理を実行するサーバであり、依頼された処理に必要なデータをチャンクサーバから取得し実行する。マスタは、ワーカ間の処理の流れや順序を管理し、適当なワーカから適当なワーカへのデータの送受信などを指示するサーバである。ワークキューは、ワーカの処理状況を監視し、どのサーバが新たな計算を実行可能かを管理し、クライアントからの要求に応じて、必要なワーカおよびマスタを確保し、クライアントの指定したプログラムをワーカおよびマスタに配置するサーバである。

クライアントによって指示され、ワーカによって実行される処理には、Ｍａｐ、Ｓｈｕｆｆｌｅ、Ｒｅｄｕｃｅの３つの段階が存在する。Ｍａｐは、異なるデータに対して同一の処理を並列に実行する段階である。入力データは、複数のキー・値ペアである。これを第１のキー・値ペアとする。一つの第１のキー・値ペアに対してＭａｐで指定されたプログラムが実行され、それぞれは複数のキーと値ペアを出力する。これを第２のキー・値ペアとする。

Ｓｈｕｆｆｌｅ段階では、Ｍａｐ段階の出力である複数の第２のキー・値ペアを入力として受け取る。そして、出力された複数の第２のキー・値ペアのうち、キーについて順序付けし、同一のキーを持つものをグループ化して、キーと値リストのペアを構成する。Ｒｅｄｕｃｅ段階では、この複数のキーと値リストのそれぞれについて、Ｒｅｄｕｃｅで指定されたプログラムを適用する。このＲｅｄｕｃｅプログラムは、一つのキーと値リストの入力から、一つのキー・値ペアを出力する。これを第３のキー・値ペアとする。

第１のキー・値ペアは、主にチャンクサーバに格納され、第３のキー・値ペアも同様にチャンクサーバに格納される。各実行段階のうち、Ｍａｐ段階の処理の実行場所は、入力データである第１のキー・値ペアが格納されるチャンクサーバの近くで実行される。ここでの「近く」とは、アドレスによって判断される。即ち、指定された第１のキー・値ペアを格納するチャンクサーバがワーカとしても動作し、他の処理を行っていない等の理由で、ワーカとして利用可能な状態の時には、Ｍａｐ段階の処理がチャンクサーバと同一のサーバで実行される。これにより入力データのネットワーク転送を省くことができる。一方、そのチャンクサーバが他の計算を行っている等の理由で利用不可能である場合には、そのデータを格納するチャンクサーバと同一のラックに存在する利用可能なサーバに処理を割り当てることで、ラックを超えた通信を行わないようにする。この時、サーバやラックの物理構成に応じて、静的にアドレス体系を割り当てているため、サーバ同士のラックが同一のラックであるか否かを判断できる。

また、他の分散データ処理システムの一例が、非特許文献４に記載されている。この分散データ処理システムにおいても、データセンタは複数のラックを含み、各ラックは複数のサーバを含んでいる。また、サーバには、ワーカに相当するデーモンが搭載されたサーバや、ファイルシステムマスタに相当するネームサーバや、マスタおよびワークキューに相当するジョブマネージャが搭載されたサーバ等がある。

この分散データ処理システムでは、非特許文献１〜３のＳｈｕｆｆｌｅ段階およびＲｅｄｕｃｅ段階に相当する処理としてＡｇｇｒｅｇａｔｉｏｎが定義されている。これは複数の分散したサーバに格納されたデータを、ユーザプログラムが指定した方式でグループ化し処理を施すものであるが、ユーザプログラムが指定しなくとも、Ａｇｇｒｅｇａｔｉｏｎで定義された処理が、まず同一のラックに存在するサーバに存在するデータが一つのサーバにまとめられて実行され、次にラック間でまとめられて実行される。これによりラック間におけるデータ通信量が減る。このような処理を行うために、この分散データ処理システムは、サーバとラックとの接続関係情報であるネットワーク・トポロジー情報が予め存在していることを前提にしている。

また、ネットワークにおいて座標を用いて遅延を推定する技術が非特許文献５に記載されている。非特許文献５に記載された方法では、計測された遅延を主成分分析することで、座標変換に必要な変換行列を作成し、データ格納サーバから複数の計測用サーバへの遅延測定結果に対して、この変換行列を施すことで、座標を算出する。具体的には、予め計測用サーバが他の計測用サーバとの遅延を測定し、その計測結果を行列で表す。この行列をＤとする。例えば、計測用サーバが４台存在し、１台目の計測用サーバの全ての計測用サーバへの遅延測定結果がベクトル（０，１，３，３）で表されるとすると、行列Ｄの１行目にはこのベクトルが与えられる。このベクトルの要素は、遅延の測定結果である。各計測用サーバが同様にこのベクトルを求めることで、行列Ｄが求められる。この行列Ｄの主成分分析を行うために、ＤをＳＶＤ分解（Singular Value Decomposition、特異値分解）すると、以下に示す式（１）のように直交行列Ｕが得られる。

Ｄ ＝ ＵＷＶ^Ｔ 式（１）

なお、Ｗは特異値行列、Ｖは直交行列、Ｖ^ＴはＶの転置行列である。

この行列Ｕのうち、予め定められた次元数分の列を取り出した行列Ｕ_ｎが、座標変換に必要な変換行列である。

ネットワークにおける自身の座標を算出しようとするサーバは、例として挙げた４台の計測用サーバとの遅延を測定して、これをベクトルｄとする。また、座標を算出しようとするサーバは、変換行列Ｕ_ｎを取得し、以下に示す式（２）の計算を行うことで自身の座標ｃを算出する。

ｃ ＝ Ｕ_ｎ ^Ｔｄ 式（２）

Ｊｅｆｆｒｅｙ Ｄｅａｎ，Ｓａｎｊａｙ Ｇｈｅｍａｗａｔ，"ＭａｐＲｅｄｕｃｅ： Ｓｉｍｐｌｉｆｉｅｄ Ｄａｔａ Ｐｒｏｃｅｓｓｉｎｇ ｏｎ Ｌａｒｇｅ Ｃｌｕｓｔｅｒｓ"，Ｐｒｏｃｅｅｄｉｎｇｓ ｏｆ ｔｈｅ ｓｉｘｔｈ Ｓｙｍｐｏｓｉｕｍ ｏｎ Ｏｐｅｒａｔｉｎｇ Ｓｙｓｔｅｍ Ｄｅｓｉｇｎ ａｎｄ Ｉｍｐｌｅｍｅｎｔａｔｉｏｎ（ＯＳＤＩ’０４），２００４年１２月６日 西田 圭介，"Ｇｏｏｇｌｅを支える技術"，技術評論社，ｐ．６３−７５，ｐ．１３７−１５７ Ｓａｎｊａｙ Ｇｈｅｍａｗａｔ, Ｈｏｗａｒｄ Ｇｏｂｉｏｆｆ, ａｎｄ Ｓｈｕｎ−Ｔａｋ Ｌｅｕｎｇ，"Ｔｈｅ Ｇｏｏｇｌｅ Ｆｉｌｅ Ｓｙｓｔｅｍ"，Ｐｒｏｃｅｅｄｉｎｇｓ ｏｆ ｔｈｅ ｎｉｎｅｔｅｅｎｔｈ ＡＣＭ ｓｙｍｐｏｓｉｕｍ ｏｎ Ｏｐｅｒａｔｉｎｇ ｓｙｓｔｅｍｓ ｐｒｉｎｃｉｐｌｅｓ（ＳＯＳＰ’０３），２００３年１０月１９日 Ｍｉｃｈａｅｌ Ｉｓａｒｄ，Ｍｉｈａｉ Ｂｕｄｉｕ，Ｙｕａｎ Ｙｕ，Ａｎｄｒｅｗ Ｂｉｒｒｅｌｌ，Ｄｅｎｎｉｓ Ｆｅｔｔｅｒｌｙ，"Ｄｒｙａｄ：Ｄｉｓｔｒｉｂｕｔｅｄ Ｄａｔａ−Ｐａｒａｌｌｅｌ Ｐｒｏｇｒａｍｓ ｆｒｏｍ Ｓｅｑｕｅｎｔｉａｌ Ｂｕｉｌｄｉｎｇ Ｂｌｏｃｋｓ"，ＥｕｒｏＳｙｓ’０７，ＡＣＭ，２００７年 Ｈｙｕｋ Ｌｉｍ，Ｊｅｎｎｉｆｅｒ Ｃ． Ｈｏｕ，Ｃｈｏｎｇ Ｈｏ Ｃｈｏｉ，"Ｃｏｎｓｔｒｕｃｔｉｎｇ Ｉｎｔｅｒｎｅｔ Ｃｏｏｒｄｉｎａｔｅ Ｓｙｓｔｅｍ Ｂａｓｅｄ ｏｎ Ｄｅｌａｙ Ｍｅａｓｕｒｅｍｅｎｔ"，Ｐｒｏｃｅｅｄｉｎｇｓ ｏｆ ｔｈｅ ３ｒｄ ＡＣＭ ＳＩＧＣＯＭＭ ｃｏｎｆｅｒｅｎｃｅ ｏｎ Ｉｎｔｅｒｎｅｔ ｍｅａｓｕｒｅｍｅｎｔ（ＩＭＣ’０３）、２００３年１０月２７日

分散データ処理では、入力とするデータが存在するサーバに最も近いサーバで処理を実行させるように処理の実行場所（サーバ）を決定する場合がある。あるいは、分散したデータを局所的に集約して、その局所集約されたデータに対して集約処理を行うように、局所実行するサーバ群を決定し、集約する処理を実行するサーバを決定する場合がある。しかし、そのような場合において、アドレスの類似性、あるいは物理的接続の情報（ネットワークの物理的な構成の情報）に基づいてネットワーク内でのサーバの近傍性を推定すると、データアクセスの速度が遅くなったり、あるいはネットワークが輻輳を起こしやすくなったりすることがある。ネットワーク内でのメッセージの到達先を表すアドレスについて、その先頭ビットが共通といった類似性では、遅延の推定精度は低い。そのため、実際には応答がより速いサーバが存在していても、より遠くのデータ管理サーバにアクセスする可能性が高い。その結果、データの入出力に時間がかかってしまう。また、ネットワーク内で離れたサーバ間でデータ通信を行うと、その経路上のスイッチで輻輳が発生しやすくなってしまう。物理的接続の情報に基づいて、同一ラックに存在しているサーバが近いと判断して、ラック単位でデータを集約して処理を行っても、実際には、ネットワーク通信における遅延が短いとは限らず、そのため、ネットワークに負荷がかかることもある。

また、分散処理システムがアドレスに基づいて近傍性を推定する場合、既にアドレス体系が決定されたシステムにその分散処理システムを導入することが困難であるという問題がある。その理由は、ネットワーク内でメッセージの到達先を表すアドレスで以て近傍性を推定するには、データセンタ、ラック、サーバといった物理的な設備の収容関係をアドレス体系に反映させる必要があるが、通常はある程度物理的な近傍性が反映されるものの、アドレス体系はセキュリティを考慮して組織毎に決定されたり、動的に変更されたりするものであるからである。

また、分散データ処理において、入力とするデータが存在するサーバに最も近いサーバで処理を実行させるように処理の実行場所を決定する際、ラックとサーバの収容関係といったネットワークトポロジの情報（物理的接続の情報）を用いて近傍性を判断する場合、ネットワークトポロジを管理するシステムが必要となるため、分散処理システムの導入が困難になるという問題がある。ネットワークトポロジ管理システムは、物理的な接続関係やアドレス構成などのネットワークトポロジを常時監視する。また、場合によっては、運用者がネットワークトポロジ管理システムに対してネットワークトポロジを入力設定する必要がある。このようなネットワークトポロジ管理システムが導入されていないシステムには、分散処理システムの導入が困難である。また、ネットワークトポロジ管理システムが導入されているシステムであっても、ネットワークトポロジの情報を分散処理システムに与えられるように、ネットワークトポロジ管理システムと分散処理システムとを接続させるサブシステムが別途必要となる。

また、分散データ処理において、入力とするデータが存在するサーバに最も近いサーバで処理を実行させるように処理の実行場所を決定する際、データの格納先と同一のサーバ、あるいは同一のラックに収容されたサーバに処理を割り当てると、最適な割り当てであることを検証できず、最適な割り当てとは異なる割り当てとなっていることがある。すなわち、同一のラックに収容されているサーバを近傍のサーバとする等の基準に従うと、実際には他のラック内に存在するサーバの遅延の方が同一ラック内のサーバより小さいとしても、同一ラックに収容されたサーバに処理を割り当ててしまう。

そこで、本発明は、ネットワーク負荷や通信遅延の増大を防止し、輻輳の発生を抑えることができる分散データ処理システム、データ処理装置、分散データ処理方法および分散データ処理用プログラムを提供することを目的とする。また、ネットワークのアドレス体系を変更したり再設計したり、あるいは、ネットワークトポロジ管理システムを構築したりする必要がなく、容易に既存のシステムに導入することができる分散データ処理システム、データ処理装置、分散データ処理方法および分散データ処理用プログラムを提供することを目的とする。

本発明による分散データ処理システムは、データを処理する複数のデータ処理装置と、個々のデータ処理装置がそれぞれ算出したデータ処理装置自身のコンピュータネットワークにおける座標を記憶する座標管理装置と、クライアントから分散処理の要求を受信すると、各データ処理装置の負荷状態に基づいて、要求された分散処理を実行するワーカと、ワーカが実行する分散処理を管理するマスタとをデータ処理装置の中から決定する割当装置とを備え、割当装置が、各データ処理装置の負荷状態を示す情報を記憶するサーバ状態記憶手段と、クライアントから分散処理の要求を受信したときに、負荷状態を示す情報に基づいてワーカとなるデータ処理装置およびマスタとなるデータ処理装置を決定し、決定した各データ処理装置にクライアントからの要求を送信し、マスタに決定したデータ処理装置に分散処理の管理開始命令を送信する割当手段とを含み、各データ処理装置が、処理対象となるデータを記憶するデータ記憶手段と、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出手段と、二つの座標が指定されると、その二つの座標間での通信遅延を計算する遅延計算手段と、管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および各ワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延計算手段に計算させ、通信遅延を用いてどのワーカに処理を実行させるかを決定し、そのワーカに処理の命令を送信するマスタ処理手段と、自装置以外のデータ処理装置のマスタ処理手段から処理の命令を受けたときに、その処理を実行するワーカ処理手段とを含むことを特徴とする。

また、本発明によるデータ処理装置は、分散データ処理システム内で処理を実行するデータ処理装置であって、処理対象となるデータを記憶するデータ記憶手段と、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出手段と、二つの座標が指定されると、その二つの座標間での通信遅延を計算する遅延計算手段と、クライアントが要求した分散処理の管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および処理の実行のために選択されたワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延計算手段に計算させ、通信遅延を用いてどのワーカに処理を実行させるかを決定し、そのワーカに処理の命令を送信するマスタ処理手段と、自装置以外のデータ処理装置から処理の命令を受けたときに、その処理を実行するワーカ処理手段とを備えることを特徴とする。

また、本発明による分散データ処理方法は、データを処理する複数のデータ処理装置が、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定し、割当装置が、クライアントから分散処理の要求を受信したときに、各データ処理装置の負荷状態を示す情報に基づいて、要求された分散処理を実行するワーカと、ワーカが実行する分散処理を管理するマスタとを決定し、決定した各データ処理装置にクライアントからの要求を送信し、マスタに決定したデータ処理装置に分散処理の管理開始命令を送信し、マスタとなるデータ処理装置が、管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および各ワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を計算し、通信遅延を用いてどのワーカに処理を実行させるかを決定し、そのワーカに処理の命令を送信し、ワーカとなるデータ処理装置が、マスタとなるデータ処理装置から処理の命令を受信したときに、その処理を実行することを特徴とする。

また、本発明による分散データ処理用プログラムは、分散データ処理システム内で処理を実行する各コンピュータに搭載される分散データ処理用プログラムであって、コンピュータに、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出処理、クライアントが要求した分散処理の管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各コンピュータおよび処理の実行のために選択されたワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を計算し、通信遅延を用いてどのワーカに処理を実行させるかを決定し、そのワーカに処理の命令を送信するマスタ処理、および、自装置以外のコンピュータから処理の命令を受けたときに、その処理を実行するワーカ処理を実行させることを特徴とする。

本発明によれば、ネットワーク負荷や通信遅延の増大を防止し、輻輳の発生を抑えることができる。また、容易に既存のシステムに導入することができる。

以下、本発明の実施形態を図面を参照して説明する。
本発明による分散データ処理システムは、分散したデータに処理を割り当てる際、あるいは、分散した処理の入出力データを配置する際に、データ入出力の通信資源を効率的に利用できるように、利用可能な複数のノードから、処理の割り当て先、あるいはデータ配置先を決定する。

図１は、本発明の分散データ処理システムの実施形態を示すブロック図である。本実施形態の分散データ処理システムは、サーバ１２１〜１２９と、データ管理マスタ１１３と、ワークキュー１１２と、座標管理マスタ１１４と、計測用サーバ１４１〜１４３と、クライアント１１１とを備える。サーバ１２１〜１２９、データ管理マスタ１１３、ワークキュー１１２、座標管理マスタ１１４、計測用サーバ１４１〜１４３、およびクライアント１１１は、それぞれプログラム制御により動作するコンピュータ（中央処理装置、プロセッサ、データ処理装置）によって実現され、ネットワーク１０１〜１０４によって接続されている。ネットワーク１０１〜１０４は、複数のコンピュータを相互接続させる。

図２は、各サーバの構成例を示すブロック図である。各サーバ１２１〜１２９は、図２に示すように、データ格納部１５１と、ワーカ処理手段１５２と、座標算出手段１５３と、マスタ処理手段１５６と、遅延推定手段１５７と、負荷状態通知手段１５８とを備える。ただし、図１では、サーバ１２１がワーカとして動作する場合を例にし、サーバ１２１の要素としてデータ格納部１５１、ワーカ処理手段１５２および座標算出手段１５３のみを示している。また、図１では、サーバ１２５がマスタとして動作する場合を例にし、サーバ１２５の要素として、データ格納部１５１、座標算出手段１５３と、マスタ処理手段１５６と、遅延推定手段１５７のみを図示している。また、他のサーバについては、これらの要素の図示を省略している。

計測用サーバ１４１〜１４３は、コンピュータネットワークにおける他の装置から、通信遅延測定メッセージを受けると、その送信元の装置に応答する。通信遅延測定メッセージは、コンピュータが計測用サーバ１４１〜１４３との通信遅延を測定するために計測用サーバ１４１〜１４３に対して送信するメッセージである。通信遅延測定メッセージの送信元となるコンピュータ（例えば、各サーバ）は、通信遅延測定メッセージを送信し、計測用サーバ１４１〜１４３との通信遅延（計測用サーバとの間の通信において生じる遅延時間）を計測する。

なお、サーバ１２１〜１２９が計測した計測用サーバ１４１〜１４３との通信遅延は、コンピュータネットワークにおける個々のサーバの座標を算出するために用いられる。計測用サーバ１４１〜１４３は、自身を含む各計測用サーバとの通信遅延を計測し、少なくとも一つの計測用サーバが、各サーバが通信遅延から座標を求めるための情報（例えば、前述のＵ_ｎ ^Ｔ）を導出しておいてもよい。

次に、図１および図２を参照して、サーバ１２１〜１２９の構成について説明する。
データ格納部１５１は、分散データ処理システムによる処理対象となるデータを記憶する記憶装置である。各データは、一意のデータ名を有している。各サーバ１２１〜１２９は、他の装置からの読み込みあるいは書き込み要求に応じて、指定されたデータの値を返したり、指定されたデータについて値の作成、更新、削除等を行う。

図１に示す例では、各サーバ１２１〜１２９が、データ格納部１５１にデータ１３１〜１３９を記憶している場合を例示している。

座標算出手段１５３は、コンピュータネットワークにおける自装置（サーバ）の位置を多次元の実数値で表した座標を求める。このとき、座標算出手段１５３は、各計測用サーバ１４１〜１４３に対して通信遅延測定メッセージを送信し、計測用サーバ１４１〜１４３との通信遅延を測定する。そして、その通信遅延から、実数の多次元座標を求める。なお、ここでは、計測用サーバの応答待ち時間（計測用サーバにメッセージを送信して計測用サーバから応答が返ってくるまでの時間）を通信遅延として用いる場合を例にして説明するが、通信遅延は応答待ち時間に限定されず、応答待ち時間以外の値であってもよい。座標算出手段１５３は、個々の計測用サーバとの通信遅延をそれぞれ座標値として定めてもよいが、次元圧縮を行うことが好ましい。

このような次元圧縮は、非特許文献５に記載された方法で行ってもよい。非特許文献５に記載の方法では、予め個々の計測用サーバが、それぞれ他の計測用サーバとの通信遅延を測定し、その結果を行列Ｄで表す。既に説明したように、仮に計測用サーバが４台存在し、１台目の計測用サーバの全ての計測用サーバへの遅延測定結果がベクトル（０，１，３，３）で表されるとすると、行列Ｄの１行目にはこのベクトルが与えられる。同様に、ｎ台目の計測用サーバにおける計測結果を示すベクトルを行列Ｄの第ｎ行として、行列Ｄを表せばよい。なお、行列Ｄを特定するためには、いずれかの計測用サーバが、他の各行列用サーバから通信遅延測定結果を収集すればよい。その計測用サーバは、行列Ｄを定めると、行列Ｄに対するＳＶＤ（Singular Value Decomposition）分解を行い、Ｄ＝ＵＷＶ^Ｔの形式で表わす。ＳＶＤ分解により得られる所定の行列Ｕのうち、予め定められた次元数分の列を取り出した行列をＵ_ｎとすると、通信遅延から座標を求めるための情報として行列Ｕ_ｎの転置行列Ｕ_ｎ ^Ｔを各サーバ１２１〜１２９に提供する。この行列Ｕ_ｎ ^Ｔを、座標軸情報と呼ぶことがある。各サーバの座標算出手段１５３は、計測用サーバとの通信遅延を測定したならば、その測定結果を要素とするベクトルｄを定め、式（２）の計算を行うことにより、自装置の座標ｃを算出すればよい。

ワーカ処理手段１５２は、マスタとして動作する他のサーバのマスタ処理手段１５６から命令を受信すると、その命令に従って処理を実行する。例えば、マスタに指定されたデータを用いて処理を行う。また、例えば、処理中に生成したデータを、マスタに指定されたデータ格納部１５１に出力する。マスタに指定されたデータは、自装置のデータ格納部１５１に記憶されたデータである場合も、他のサーバのデータ格納部１５１に記憶されたデータである場合もあり得る。また、マスタに指定されたデータ格納部１５１も、自装置のデータ格納部１５１に限らず、他のサーバのデータ格納部１５１であることもある。

マスタがクライアント１１１（図１参照）からの要求された分散処理を各ワーカに割り当て、ワーカ処理手段１５２は、マスタからの命令により、マスタが割り当てた処理を実行することになる。

遅延推定手段１５７は、マスタ処理手段１５６から２つの座標を指定されると、その座標間の距離を計算することにより、その２つの座標間での通信遅延を計算する。

マスタ処理手段１５６は、ワークキュー１１２から、分散処理の管理開始命令を受信したときに、以下の動作を開始する。なお、マスタ処理手段１５６には、クライアント１１１からの要求（プログラム）、分散処理の管理開始命令、どのサーバがワーカとして動作するかを示す情報が、ワークキュー１１２から送られる。

マスタ処理手段１５６は、ワークキュー１１２を介して受信したクライアント１１１からの要求（プログラム）を解釈し、プログラム中で指定されたデータを記憶しているサーバの情報をデータ管理マスタ１１３から取得する。また、マスタ処理手段１５６には、ワークキュー１１２から、どのサーバがワーカとして動作するかを示す情報も通知される。マスタ処理手段１５６は、クライアント１１１から要求された分散処理に関連する各サーバおよび各ワーカの座標から得られる二つの座標の組み合わせをそれぞれ指定し、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。

なお、要求された分散処理に関連するサーバはワーカに限定されない。要求された分散処理に関連するサーバであってワーカ以外のサーバ（以下、要求された分散処理に関連するワーカ以外のサーバ）として、例えば、クライアント１１１からの要求として受信したプログラムで指定されたデータを記憶するワーカ以外のサーバが挙げられる。また、ワーカが生成したデータを取得すべきワーカ以外のサーバも該当する。なお、ワーカ以外のサーバとは、ワーカでないサーバを意味し、非ワーカのサーバということもできる。

マスタ処理手段１５６は、要求された分散処理に関連するサーバやワーカの座標から計算された通信遅延を用いて、どのワーカに処理を実行させるのかを決定し、そのワーカに処理の実行命令を送信して、処理を実行させる。例えば、マスタ処理手段１５６は、通信遅延と、要求された分散処理に関連するワーカ以外のサーバとワーカとの間で送受信されるデータ量とから、ワーカの処理実行時におけるネットワーク負荷を計算する。換言すれば、ネットワーク負荷を見積もる。そして、マスタ処理手段１５６は、ネットワーク負荷に基づいて、処理を実行させるワーカを決定する。

マスタ処理手段１５６は、サーバ間のネットワーク負荷を計算する場合、そのサーバ間の通信遅延と、そのサーバ間で送受信されるデータ量との積をネットワーク負荷として計算してもよい。例えば、ワーカＡが、要求された分散処理に関連するワーカ以外のサーバＢとデータを送受信して処理を行う場合のネットワーク負荷は、ワーカＡとサーバＢ間の通信遅延に、ワーカＡとサーバＢが送受信するデータ量の積として求めればよい。このネットワーク負荷の計算方法は例示であり、ワーカの処理実行時におけるネットワーク負荷を他の計算方法で見積もってもよい。

マスタ処理手段１５６が処理を実行させるワーカを決定する態様の例として、以下の態様が挙げられる。

例えば、処理対象となるデータを記憶するワーカ以外のサーバが存在しているとする。マスタ処理手段１５６は、そのサーバと、一つ一つの各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少のワーカに、そのサーバからデータを取得させ処理を実行させることを決定する。

また、ワーカが処理を行って生成したデータを、他のワーカが入力データとして用いて処理を実行していく場合もある。その場合においても、ワーカと、一つ一つの他の各ワーカの組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少の他のワーカに、処理を完了したワーカからデータを取得させ次の処理を実行させることを決定する。

また、例えば、ワーカが、処理対象となるデータを記憶していて、そのデータ量が多い場合（例えば、閾値より多い場合）、マスタ処理手段１５６は、そのワーカと、一つ一つの他の各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少となる他のワーカを決定する。マスタ処理手段１５６は、決定した他のワーカに、データを有しているワーカの一部のデータを取得させ処理を実行させる。また、データを記憶していたワーカには、その残りのデータを用いて処理を実行させる。

また、例えば、ワーカが処理を行って生成したデータを取得すべきサーバ（ワーカであっても非ワーカであってもよい）が存在する場合がある。例えば、そのようなサーバが、クライアントからの要求内で定められている場合がある。マスタ処理手段１５６は、そのサーバと、一つ一つの各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少のワーカに、処理を実行させ、その処理で生成したデータを、そのサーバに出力させる。

マスタ処理手段１５６は、一つのワーカにのみ処理を実行させるわけではない。クライアント１１１からの要求（プログラム）で指定されている処理の並列性や、データの入出力関係に応じて、処理を実行させるべきワーカを、上記に例示した態様等で順番に決定していけばよい。このように、マスタとなるサーバは、ワーカとなるサーバが実行する分散処理を管理する。

マスタ処理手段１５６は、各ワーカが生成したデータを特定のワーカに集約させてもよい。データを集約するワーカ（すなわち、各ワーカが生成したデータを集めるワーカ）を集約ワーカと呼ぶ。マスタ処理手段１５６は、集約ワーカを定めて、集約ワーカに他の各ワーカからデータを取得させる。

集約ワーカを定める場合、マスタ処理手段１５６は、例えば、各ワーカの座標を入力として、クラスタリングアルゴリズムを用いてワーカをグルーピングして、グループの中のいずれかのワーカを集約ワーカに決定すればよい。この場合、クラスタリングアルゴリズムとして、例えば、階層的クラスタリングアルゴリズムを用いるとよい。

マスタ処理手段１５６は、各ワーカに対して、処理の実行命令を送信することで、各ワーカに各種処理を実行させればよい。例えば、ワーカに、他のサーバからデータを取得して処理を実行させる場合、その旨の命令をワーカに送信すればよい。ワーカのワーカ処理手段１５２は、この命令に従って、例えば、データの取得、処理の実行、データの出力などを行う。

負荷状態通知手段１５８は、自装置の負荷状態を検出し、その負荷状態を示す情報および自装置の識別情報をワークキュー１１２に送信する。ここで負荷状態とは、例えば、何らかの処理を実行しているという状態、あるいはしていないという状態、またはＣＰＵ（中央処理装置）の利用率といったサーバの計算負荷を表す状態である。

ワーカ処理手段１５２、座標算出手段１５３、マスタ処理手段１５６、および遅延推定手段１５７は、例えば、分散データ処理用プログラムに従って動作するＣＰＵによって実現される。個々のサーバの記憶装置に分散データ処理用プログラムが記憶され、各サーバのＣＰＵはそれぞれ分散データ処理用プログラムを読み込み、分散データ処理用プログラムに従ってワーカ処理手段１５２、座標算出手段１５３、マスタ処理手段１５６、遅延推定手段１５７、および負荷状態通知手段１５８として動作してもよい。

次に、図１を参照して、データ管理マスタ１１３、ワークキュー１１２、座標管理マスタ１１４、およびクライアントについて説明する。

データ管理マスタ１１３は、アドレス格納部１１７を備える。アドレス格納部１１７は、分散データ処理システムにおいて、データ毎に一意に与えられているデータ名と、そのデータ名から特定されるデータを記憶するサーバの情報（アドレス）とを対応付けて記憶する記憶装置である。一つのデータ名に対応するサーバの数は１つであっても複数であってもよい。複数の場合には、同一のデータが複数のサーバに記憶されていることになる。

また、データ管理マスタ１１３が備えるＣＰＵ（図示せず）は、サーバ（例えば、マスタとして動作するサーバ）からの問い合わせに応じて、指定されたデータ名に対応するサーバの情報を送り返す。すなわち、データ名を受信すると、そのデータ名に対応するサーバのアドレスをアドレス格納部１１７から読み込み、問い合わせに対する応答として送信する。従って、各サーバは、データ名に対応するサーバをデータ管理マスタ１１３に問い合わせることによって、そのデータ名のデータを記憶しているサーバを特定することができる。

なお、アドレス格納部１１７にデータ名とアドレスとの対応関係を記憶させる方法は特に限定されない。例えば、各サーバ１２１〜１２９が、自装置のデータ格納部１５１に新規にデータを記憶させたならば、そのデータのデータ名と自装置のアドレスとをデータ管理マスタ１１３に通知し、データ管理マスタ１１３のＣＰＵ（図示せず）がそのデータ名およびアドレスをアドレス格納部１１７に記憶させてもよい。あるいは、クライアント１１１が新規データ（例えば新規ファイル）の生成を要求する場合、クライアント１１１がデータ管理マスタ１１３に新規データの生成を要求し、データ管理マスタ１１３のＣＰＵ（図示せず）がどのサーバに新規データを格納すべきかを指定し、データ名とそのサーバのアドレスとをアドレス格納部１１７に記憶させてもよい。この場合、そのサーバのアドレスを含むファイルハンドラをクライアント１１１に返し、クライアント１１１は、そのファイルハンドラを使って新規データをサーバに書き込む。

ワークキュー１１２は、サーバ状態格納部１１６とサーバ割当手段１１９とを有する。サーバ状態格納部１１６は、複数のサーバ１２１〜１２９の負荷状態を示す情報を記憶する記憶装置である。ワークキュー１１２のＣＰＵは、各サーバの負荷状態通知手段１５８（図２参照）が送信した負荷状態を示す情報およびサーバの識別情報を受信すると、両者を対応付けてサーバ状態格納部１１６に記憶させる。

サーバ割当手段１１９は、クライアント１１１から分散処理の要求（プログラム）を受けると、サーバ状態格納部１１６に記憶された負荷状態の情報を用いて、各サーバ１２１〜１２９の中からワーカとなるサーバおよびマスタとなるサーバを決定する。例えば、ｘ台のサーバを用いて分散処理を実行する要求をクライアント１１１から受信している場合、サーバ割当手段１１９は、サーバ状態格納部１１６は、負荷状態の低い順にｘ台のサーバを選択し、そのｘ台のうち１台をマスタに決定し、残りをワーカに決定する。ただし、ここではマスタが１台である場合を例にしているが、冗長化のために２台以上のサーバをマスタとして定めてもよい。また、上記の例では、負荷状態の低い順にサーバを選択する場合を例示したが、負荷状態が低いという制約の他に、サーバの物理的な制約や論理的な制約を満たすように、サーバを選択してもよい。物理的な制約とは、処理に要するメモリ量などであり、論理的な制約とは、何らかのライブラリ、ＯＳ、ソフトウェアが搭載されていること等の制約である。例えば、クライアント１１１からの要求に、これらの制約が指定されている場合、その制約を満足しているサーバの中から、負荷状態の低い順に選択すればよい。なお、この場合、各サーバのメモリ量や、各サーバにどのようなソフトウェアが搭載されているか等を予めサーバ状態格納部１１６に記憶させておけばよい。

サーバ割当手段１１９は、マスタまたはワーカとして決定した各サーバに、クライアント１１１から受信した分散処理の要求（プログラム）を送信する。また、このとき、マスタに決定したサーバに、ワーカに決定したサーバを通知するとともに、分散処理の管理開始命令を送信する。

また、ワークキューとして動作する装置と、マスタとして動作するサーバとが同一装置であっても良い。

座標管理マスタ１１４は、座標格納部１１８を備える。座標格納部１１８は、個々のサーバの座標算出手段１５３がそれぞれ算出したサーバの座標を、サーバ毎にそれぞれ記憶する記憶装置である。サーバの座標算出手段１５３は、自装置の座標を算出したときにその座標と自装置の識別情報を座標管理マスタ１１４に送信すればよい。そして、座標管理マスタ１１４のＣＰＵ（図示せず）は、座標およびサーバの識別情報を受信すると、両者を対応付けて座標格納部１１８に記憶させればよい。

座標管理マスタ１１４のＣＰＵ（図示せず）は、サーバから、別のサーバを指定してそのサーバの座標の問い合わせを受けると、そのサーバに対応する座標を座標格納部１１８から読み込んで、問い合わせ元のサーバに返す。例えば、座標の問い合わせ元のサーバから、他のサーバの識別情報を受信したときに、その識別情報に対応する座標を検索し、その座標を問い合わせ元のサーバに送信する。

また、座標管理マスタ１１４のＣＰＵ（図示せず）は、特定の装置に関する座標の登録要求、修正要求、削除要求があった場合、その要求に応じて、座標の削除、修正、追加を行う。

クライアント１１１は、分散データ処理システムのユーザにより編集されたプログラムの処理を依頼する処理依頼手段１１５を有する。このプログラムは、分散処理に含まれる各処理をコンピュータ（サーバ）に実行させるためのプログラムである。例えば、どのデータを用いてどのような処理を行うか、生成したデータをどの装置に出力するか等の手順が記述される。処理依頼手段１１５は、このプログラムをワークキュー１１２に送信することにより、分散処理を要求する。

次に、本発明の動作の流れについて説明する。
図３は、各サーバが正常に稼働しているときに、負荷状態を示す情報や座標をワークキュー１１２や座標管理マスタ１１４に登録する動作の処理経過の例を示すフローチャートである。各サーバ１２１〜１２９の座標算出手段１５３は、各計測用サーバ１４１〜１４３に対して通信遅延測定メッセージを送信し、計測用サーバ１４１〜１４３との通信遅延を測定する。そして、座標算出手段１５３は、その測定結果から自装置の座標を導出し、自装置の座標を、自装置の情報（例えば、識別情報）とともに座標管理マスタ１１４に送信する（ステップＳ２０１）。座標管理マスタ１１４は、これらの情報を受信すると、座標格納部１１８に記憶させる。

ステップＳ２０１の後、各サーバ１２１〜１２９の負荷状態通知手段１５８（図２参照）は、サーバの負荷状態を示す情報を検出し、自装置の識別情報とともに、ワークキュー１１２に送信する（ステップＳ２０２）。サーバの負荷状態には、ワーカ処理手段１５２の処理負荷や、マスタ処理手段１５６の処理負荷も含まれる。負荷状態通知手段１５８は、例えば、ＣＰＵ利用率等で処理負荷を表し、ワークキュー１１２に送信すればよい。ただし、ＣＰＵ利用率は例示であり、負荷状態を示す情報を他の指標で表してもよい。ワークキュー１１２のＣＰＵは、各サーバから負荷状態を示す情報および識別情報を受信すると、両者を対応付けてサーバ状態格納部１１６に記憶させる。

ステップＳ２０２の次に、各サーバ１２１〜１２９は、終了フラグがオンになっているか否かを判定する（ステップＳ２０３）。終了フラグは、サーバに異常が発生したときにサーバ自身によってオンに設定されるフラグであり、終了フラグがオンになっているということは、サーバの正常時の処理を終了させることを意味する。ステップＳ２０３において、終了フラグがオフであれば、座標算出手段１５３および負荷状態通知手段１５８は、一定時間待機し（ステップＳ２０４）、ステップＳ２０１以降の処理を繰り返す。ステップＳ２０３において、終了フラグがオンであれば、処理を終了する。

次に、クライアント１１１が分散処理を要求し、マスタがワーカに分散処理を実行させる処理について説明する。図４は、この動作の流れを示すフローチャートである。クライアント１１１の処理依頼手段１１５は、分散処理の内容を記述した分散処理プログラムが分散データ処理システムのユーザにより編集されると、その分散処理プログラムをワークキュー１１２に送信することによって、分散処理を依頼する。ワークキュー１１２のサーバ割当手段１１９は、プログラムとして記述された分散処理要求をクライアント１１１から受信すると、サーバ状態格納部１１６に記憶された各サーバの負荷状態を参照して、マスタとして動作するサーバ、およびワーカとして動作するサーバを選択する（ステップＳ３０１）。

既に説明したように、サーバ割当手段１１９は、クライアント１１１からの要求において指定された分散処理で使用するサーバの台数分のサーバを、負荷状態の低い順に選択すればよい。このとき、サーバの物理的な制約や論理的な制約が定められているならば、その制約を満たすサーバを、指定された台数分選択すればよい。

続いて、ワークキュー１１２のサーバ割当手段１１９は、指定された台数分のサーバを選択できたか否かを判定する（ステップＳ３０２）。サーバを選択できていなければ（ステップＳ３０２におけるＮＯ）、異常終了として処理を終了する。

サーバの選択に成功したならば（ステップＳ３０２におけるＹＥＳ）、ワークキュー１１２のサーバ割当手段１１９は、例えば、選択したサーバのうちの１台をマスタに決定し、残りをワーカに決定する。そして、ワーカおよびマスタに定めた各サーバに対して、クライアントから受信した分散処理要求（プログラム）を送信する。このとき、サーバ割当手段１１９は、ワーカとなる各サーバの情報と、管理開始命令も送信する（ステップＳ３０３）。

続いて、マスタとされたサーバは、要求された処理を、ワーカとなる各サーバに振り分け、振り分けた処理を各ワーカに実行させる（ステップＳ３０４）。マスタとされたサーバは、ワーカに振り分けた処理の実行が成功していれば（ステップＳ３０５におけるＹＥＳ）、処理を正常に終了し、処理の実行に失敗していれば（ステップＳ３０５におけるＮＯ）、異常終了として処理を終了する。

図５は、マスタ（マスタとされたサーバ）のマスタ処理手段１５６のステップＳ３０４における処理経過の例を示す説明図である。マスタのマスタ処理手段１５６は、クライアントが送信した分散処理要求（プログラム）を解釈して、分散処理で使用するデータを記憶しているサーバの情報をデータ管理マスタ１１３から取得する（ステップＳ４０１）。すなわち、マスタ処理手段１５６は、受信したプログラムを解釈して、分散処理で使用する各データのデータ名を特定する。そして、その各データ名をデータ管理マスタ１１３に送信することにより、データ名が示すデータを記憶しているサーバを問い合わせる。データ管理マスタ１１３は、マスタから受信したデータ名に対応するサーバのアドレスをアドレス格納部１１７から読み込み、マスタに送信すればよい。

次に、マスタ処理手段１５６は、分散処理で使用するデータを記憶しているサーバの座標、およびワーカ（ワーカとされたサーバ）の情報を座標管理マスタ１１４に送信し、それらの各サーバの座標を問い合わせ、座標管理マスタ１１４からその各サーバの座標を取得する（ステップＳ４０２）。座標管理マスタ１１４は、マスタからサーバの情報を受信したならば、そのサーバの座標を座標格納部１１８から読み込み、マスタに送信すればよい。

次に、マスタ処理手段１５６は、分散処理で使用するデータを記憶しているサーバの座標、およびワーカの座標の組み合わせを作成し、作成した組毎に、座標間の距離（すなわち、通信遅延）を遅延推定手段１５７に算出させる。遅延推定手段１５７は、２つの座標からなる組を指定されると、その座標間のユークリッド距離を計算すればよい。この計算結果が通信遅延である。マスタ処理手段１５６は、組毎に通信遅延が求められると、組をなすサーバ間で送受信されるデータ量と通信遅延からネットワーク負荷を見積もる（ステップＳ４０３）。例えば、データ量と通信遅延の積をネットワーク負荷としてもよい。

マスタ処理手段１５６は、ネットワーク負荷の見積もり結果を参照して、クライアント１１１の要求に含まれる各処理を、どのワーカに実行させるのかを決定する（ステップＳ４０４）。

ステップＳ４０４では、例えば、どのワーカに、どのサーバ（ワーカとは限らない）のデータを入力データとして、要求された処理を実行させるのかを決定する。

具体例を挙げると、処理対象となるデータを記憶するワーカ以外のサーバが存在している場合、マスタ処理手段１５６は、そのサーバと、一つ一つの各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少のワーカに、そのサーバからデータを取得させ処理を実行させることを決定する。マスタ処理手段１５６は、決定した内容の実行命令を、そのワーカに送信し、命令を受信したワーカは、サーバからデータを読み込んで、処理を実行する。

あるいは、ワーカが処理を行って生成したデータを、他のワーカが入力データとして用いて処理を実行する場合、マスタ処理手段１５６は、先行の処理を実行するワーカと、一つ一つの他の各ワーカの組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少の他のワーカに、先行処理を完了したワーカからデータを取得させ次の処理を実行させることを決定する。マスタ処理手段１５６は、決定した内容の実行命令を、決定したワーカに送信し、命令を受信したワーカは、先行の処理を実行したワーカからデータを読み込んで、処理を実行する。

また、ワーカが、処理対象となるデータを記憶していて、そのデータ量が多い場合、マスタ処理手段１５６は、そのワーカと、一つ一つの他の各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少となる他のワーカを決定する。マスタ処理手段１５６は、決定した他のワーカに、データを記憶しているワーカからデータを取得して処理を実行する命令を送信する。この命令を受信したワーカは、データを読み込んで、処理を実行する。また、マスタ処理手段１５６は、データを記憶していたワーカに対して残りのデータを用いて処理を実行する命令を送信し、命令を受信したワーカは残りのデータを用いて処理を実行する。

ステップＳ４０４では、例えば、クライアント１１１の要求に含まれる処理の実行時に生成されたデータを取得すべきサーバが定められているときに、マスタ処理手段１５６は、どのワーカにその処理を実行させるのかを決定する。

具体例を挙げると、そのようなサーバが、クライアントからの要求内で定められている場合、マスタ処理手段１５６は、そのサーバと、一つ一つの各ワーカとの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延推定手段１５７に計算させる。そして、通信遅延からネットワーク負荷を計算し、ネットワーク負荷が最少のワーカに、処理を実行させ、その処理で生成したデータを、そのサーバに対して出力させることを決定する。マスタ処理手段１５６は、決定した内容の実行命令を、そのワーカに送信し、命令を受信したワーカは、処理を実行して、その際に生成したデータを、指定されたサーバに出力する。

マスタ処理手段１５６は、クライアント１１１が要求した処理の並列性や、データの入出力関係に応じて、処理を実行させるワーカを、順番に決定していく。

さらに、マスタ処理手段１５６は、ステップＳ４０４において、各ワーカの座標を入力として、クラスタリングアルゴリズムを用いてグルーピングを行って、集約ワーカを定めて、各ワーカが生成したデータを集約ワーカに取得させる命令を集約ワーカに送信してもよい。命令を受信した集約ワーカは、命令に従って各ワーカからデータを取得する。例えば、各ワーカに対してデータを要求し、要求に応じて他のワーカが送信したデータを受信することによってデータを集約ワーカに集約させる。

また、局所的なワーカ群中のいずれかのワーカにデータを集約させ、さらにそのデータを別のワーカが集約するように、段階的にデータを集約させていってもよい。

ステップＳ４０４の後、クライアント１１１から要求された分散処理が成功していれば（ステップＳ４０５におけるＹＥＳ）、マスタ処理手段１５６は、処理を正常に終了する。一方、クライアント１１１から要求された分散処理が成功していなければ、異常終了として処理を終了する。

本実施形態では、通信遅延からネットワーク負荷を計算し、ネットワーク負荷の小さいワーカを選択する場合を示したが、処理を実行させる候補となるワーカが複数ある場合、それらのワーカについて、データを記憶しているサーバ（または、生成したデータを出力すべきサーバ）との通信遅延が最も小さいサーバを選択し、そのサーバに処理を実行させてもよい。

次に、本実施の形態の効果について説明する。
本発明によれば、各ワーカと他のサーバとの通信遅延を用いて、クライアントからの要求に含まれる各処理をどのワーカに実行させるのかを決定する。例えば、ワーカとサーバの座標から通信遅延を求め、その通信遅延とデータ量からワーカでの処理実行時におけるネットワーク負荷を求め、ネットワーク負荷の少ないワーカに処理を実行させる。従って、処理負荷の増加や通信遅延の増大を防止することができる。また、輻輳の発生を抑えることができる。また、不要な通信量の増加を防止でき、データの読み込みや書き込みなどを高速に行うことができる。

また、個々のサーバがそれぞれ座標算出手段１５３（図２参照）を備え、ネットワークにおける自装置の座標を算出する。そして、その座標を用いて、処理を実行させるワーカを上記のように決定する。従って、分散データ処理システムの導入に伴って、ネットワークのアドレス体系を変更したり再設計したりする必要がない。また、分散データ処理システムの他に別途ネットワークトポロジ管理システムを導入しなくてもよい。従って、既存のシステムに容易に本発明による分散データ処理システムを導入することができる。

また、各サーバが求めた自装置の座標を用いて、ワーカと他のサーバとの通信遅延を計算することができる。従って、ワーカと他のサーバとの組み合わせの最適性を評価することができる。上記の実施形態では、通信遅延を用いて計算したネットワーク負荷を用いて組み合わせの最適性を評価している。

図６は、本発明の実施例を示す説明図である。図１と同様の構成要素については、図１と同一の符号を付し、詳細な説明を省略する。

各計測用サーバ１４１〜１４３は、自身を含む各計測用サーバの通信遅延を計測する。計測用サーバ自身の通信遅延は０とすればよい。そして、各計測用サーバ１４１〜１４３のうちの任意の計測サーバが、他の各計測サーバから通信遅延の計測結果を収集する。そして、ｎ台目の計測用サーバにおける計測結果を示すベクトルを行列Ｄにおける第ｎ行として行列Ｄを決定し、主成分分析を行うことによって、各サーバが通信遅延から座標を求めるための情報を生成する。計測用サーバは、行列Ｄを決定後、ＳＶＤ分解を行い、行列ＤをＤ＝ＵＷＶ^Ｔの形式で表わし、行列Ｕから、予め定められた次元数の列を抽出し、その列からなる行列Ｕ_ｎを求め、さらにその転置行列Ｕ_ｎ ^Ｔを求める。この行列Ｕ_ｎ ^Ｔは、通信遅延から座標を求めるための情報（座標軸情報）に相当する。図７は、座標軸情報の例を示す。以下、図７に例示する座標軸情報が生成されている場合を例にして説明する。

また、各サーバ１２１〜１２９の座標算出手段１５３（図２参照）は、各計測用サーバ１４１〜１４３の通信遅延を計測する。なお、通信遅延計測では、例えば、測用サーバ１４１〜１４３に対してＰｉｎｇ等の測定用の通信データを送信し、その応答が返ってくるまでの時間を通信遅延として計測すればよい。サーバ１２１が計測した計測用サーバ１４１の通信遅延が２ｍｓであり、同様に、計測用サーバ１４２の通信遅延が５ｍｓであり、計測用サーバ１４３の通信遅延が６ｍｓであったとする。すると、サーバ１２１の座標算出手段５１３は、これらの計測結果（２ｍｓ、５ｍｓ、６ｍｓ）を要素とするベクトルをベクトルｄとし、式（２）の計算を行って、サーバ１２１の座標を計算すればよい。なお、座標算出手段１５３は、行列Ｕ_ｎを保持している計測用サーバに行列Ｕ_ｎ ^Ｔを要求して、Ｕ_ｎ ^Ｔを取得し、式（２）の計算を行えばよい。本例では、サーバ１２１の座標算出手段１５３は、２次元座標（４，２２）を得る。他のサーバ１２２〜１２９も同様に自装置の座標を計算する。

各サーバ１２１〜１２９の座標算出手段１５３は計算した自装置の座標および識別情報を座標管理マスタ１１４に送信し、座標管理マスタ１１４はその情報を受信すると座標格納部１１８に記憶させる。図８は、座標格納部１１８に記憶された情報の例を示す。

また、各サーバ１２１〜１２９のデータ格納部１５１には、予めデータ名がＡ〜Ｉであるデータ１３１〜１３９が記憶されている。どのサーバにどのデータが記憶されているかを示す情報は、データ管理マスタ１１３のアドレス格納部１１７に記憶されている。図９は、アドレス格納部１１７に記憶された情報の例を示す。本例では、アドレスとは別にサーバの識別情報が定められ、図９に示すように、データ名と識別情報とアドレス（ここではＩＰアドレス）とを対応付けて記憶する場合を例にする。

また、各サーバ１２１〜１２９の負荷状態通知手段１５８（図２参照）は、自装置の負荷状態を示す情報をワークキュー１１２に送信し、ワークキュー１１２は、各サーバの負荷状態を示す情報をサーバ状態格納部１１６に記憶させる。図１０は、サーバ状態格納部１１６に記憶された情報の例を示す。

図８〜図１０に例示する情報が生成されている状態において、クライアント１１１が、データ名Ａ，Ｂ，Ｅ，Ｆの各データに対してあるＭａｐ処理を施し、その結果をあるＲｅｄｕｃｅ処理でまとめる処理の実行をワークキュー１１２に要求したとする。このとき、６台のサーバを利用することを指定して、処理を要求していたとする。本例では、各サーバ１２１〜１２９の処理負荷の状態は、全て「未利用」であり同様である。ここでは、ワークキュー１１１のサーバ割当手段１１９が、サーバ名１，２，４，５，８，９であるサーバ１２１，１２２，１２４，１２５，１２８，１２９の６台を選択し、このうち、サーバ１２５をマスタとして用い、他の５台をワーカとして用いることに決定したとする。

割当手段１１９は、クライアント１１１からの要求（プログラム）を、選択した６台のサーバに送信し、マスタ１２５に管理開始命令、およびワーカとなる各サーバを通知する。図１１は、この状態を模式的に示す説明図である。

マスタ１２５のマスタ処理手段１５６は、クライアント１１１からのプログラムに含まれるデータ名Ａ，Ｂ，Ｅ，Ｆを指定して、そのデータを記憶しているサーバをデータ管理マスタ１１７に問い合わせる。そして、そのデータを記憶しているサーバの名称１，２，３，７をデータ管理マスタ１１７から受信する。

マスタ１２５のマスタ処理手段１５６は、そのサーバ名１，２，３，７のサーバと、ワークキュー１１２から通知されたサーバ名１，２，４，８，９のワーカについて、その各サーバの座標を座標管理マスタ１１４に問い合わせ、それらの各座標を座標管理マスタ１１４から受信する。本例では、マスタ処理手段１５６は、サーバ名１，２，３，７のサーバの座標として、それぞれ（４，２２）、（２，１５）、（１５，３）、（３２，２２）を得る。また、各ワーカの座標として、それぞれ（４，２２）、（２，１５）、（２２，５）、（３０，２０）、（２４，１０）を得る（図８参照）。

マスタ１２５のマスタ処理手段１５６は、これらの座標のうちの二つの座標の組み合わせを指定して、そのユークリッド距離を遅延推定手段１５７に計算させ、クライアントから要求された処理を実行するワーカを決定する。例えば、データＡを記憶するサーバ３の座標は（１５，３）と、利用可能な各ワーカの座標（４，２２）、（２，１５）、（２２，５）、（３０，２０）、（２４，１０）との距離は、それぞれ約２２，１８，７，２２，１１となる。よって、マスタ処理手段１５６は、データＡを入力データとするＭａｐ処理の実行をサーバ名４のワーカに実行させると決定する。ここでは、データＡのＭａｐ処理について例示したが、他のデータに関しても同様に、処理を実行するワーカを決定する。マスタ処理手段１５６は、各ワーカを決定したならば、そのワーカに処理を実行させる。図１２は、ワーカにマップ処理を実行させる状況を示す説明図である。図１２に示す個々の丸はサーバを表し、図６と同一の符号を付して表している。黒色で示したサーバ１２１，１２２，１２４，１２８，１２９はワーカに定められたサーバであり、白色で示したサーバ１２３，１２５，１２６，１２７は、ワーカ以外のサーバである（後述の図１３、図１４においても同様である）。図１２に示す例では、マスタ１２５のマスタ処理手段１５６は、ワーカ１２４にサーバ１２３からデータＡを取得させてＭａｐ処理を実行させ、同様に、ワーカ１２８にサーバ１２７からデータＥを取得させてＭａｐ処理を実行させる。また、ワーカ１２１，１２２に、それぞれ自装置に記憶しているデータＦ，Ｂを用いてＭａｐ処理を実行させる（図１２参照）。

クライアントから受け付けた分散データ処理プログラムでは、各Ｍａｐ処理の出力をＲｅｄｕｃｅ処理の入力として用いることを規定しているとする。その場合、マスタ１２５のマスタ処理手段１５６は、ワーカとして利用可能なサーバ名１，２，４，８，９のワーカの中から、Ｍａｐ処理を実行したサーバとワーカとの通信遅延を用いて集約サーバを決定する。ここではサーバ名９のワーカを集約サーバに決定したとする。マスタ１２５のマスタ処理手段１５６は、決定した集約サーバに各サーバからデータを集約させる。図１３は、集約ワーカにデータを集約させる状況を示す説明図である。図１３に示す例では、マスタ１２５のマスタ処理手段１５６は、ワーカ１２９を集約ワーカとして、Ｒｅｄｕｃｅ処理における集約をワーカ１２９に実行させる。

図１３では集約を一度に行う場合を示したが、ワーカ群において局所的に集約を行って、その局所的に集約されたデータをさらに集約してもよい。図１４は、このような段階的な集約の例を示す説明図である。マスタ１２５のマスタ処理手段１５６は、Ｍａｐ処理の出力データを有するワーカのうち、座標の近いワーカ同士をグループとして、そのグループ内のワーカに、グループ内における局所的なデータの集約を実行させてもよい。図１４に示す例では、ワーカ１２１，１２２におけるＭａｐ処理の出力データを集約するＲｅｄｕｃｅ処理をワーカ１２２に実行させる。そして、ワーカ１２４，１２８におけるＭａｐ処理の出力データ、およびワーカ１２２におけるＲｅｄｕｃｅ処理の出力を集約する上位のＲｅｄｕｃｅ処理をワーカ１２９に実行させる

なお、本実施例では、ワークキュー１１２が分散データ処理で用いるサーバを決定する際に、データの近傍性を考慮していないが、ワークキュー１１２が、データの近傍性を考慮してそれらのサーバを決定してもよい。また、ワークキュー１１２がマスタとするサーバを決定するときにも、データの近傍性を考慮してもよい。

以下、本発明の概要を説明する。図１５は、本発明の概要を示すブロック図である。本発明による分散データ処理システムは、データを処理する複数のデータ処理装置５（例えば、サーバ１２１〜１２９）と、座標管理装置１（例えば、座標管理マスタ１１４）と、割当装置２（例えば、ワークキュー１１２）とを備える。

座標管理装置１は、個々のデータ処理装置５がそれぞれ算出したデータ処理装置自身のコンピュータネットワークにおける座標を記憶する。

割当装置２は、クライアントから分散処理の要求を受信すると、各データ処理装置５の負荷状態に基づいて、要求された分散処理を実行するワーカと、ワーカが実行する分散処理を管理するマスタとをデータ処理装置５の中から決定する。割当装置２は、サーバ状態記憶手段３（例えば、サーバ状態格納部１１６）と、割当手段４（サーバ割当手段１１９）とを備える。サーバ状態記憶手段３は、各データ処理装置５の負荷状態を示す情報を記憶する。割当手段４は、クライアントから分散処理の要求を受信したときに、負荷状態を示す情報に基づいてワーカとなるデータ処理装置およびマスタとなるデータ処理装置を決定し、決定した各データ処理装置にクライアントからの要求を送信し、マスタに決定したデータ処理装置に分散処理の管理開始命令を送信する。

各データ処理装置５は、データ記憶手段６と、座標導出手段７と、遅延計算手段８と、マスタ処理手段９と、ワーカ処理手段１０とを備える。

データ記憶手段６（例えば、データ格納部１５１）は、処理対象となるデータを記憶する。座標導出手段７（例えば、座標算出手段１５３）は、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する。遅延計算手段８（例えば、遅延推定手段１５７）は、二つの座標が指定されると、その二つの座標間での通信遅延を計算する。

マスタ処理手段９（例えば、マスタ処理手段１５６）は、管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および各ワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延計算手段８に計算させ、通信遅延を用いてどのワーカに処理を実行させるかを決定し、そのワーカに処理の命令を送信する。ワーカ処理手段１０（例えば、ワーカ処理手段５１２）は、自装置以外のデータ処理装置のマスタ処理手段から処理の命令を受けたときに、その処理を実行する。

そのような構成により、処理負荷の増加や通信遅延の増大を防止し、輻輳の発生を抑えることができる。また、既存のシステムに容易に本発明による分散データ処理システムを導入することができる。

また、上記の実施形態には、各データ処理装置５のマスタ処理手段９が、通信遅延を遅延計算手段８に計算させたときに、通信遅延の計算対象としたデータ処理装置間で送受信されるデータ量と、通信遅延とから、データ処理装置間でのネットワーク負荷を計算し、ネットワーク負荷に基づいて処理を実行させるワーカを決定する構成が開示されている。

また、上記の実施形態には、各データ処理装置５のマスタ処理手段９が、処理対象となるデータを記憶するワーカ以外のデータ処理装置が存在する場合に、そのデータ処理装置と各ワーカの座標の組み合わせを指定して、遅延計算手段８に通信遅延を計算させ、通信遅延を用いて、どのワーカに処理を実行させるかを決定し、そのワーカにデータを取得して処理を実行する命令を送信する構成が開示されている。

また、上記の実施形態には、各データ処理装置５のマスタ処理手段９が、一のワーカが記憶するデータ量が閾値よりも多い場合、一のワーカと他のワーカの座標の組み合わせを指定して、遅延計算手段８に通信遅延を計算させ、通信遅延を用いて、一のワーカとともにデータに対する処理を実行するワーカを決定し、決定したワーカに一のワーカのデータの一部を取得して処理を実行する命令を送信するとともに、一のワーカに残りのデータを用いて処理を実行する命令を送信する構成が開示されている。

また、上記の実施形態には、各データ処理装置５のマスタ処理手段９が、ワーカが処理を行って生成したデータを取得すべきデータ処理装置が存在する場合に、そのデータ処理装置と各ワーカの座標の組み合わせを指定して、遅延計算手段８に通信遅延を計算させ、通信遅延を用いて、どのワーカに処理を実行させるかを決定し、そのワーカに処理を実行して生成したデータをデータ処理装置に送信する命令を送信する構成が開示されている。

本発明は、例えば、データセンタ内や企業内の分散ストレージや分散ファイルシステム、分散データベースに格納された多量のデータを抽出するデータウェアハウスシステム等の分散データ処理システムとして利用可能である。

本発明の分散データ処理システムの実施形態を示すブロック図である。 各サーバの構成例を示すブロック図である。 負荷状態を示す情報や座標を登録する動作の処理経過の例を示すフローチャートである。 クライアントが分散処理を要求し、マスタがワーカに分散処理を実行させる動作の流れを示すフローチャートである。 ステップＳ３０４におけるマスタ処理手段の処理経過の例を示す説明図である。 本発明の実施例を示す説明図である。 座標軸情報の例を示す説明図である。 座標格納部に記憶された情報の例を示す説明図である。 アドレス格納部に記憶された情報の例を示す説明図である。 サーバ状態格納部に記憶された情報の例を示す説明図である。 ワークキューが選択した６台のサーバにクライアンからの要求を送信する状態を模式的に示す説明図である。 ワーカにＭａｐ処理を実行させる状況を示す説明図である。 集約ワーカにデータを集約させる状況を示す説明図である。 段階的な集約の例を示す説明図である。 本発明の概要を示すブロック図である。

符号の説明

１１１ クライアント
１１２ ワークキュー
１１３ データ管理マスタ
１１４ 座標管理マスタ
１１５ 処理依頼手段
１１６ サーバ状態格納部
１１７ アドレス格納部
１１８ 座標格納部
１１９ サーバ割当手段
１２１〜１２９ サーバ
１５１ データ格納部
１５２ ワーカ処理手段
１５３ 座標算出手段
１５６ マスタ処理手段
１５７ 遅延推定手段

Claims (11)

1. データを処理する複数のデータ処理装置と、
   個々のデータ処理装置がそれぞれ算出したデータ処理装置自身のコンピュータネットワークにおける座標を記憶する座標管理装置と、
   クライアントから分散処理の要求を受信すると、各データ処理装置の負荷状態に基づいて、要求された分散処理を実行するワーカと、ワーカが実行する分散処理を管理するマスタとをデータ処理装置の中から決定する割当装置とを備え、
   割当装置は、
   各データ処理装置の負荷状態を示す情報を記憶するサーバ状態記憶手段と、
   クライアントから分散処理の要求を受信したときに、前記負荷状態を示す情報に基づいてワーカとなるデータ処理装置およびマスタとなるデータ処理装置を決定し、決定した各データ処理装置にクライアントからの要求を送信し、マスタに決定したデータ処理装置に分散処理の管理開始命令を送信する割当手段とを含み、
   各データ処理装置は、
   処理対象となるデータを記憶するデータ記憶手段と、
   コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出手段と、
   二つの座標が指定されると、当該二つの座標間での通信遅延を計算する遅延計算手段と、
   前記管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および各ワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延計算手段に計算させ、当該通信遅延を用いてどのワーカに処理を実行させるかを決定し、当該ワーカに処理の命令を送信するマスタ処理手段と、
   自装置以外のデータ処理装置のマスタ処理手段から処理の命令を受けたときに、当該処理を実行するワーカ処理手段とを含む
   ことを特徴とする分散データ処理システム。
2. 各データ処理装置のマスタ処理手段は、
   通信遅延を遅延計算手段に計算させたときに、通信遅延の計算対象としたデータ処理装置間で送受信されるデータ量と、前記通信遅延とから、前記データ処理装置間でのネットワーク負荷を計算し、ネットワーク負荷に基づいて処理を実行させるワーカを決定する
   請求項１に記載の分散データ処理システム。
3. 各データ処理装置のマスタ処理手段は、
   処理対象となるデータを記憶するワーカ以外のデータ処理装置が存在する場合に、当該データ処理装置と各ワーカの座標の組み合わせを指定して、遅延計算手段に通信遅延を計算させ、当該通信遅延を用いて、どのワーカに処理を実行させるかを決定し、当該ワーカに前記データを取得して処理を実行する命令を送信する
   請求項１または請求項２に記載の分散データ処理システム。
4. 各データ処理装置のマスタ処理手段は、
   一のワーカが記憶するデータ量が閾値よりも多い場合、前記一のワーカと他のワーカの座標の組み合わせを指定して、遅延計算手段に通信遅延を計算させ、当該通信遅延を用いて、前記一のワーカとともに前記データに対する処理を実行するワーカを決定し、決定したワーカに前記一のワーカのデータの一部を取得して処理を実行する命令を送信するとともに、前記一のワーカに残りのデータを用いて処理を実行する命令を送信する
   請求項１から請求項３のうちのいずれか１項に記載の分散データ処理システム。
5. 各データ処理装置のマスタ処理手段は、
   ワーカが処理を行って生成したデータを取得すべきデータ処理装置が存在する場合に、当該データ処理装置と各ワーカの座標の組み合わせを指定して、遅延計算手段に通信遅延を計算させ、当該通信遅延を用いて、どのワーカに処理を実行させるかを決定し、当該ワーカに処理を実行して生成したデータを前記データ処理装置に送信する命令を送信する
   請求項１から請求項４のうちのいずれか１項に記載の分散データ処理システム。
6. 分散データ処理システム内で処理を実行するデータ処理装置であって、
   処理対象となるデータを記憶するデータ記憶手段と、
   コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出手段と、
   二つの座標が指定されると、当該二つの座標間での通信遅延を計算する遅延計算手段と、
   クライアントが要求した分散処理の管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および処理の実行のために選択されたワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を遅延計算手段に計算させ、当該通信遅延を用いてどのワーカに処理を実行させるかを決定し、当該ワーカに処理の命令を送信するマスタ処理手段と、
   自装置以外のデータ処理装置から処理の命令を受けたときに、当該処理を実行するワーカ処理手段とを備える
   ことを特徴とするデータ処理装置。
7. マスタ処理手段は、通信遅延を遅延計算手段に計算させたときに、通信遅延の計算対象としたデータ処理装置間で送受信されるデータ量と、前記通信遅延とから、前記データ処理装置間でのネットワーク負荷を計算し、ネットワーク負荷に基づいて処理を実行させるワーカを決定する
   請求項６に記載のデータ処理装置。
8. データを処理する複数のデータ処理装置が、コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定し、
   割当装置が、クライアントから分散処理の要求を受信したときに、各データ処理装置の負荷状態を示す情報に基づいて、要求された分散処理を実行するワーカと、ワーカが実行する分散処理を管理するマスタとを決定し、決定した各データ処理装置にクライアントからの要求を送信し、マスタに決定したデータ処理装置に分散処理の管理開始命令を送信し、
   マスタとなるデータ処理装置が、前記管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各データ処理装置および各ワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を計算し、当該通信遅延を用いてどのワーカに処理を実行させるかを決定し、当該ワーカに処理の命令を送信し、
   ワーカとなるデータ処理装置が、マスタとなるデータ処理装置から前記処理の命令を受信したときに、当該処理を実行する
   ことを特徴とする分散データ処理方法。
9. マスタとなるデータ処理装置は、通信遅延を計算したときに、通信遅延の計算対象としたデータ処理装置間で送受信されるデータ量と、前記通信遅延とから、前記データ処理装置間でのネットワーク負荷を計算し、ネットワーク負荷に基づいて処理を実行させるワーカを決定する
   請求項８に記載の分散データ処理方法。
10. 分散データ処理システム内で処理を実行する各コンピュータに搭載される分散データ処理用プログラムであって、
    コンピュータに、
    コンピュータネットワークにおける自装置の位置を多次元の実数値で表した座標を、所定の装置との通信遅延の計測結果から決定する座標導出処理、
    クライアントが要求した分散処理の管理開始命令を受信したときに、クライアントが要求した分散処理に関連する各コンピュータおよび処理の実行のために選択されたワーカの座標から得られる二つの座標の組み合わせを指定して、指定した座標間における通信遅延を計算し、当該通信遅延を用いてどのワーカに処理を実行させるかを決定し、当該ワーカに処理の命令を送信するマスタ処理、および、
    自装置以外のコンピュータから処理の命令を受けたときに、当該処理を実行するワーカ処理
    を実行させるための分散データ処理用プログラム。
11. コンピュータに、
    マスタ処理で、通信遅延を計算したときに、通信遅延の計算対象としたコンピュータ間で送受信されるデータ量と、前記通信遅延とから、前記コンピュータ間でのネットワーク負荷を計算し、ネットワーク負荷に基づいて処理を実行させるワーカを決定させる
    請求項１０に記載の分散データ処理用プログラム。

Images