プロセッサと記憶装置とを備えて処理対象のデータとして予め分割されたデータブロックを前記記憶装置に格納し、前記プロセッサが前記データブロックを処理するタスクを並列して実行する複数の並列分散処理実行サーバと、前記複数の並列分散処理実行サーバを制御する管理計算機と、を備えた並列分散処理システムで、前記管理計算機が、第1の並列分散処理実行サーバのタスクに割り当てるデータブロックの送信元となる第2の並列分散処理実行サーバを選択する並列分散処理システムのデータ転送制御方法であって、前記管理計算機が、前記第1の並列分散処理実行サーバから前記タスクが完了したことを示す完了通知を受信する第1のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバのリソース使用量をそれぞれ収集する第2のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバが保持するデータブロックとタスクの状態を取得する第3のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバがそれぞれ保持するデータブロックの処理の進行状況と前記複数の並列分散処理実行サーバのリソース使用量に基づいて、前記第1の並列分散処理実行サーバへデータブロックを転送する第2の並列分散処理実行サーバを選択する第4のステップと、前記管理計算機が、前記選択した第2の並列分散処理実行サーバに対して前記データブロックを前記第1の並列分散処理実行サーバへ転送する指令を送信する第5のステップと、前記管理計算機が、前記第1の並列分散処理実行サーバに対して前記転送されたデータブロックを処理するタスクを実行する指令を送信する第6のステップと、を含むことを特徴とする並列分散処理システムのデータ転送制御方法。請求項1に記載の並列分散処理システムのデータ転送制御方法であって、前記第4のステップは、前記複数の並列分散処理実行サーバが保持するデータブロックのうち、タスクによる処理が完了した処理済みのデータブロックの比率を処理済みデータ率として演算する第7のステップと、前記処理済みデータ率が、予め設定した閾値未満の並列分散処理実行サーバを処理遅延サーバとして抽出する第8のステップと、前記抽出された処理遅延サーバから前記第2の並列分散処理実行サーバを選択する第9のステップと、を含むことを特徴とする並列分散処理システムのデータ転送制御方法。請求項2に記載の並列分散処理システムのデータ転送制御方法であって、前記第9のステップは、前記並列分散処理実行サーバの単位時間当たりの処理能力として前記タスクの実行多重度を取得し、前記実行多重度が最も少ない処理遅延サーバを前記第2の並列分散処理実行サーバとして選択することを特徴とする並列分散処理システムのデータ転送制御方法。請求項2に記載の並列分散処理システムのデータ転送制御方法であって、前記第9のステップは、前記並列分散処理実行サーバの単位時間当たりの処理能力として前記並列分散処理実行サーバのハードウェアスペックを示す値を取得し、前記ハードウェアスペックを示す値が最も低い処理遅延サーバを前記第2の並列分散処理実行サーバとして選択することを特徴とする並列分散処理システムのデータ転送制御方法。請求項2に記載の並列分散処理システムのデータ転送制御方法であって、前記第9のステップは、前記第1の並列分散処理実行サーバの空きI/Oリソース量を求める第10のステップと、前記処理遅延サーバの空きI/Oリソース量を求める第11のステップと、前記第1の並列分散処理実行サーバの空きI/Oリソース量と、前記処理遅延サーバの空きI/Oリソース量の差が最も小さい前記処理遅延サーバを第2の並列分散処理実行サーバを選択する第12のステップと、を含むことを特徴とする並列分散処理システムのデータ転送制御方法。請求項5に記載の並列分散処理システムのデータ転送制御方法であって、前記第10のステップは、前記第1の並列分散処理実行サーバのネットワークの空きI/Oリソース量と、ディスクI/Oの空きリソース量とを求め、前記第11のステップは、前記処理遅延サーバのネットワークの空きI/Oリソース量と、ディスクI/Oの空きリソース量とを求め、前記第12のステップは、前記第1の並列分散処理実行サーバのネットワークI/Oの空きリソース量と、処理遅延サーバのネットワークI/Oの空きリソース量との差の絶対値を第1の絶対値として求め、第1の並列分散処理実行サーバのディスクI/Oの空きリソース量と、処理遅延サーバのディスクI/Oの空きリソース量との差の絶対値を第2の絶対値として求め、前記第1の絶対値と第2の絶対値のうち大きい方の値を前記処理遅延サーバの絶対値として選択し、当該処理遅延サーバの絶対値のうち最も小さい値の処理遅延サーバを前記第2の並列分散処理実行サーバとして選択することを特徴とする並列分散処理システムのデータ転送制御方法。請求項2に記載の並列分散処理システムのデータ転送制御方法であって、前記第9のステップは、前記第1の並列分散処理実行サーバの空きI/Oリソース量を求める第10のステップと、前記処理遅延サーバの空きI/Oリソース量を求める第11のステップと、前記第1の並列分散処理実行サーバの空きI/Oリソース量よりも多い空きI/Oリソース量を有する処理遅延サーバのうち、前記空きI/Oリソース量の差が最も小さい前記処理遅延サーバを第2の並列分散処理実行サーバを選択する第12のステップと、を含むことを特徴とする並列分散処理システムのデータ転送制御方法。請求項1に記載の並列分散処理システムのデータ転送制御方法であって、前記並列分散処理実行サーバは、物理サーバ上で実行される仮想化部が提供する仮想サーバとして実行され、前記第4のステップは、前記第1の並列分散処理実行サーバを実行する物理サーバと同一の物理サーバで実行される仮想サーバを優先して前記第2の並列分散処理実行サーバとして選択することを特徴とする並列分散処理システムのデータ転送制御方法。プロセッサと記憶装置とを備えて処理対象のデータとして予め分割されたデータブロックを前記記憶装置に格納し、前記プロセッサが前記データブロックを処理するタスクを並列して実行する複数の並列分散処理実行サーバと、前記複数の並列分散処理実行サーバを制御する管理計算機と、を備えた並列分散処理システムであって、前記管理計算機は、前記複数の並列分散処理実行サーバのデータブロックに対すタスクを制御する並列分散処理制御部と、前記複数の並列分散処理実行サーバのうち、タスクにデータブロックを割り当てる第1の並列分散処理実行サーバ対して、前記データブロックの送信元となる第2の並列分散処理実行サーバを選択するデータ転送制御部と、前記複数の並列分散処理実行サーバのリソース使用量をそれぞれ管理するリソース使用量管理部と、を有する並列分散処理システムであって、前記並列分散処理制御部は、前記第1の並列分散処理実行サーバから前記タスクが完了したことを示す完了通知を受信し、前記複数の並列分散処理実行サーバが保持するデータブロックとタスクの状態を取得し、前記データ転送制御部は、前記複数の並列分散処理実行サーバがそれぞれ保持するデータブロックの処理の進行状況と前記複数の並列分散処理実行サーバのリソース使用量に基づいて、前記第1の並列分散処理実行サーバへデータブロックを転送する第2の並列分散処理実行サーバを選択し、前記選択した第2の並列分散処理実行サーバに対して前記データブロックを前記第1の並列分散処理実行サーバへ転送する指令を送信し、前記並列分散処理制御部は、前記第1の並列分散処理実行サーバに対して前記転送されたデータブロックを処理するタスクを実行する指令を送信することを特徴とする並列分散処理システム。請求項9に記載の並列分散処理システムであって、前記データ転送制御部は、前記複数の並列分散処理実行サーバが保持するデータブロックのうち、タスクによる処理が完了した処理済みのデータブロックの比率を処理済みデータ率として演算し、前記処理済みデータ率が、予め設定した閾値未満の並列分散処理実行サーバを処理遅延サーバとして抽出し、前記抽出された処理遅延サーバから前記第2の並列分散処理実行サーバを選択することを特徴とする並列分散処理システム。請求項10に記載の並列分散処理システムであって、前記データ転送制御部は、前記第1の並列分散処理実行サーバの空きI/Oリソース量を求め、前記処理遅延サーバの空きI/Oリソース量を求めて、前記第1の並列分散処理実行サーバの空きI/Oリソース量と、前記処理遅延サーバの空きI/Oリソース量の差が最も小さい前記処理遅延サーバを第2の並列分散処理実行サーバを選択することを特徴とする並列分散処理システム。請求項9に記載の並列分散処理システムであって、前記並列分散処理実行サーバは、物理サーバ上で実行される仮想化部が提供する仮想サーバとして実行され、前記データ転送制御部は、前記第1の並列分散処理実行サーバを実行する物理サーバと同一の物理サーバ上の仮想サーバを優先して前記第2の並列分散処理実行サーバとして選択することを特徴とする並列分散処理システム。複数の並列分散処理実行サーバのうち、第1の並列分散処理実行サーバのタスクに割り当てるデータブロックの送信元となる第2の並列分散処理実行サーバを選択するプログラムを格納した記憶媒体であって、前記第1の並列分散処理実行サーバから前記タスクが完了したことを示す完了通知を受信する第1の手順と、前記複数の並列分散処理実行サーバのリソース使用量をそれぞれ収集する第2の手順と、前記複数の並列分散処理実行サーバが保持するデータブロックとタスクの状態を取得する第3の手順と、前記複数の並列分散処理実行サーバがそれぞれ保持するデータブロックの処理の進行状況と前記複数の並列分散処理実行サーバのリソース使用量に基づいて、前記第1の並列分散処理実行サーバへデータブロックを転送する第2の並列分散処理実行サーバを選択する第4の手順と、前記選択した第2の並列分散処理実行サーバに対して前記データブロックを前記第1の並列分散処理実行サーバへ転送する指令を送信する第5の手順と、前記第1の並列分散処理実行サーバに対して前記転送されたデータブロックを処理するタスクを実行する指令を送信する第6の手順と、を計算機に実行させるプログラムを格納した非一時的な計算機読み取り可能な記憶媒体。非特許文献1に記載の技術を用いることにより、出来るだけ自サーバに保存されているデータを処理できるため、データ転送の発生を抑制できる。さらに、他サーバに保存されているデータの処理を割り当てる場合には、ネットワーク的に最も近いサーバに保存されているデータの処理を割り当てるため、データ転送コストを抑制できる。しかしながら、上記非特許文献1の技術では、優先度の高い既存のシステムと、並列分散処理システムが共存して稼働する状況下では、必ずしもデータ転送コストを抑制できるとは限らない。例えば、既存のシステムと共存して稼働する並列分散処理システムでは、各サーバの単位時間当たりの処理能力に差が生まれやすく、各サーバの未処理データ量に偏りが生じる可能性がある。そのため、あるサーバ上のタスクに処理を割り当てる際に、最も近いサーバの未処理データが少ない場合、転送元サーバのタスクでは、自サーバの未処理データが早い段階で無くなってしまう。よって、他サーバからデータを取得する回数が多くなり、データ転送が多発してしまう。並列分散処理システムにおいて、データ転送が多発するのを抑制する方法の1つとして、特許文献1が知られている。特許文献1に記載の技術を用いることにより、他サーバに保存されているデータの処理を自サーバのタスクに割り当てる場合に、各サーバの割り当て済みデータの割合を算出し、どのサーバに保存されているデータの処理を割り当てるのかを決定する。この決定の際には、未割り当てデータの少ないサーバから自サーバへデータを転送して取得する可能性が低くなる。よって、各サーバが可能な限り自サーバに保存されているデータを処理できるようになり、データ転送回数の削減するものである。しかしながら、優先度の高い既存のシステムに、並列分散処理システムを共存させて稼動する状況下においては、前記特許文献1等の従来技術では必ずしも十分とは言えない。例えば、並列分散処理システムは、既存のシステムの負荷変動やリソースの変動に合わせて実行多重度を変更する必要がある。このため、既存のシステムの負荷が増加した場合、優先度が低い並列分散処理実行システムでは実行多重度を減らさなければならない。このため、既に処理を割り当てられていたタスクが、実行多重度の減少により、処理を実行できなくなる可能性がある。また、各サーバの実行多重度が異なるため、各サーバで単位時間あたりに処理できるデータ量も異なる。そのため、割り当て済みデータの割合が小さい場合でも、単位時間当たりに処理可能なデータ量が多く、他のサーバより未処理データを短時間で処理できる可能性がある。さらに、並列分散処理システムが既存システムと共存する状況下では、並列分散処理システムがデータ転送に利用できるI/Oリソース量もサーバ毎に異なる。そのため、データの転送先サーバの空きI/Oリソース量が多く、一度に大量のデータを転送可能な場合でも、転送元サーバに利用可能なI/Oリソース量の小さいサーバを選んだ場合、空きI/Oリソース量の小さい方に合わせてデータを転送する必要があり、転送先サーバの空きI/Oリソースを最大限利用できない。これにより、データ転送時間が長くなり、並列分散処理システム全体の処理時間に影響を与える可能性がある。従って、優先度の高い既存システムに、並列分散処理システムを共存させて稼働する環境下のように、並列分散処理システムに割り当て可能な計算機資源が変動して、並列分散処理システムの実効多重度が変化した場合であっても、以降の並列処理をデータ転送の回数、使用するリソース量の点で効率的に実行できるように、各サーバにデータを割り当てることが課題になる。本発明は上記問題点に鑑みてなされたもので、並列分散処理システムに割当可能な計算機資源が変化して、並列分散処理の実行多重度が変化する状況で、データ転送回数を削減することを目的とする。本発明は、プロセッサと記憶装置とを備えて処理対象のデータとして予め分割されたデータブロックを前記記憶装置に格納し、前記プロセッサが前記データブロックを処理するタスクを並列して実行する複数の並列分散処理実行サーバと、前記複数の並列分散処理実行サーバを制御する管理計算機と、を備えた並列分散処理システムで、前記管理計算機が、第1の並列分散処理実行サーバのタスクに割り当てるデータブロックの送信元となる第2の並列分散処理実行サーバを選択する並列分散処理システムのデータ転送制御方法であって、前記管理計算機が、前記第1の並列分散処理実行サーバから前記タスクが完了したことを示す完了通知を受信する第1のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバのリソース使用量をそれぞれ収集する第2のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバが保持するデータブロックとタスクの状態を取得する第3のステップと、前記管理計算機が、前記複数の並列分散処理実行サーバがそれぞれ保持するデータブロックの処理の進行状況と前記複数の並列分散処理実行サーバのリソース使用量に基づいて、前記第1の並列分散処理実行サーバへデータブロックを転送する第2の並列分散処理実行サーバを選択する第4のステップと、前記管理計算機が、前記選択した第2の並列分散処理実行サーバに対して前記データブロックを前記第1の並列分散処理実行サーバへ転送する指令を送信する第5のステップと、前記管理計算機が、前記第1の並列分散処理実行サーバに対して前記転送されたデータブロックを処理するタスクを実行する指令を送信する第6のステップと、を含む。本発明によれば、並列分散処理の実行中に利用可能なコンピュータリソースが変化し、並列分散処理の実行多重度が変化する並列分散処理システムにおいて、データ転送回数を削減し、さらに、データ転送時間を短縮することにより、並列分散処理全体を効率的に実行可能なデータ転送制御方法及び装置を提供することができる。本発明を実施するための形態以下、実施形態というについて、適宜図面を参照しながら詳細に説明する。<第1実施形態>まず、図1~図13を参照して、本発明の第1実施形態に係わる説明を行なう。<システム構成>図1A、図1Bは、本発明の第1の実施形態を示し、計算機システムの構成の一例を示すブロック図である。計算機システム10は、クライアント装置110、複数の並列分散処理実行サーバ120-1~120-n、並列分散処理制御サーバ130、データ転送制御サーバ140、リソース使用量管理サーバ150がネットワーク100を介して互いに接続されている。なお、ネットワーク100は、LANLocalAreaNetworkや、WANWideAreaNetworkや、インターネットなどのグローバルネットワークである。また、ネットワーク100は、複数のネットワーク100に分けられても良い。また、並列分散処理実行サーバ120-1~120-nの総称は並列分散処理実行サーバ120とする。<クライアント装置>クライアント装置110は、ネットワークインタフェース111、CPU112、主記憶装置113、二次記憶装置114、及び、これらを相互に接続するバスまたはインターコネクト115を備える計算機である。ネットワークインタフェース111は、クライアント装置110がネットワーク100に接続するためのインタフェースである。CPU112は、主記憶装置113に記憶されているプログラムを実行することによってクライアント装置110の所定の機能を実現する演算処理装置である。主記憶装置113は、CPU112によって実行されるプログラム、及び、プログラムの実行に必要なデータを記憶するRAM等の記憶装置である。プログラムとは、例えば、図示しないOSや、クライアント処理部1131の機能を実現するためのプログラムである。二次記憶装置114は、クライアント装置110の所定の機能を実現するために必要なプログラム、及び、データ等を格納するハードディスク装置等の不揮発性記憶媒体である。なお、二次記憶装置114は、ハードディスク装置等の磁気的記憶媒体に限定されるものではなく、フラッシュメモリ等の不揮発性の半導体記憶媒体であってもよい。<並列分散処理実行サーバ>並列分散処理実行サーバ120は、ネットワークインタフェース121、CPU122、主記憶装置123、二次記憶装置124、及び、これらを相互に接続するバスまたはインターコネクト125を備える計算機である。なお、並列分散処理実行サーバ120-1~120-nの構成は同様とする。ネットワークインタフェース121は、並列分散処理実行サーバ120がネットワーク100に接続するためのインタフェースである。CPU122は主記憶装置133に記憶されているプログラムを実行することによって並列分散処理実行サーバの所定の機能を実現する演算処理装置である。主記憶装置123は、CPU122によって実行されるプログラム、及び、プログラムの実行に必要なデータを記憶するRAM等の記憶装置である。プログラムとは、例えば、不図示のOS、及び、ユーザ定義処理実行部1231、データ管理部1232の機能を実現するためのプログラムである。二次記憶装置124は、並列分散処理実行サーバ120が所定の機能を実現するために必要なプログラム、及び、入力データ1241、出力データ1242、データ管理テーブル1243等のデータを格納するハードディスク装置等の不揮発性記憶媒体である。なお、二次記憶装置124は、ハードディスク装置等の磁気的記憶媒体に限定されるものではなく、フラッシュメモリなどの不揮発性の半導体記憶媒体であってもよい。入力データ1241、出力データ1242について補足する。入力データ1241は、規定のサイズに分割された複数のデータブロックからなる論理的なデータであり、名前と構成するデータブロックを識別するための情報とを含む。入力データ1241のデータブロックを識別するための情報とは、例えば、データブロックを格納する並列分散処理実行サーバ120のアドレス情報とデータブロックの名前である。データの実体はデータブロックとして並列分散処理実行サーバ120に格納される。出力データ1242は、前述の並列分散処理によって出力されるデータである。ユーザ定義処理実行部1231は、割り当てられたタスクを実行する。データ管理部1232は、タスクに対するデータブロック入力データ1241の割り当てを管理する。<並列分散処理制御サーバ>並列分散処理制御サーバ130は、各並列分散処理実行サーバ120に処理を割り当て、並列分散処理全体の実行を制御するためのサーバであり、並列分散処理ネットワークインタフェース131、CPU132、主記憶装置133、二次記憶装置134、及び、これらを相互に接続するバス135を備える。ネットワークインタフェース131は、並列分散処理制御サーバ130がネットワーク100に接続するためのインタフェースである。CPU132は、主記憶装置133に記憶されているプログラムを実行することによって並列分散処理制御サーバの所定の機能を実現する演算処理装置である。主記憶装置133は、CPU132によって実行されるプログラム、及び、プログラムの実行に必要なデータを記憶するRAM等の記憶装置である。プログラムとは、例えば、不図示のOS、及び、処理割当制御部1331、データ情報管理部1332、処理実行サーバ管理部1333、タスク管理部1334の機能を実現するためのプログラムである。二次記憶装置134は、並列分散処理制御サーバ130が所定の機能を実現するために必要なプログラム、及び、データ情報管理テーブル300、処理実行サーバ管理テーブル500、タスク管理テーブル600等のデータを格納するハードディスク装置等の不揮発性記憶媒体である。なお、二次記憶装置134は、ハードディスク装置等の磁気的記憶媒体に限定されるものではなく、フラッシュメモリ等の不揮発性の半導体記憶媒体であってもよい。なお、各処理割当制御部1331、データ情報管理部1332、処理実行サーバ管理部1333、タスク管理部1334の機能については、処理の説明において後述する。<データ転送制御サーバ>データ転送制御サーバ140は、処理を割り当てる並列分散処理実行サーバ120とは別の並列分散処理実行サーバ120に格納されているデータブロックの処理を割り当てる場合に、処理するデータブロックを格納している並列分散処理実行サーバの選択を行うサーバであり、ネットワークインタフェース141、CPU142、主記憶装置143、二次記憶装置144、及び、これらを相互に接続するバス145を備える。ネットワークインタフェース141は、データ転送制御サーバ140がネットワーク100に接続するためのインタフェースである。CPU142は、主記憶装置143に記憶されているプログラムを実行することによってデータ転送制御サーバ140の所定の機能を実現する演算処理装置である。主記憶装置143は、CPU142によって実行されるプログラム、及び、プログラムの実行に必要なデータを記憶するRAM等の記憶装置である。プログラムとは、例えば、図示しないOS、及び、処理遅延サーバ抽出処理部1431、空きI/Oリソース比較処理部1432、処理状況管理部1433の機能を実現するためのプログラムである。二次記憶装置144は、データ転送制御サーバ140が所定の機能を実現するために必要なプログラム、及び、処理状況管理テーブル800、処理遅延閾値管理テーブル1000等のデータを格納するハードディスク装置等の不揮発性記憶媒体である。なお、二次記憶装置144は、ハードディスク装置等の磁気的記憶媒体に限定されるものではなく、フラッシュメモリ等の不揮発性の半導体記憶媒体であってもよい。なお、処理遅延サーバ抽出処理部1431、空きI/Oリソース量比較処理部1432、処理状況管理部1433の各機能については、処理の説明において後述する。<リソース使用量管理サーバ>リソース使用量管理サーバ150は、各サーバのI/Oリソース使用量を管理するためのサーバであり、ネットワークインタフェース151、CPU152、主記憶装置153、二次記憶装置154、及び、これらを相互に接続するバス155を備える。ネットワークインタフェース151は、リソース使用量管理サーバ150がネットワーク100に接続するためのインタフェースである。CPU152は、主記憶装置153に記憶されているプログラムを実行することによってリソース使用量管理サーバ150の所定の機能を実現する演算処理装置である。主記憶装置153は、CPU152によって実行されるプログラム、及び、プログラムの実行に必要なデータを記憶するRAM等の記憶装置である。プログラムとは、例えば、図示しないOS、及び、リソース使用量管理部1531、リソース使用量監視部1532の機能を実現するためのプログラムである。二次記憶装置154は、リソース使用量管理サーバ150が所定の機能を実現するために必要なプログラム、及びリソース使用量管理テーブル1200等のデータを格納するハードディスク装置等の不揮発性記憶媒体である。なお、二次記憶装置154は、ハードディスク装置等の磁気的記憶媒体に限定されるものではなく、フラッシュメモリ等の不揮発性の半導体記憶媒体であってもよい。なお、リソース使用量管理部1531、リソース使用量監視部1532の各機能については、処理の説明において後述する。以上、各装置のハードウェア構成、及び、ソフトウェア構成について説明してきたが、クライアント装置110、並列分散処理実行サーバ120、並列分散処理制御サーバ130、データ転送制御サーバ140、及び、リソース使用量管理サーバ150の構成は、図1A、図1Bに示す構成に限定されるものではない。例えば、並列分散処理制御サーバ130、データ転送制御サーバ140、及び、リソース使用量管理サーバは、クライアント装置110、または、並列分散処理実行サーバ120の何れかのサーバ上で稼動する構成でも構わない。さらに、図1A、図1Bでは、並列分散処理制御サーバ130、データ転送制御サーバ140、及び、リソース使用量管理サーバ150は異なるサーバ上で実行される構成を示したが、これらの全部、または一部が同一のサーバ上で実行されても構わない。ひとつのサーバで、並列分散処理制御サーバ130、データ転送制御サーバ140及びリソース使用量管理サーバ150を実行する場合には、各サーバのプログラムは、並列分散処理制御部、データ転送制御部及びリソース使用量管理部として機能すればよい。また、並列分散処理制御部、データ転送制御部及びリソース使用量管理部のプログラムは、プログラム配布サーバや非一時的計算機読み取り可能な記憶媒体によって同一のサーバにインストールすることができる。次に、図1A、図1Bを参照しつつ、図2A~図13に沿って本発明の第1の実施形態を示し、処理の説明を行う。<全体処理>図2A、図2Bは、本発明の第1の実施形態を示し、並列分散処理に関する全体的な処理の流れを示すフローチャートである。まず、クライアント装置110から並列分散処理制御サーバ130へ、処理対象データ入力データ1241のデータロードの要求が送信される。データロードの要求を受信した並列分散処理制御サーバ130は、ロード対象データを規定のサイズのデータブロック入力データ1241に分割し、複数の並列分散処理実行サーバ120に分散してロードするS101。次に、クライアント装置110から並列分散処理制御サーバ130へ、並列分散処理実行要求が送信される。並列分散処理実行要求を受信した並列分散処理制御サーバ130の処理割当制御部1331は、各並列分散処理実行サーバ120に、ロードされたデータブロックの処理実行要求を送信し、データ情報管理部1332がデータ情報管理テーブル300の処理対象となるデータブロックの割当状態303を割当済みに更新するS102。なお、処理割当制御部1331が送信する処理実行要求は、処理対象のデータブロックのデータブロックIDと、各データブロックの処理を実行するタスクのタスクIDとを含む。ここでタスクは、処理対象のデータブロックを入力データ1241として所定の処理を実行するプログラムである。続いて、処理実行要求を受信した並列分散処理実行サーバ120のユーザ定義処理実行部1231は、各タスクにおいて処理実行要求で指定されたデータブロックを入力データ1241として所定の処理を実行するS103。すなわち、並列分散処理実行サーバ120のユーザ定義処理実行部1231は、処理実行要求で指定されたタスクを起動し、各タスクに対して処理実行要求で指定されたデータブロックを入力データ1241として割り当てて処理を実行させる。並列分散処理実行サーバ120は、割り当てられたタスクに入力データ1241として指定されたデータブロックの処理が完了すると、当該並列分散処理実行サーバ120から並列分散処理制御サーバ130へ、処理完了通知として処理の完了したタスクのタスクIDを送信するS104。そして、処理完了通知を受信した並列分散処理制御サーバ130のデータ情報管理部1332が、データ情報管理テーブル300の該当するデータブロックID302の処理状態304を処理済みに更新するS105。続いて、並列分散処理制御サーバ130の処理割当制御部1331がデータ管理情報テーブル300を参照し、全てのデータブロックの処理状態304が処理済みであるか否かを判定するS106。ステップS106の判定の結果、全てのデータブロックが処理済みである場合S106→Yesは、並列分散処理制御サーバ130の処理割当制御部1331が、クライアント装置110に並列分散処理完了通知を送信し、計算機システム10は並列分散処理を終了する。ステップS106判定の結果、何れかのデータブロックが処理済みではない場合S106→Noは、並列分散処理制御サーバ130は、ステップS116で図2Bの割り当て処理S116を行った後に、ステップS103に戻って上記処理を繰り返す。図2Bの割当処理では、並列分散処理制御サーバ130が、データ管理情報テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120に、割当状態403が未割当のデータブロックが存在するか否かを判定するS107。ステップS107の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120に未割当のデータブロックが存在する場合S107→Yesは、並列分散処理制御サーバ130の処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120に存在する未割当のデータブロックの中から、任意に一つのデータブロックを選択するS108。ステップS107の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120に未割当のデータブロックが存在しない場合S107→Noは、ステップS109に進む。並列分散処理制御サーバ130の処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストをデータ情報管理テーブル300から生成してデータ転送制御サーバ140に送信するS109。次に、データ転送制御サーバ140の処理遅延サーバ抽出処理部1431が、受信した未割当のデータブロックを持つサーバ名リストに含まれる並列分散処理実行サーバ120の中から、処理遅延サーバを抽出する。ここで、処理遅延サーバは、処理の進捗が遅延している並列分散処理実行サーバ120を指す。本実施形態では、後述するように、並列分散処理実行サーバ120が保持するデータブロックの総数と、既にタスクの実行によって処理が完了した処理済みのデータブロック数の比率を処理済みデータ率として求め、処理済みデータ率が閾値1001未満の並列分散処理実行サーバ120を、処理の進捗が遅延している処理遅延サーバとして抽出する。データ転送制御サーバ140の処理遅延サーバ抽出処理部1431は、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、抽出したサーバのサーバ名リストと、をリソース使用量管理サーバ150へ送信し、各並列分散処理実行サーバ120の空きI/Oリソース量を要求するS110。続いて、リソース使用量管理サーバ150のリソース使用量管理部1531が、リソース使用量管理テーブル1200を参照し、処理完了通知を送信した並列分散処理実行サーバ120と、サーバ名リストに含まれる並列分散処理実行サーバ120の空きI/Oリソース量を取得してデータ転送制御サーバ140へ送信するS111。なお、空きI/Oリソース量は、ネットワークI/Oの空きI/Oリソース量と、ディスクI/Oの空きI/Oリソース量とを含む。ここで、ネットワークI/Oの空きI/Oリソース量は、並列分散処理実行サーバ120のネットワークインターフェース121で利用可能なデータ転送速度Gbit/secを比率で示す。ネットワークインターフェース121で利用可能なデータ転送速度は、ネットワークインターフェース121で転送可能なデータ転送速度の理論値から、現在使用している帯域転送速度を差し引いた値を、データ転送速度の理論値で除した比率で表した値である。すなわち、ネットワークI/Oの空きI/Oリソース量=理論値-現在使用帯域÷理論値である。現在使用している帯域は、所定時間例えば、1分間のデータ転送速度の平均値などを用いることができる。また、データ転送速度の理論値に代わって、ネットワークインターフェース121のリンク速度の実効値または理論値に所定の比率を乗じた値を用いてもよい。ディスクI/Oの空きI/Oリソース量は、並列分散処理実行サーバ120の二次記憶装置124で利用可能なデータ転送速度MByte/secを比率で示す。二次記憶装置124で利用可能なデータ転送速度は、二次記憶装置124で転送可能なデータ転送速度の理論値から、現在使用している帯域転送速度を差し引いた値を、理論値で除した比率である。すなわち、ディスクI/Oの空きI/Oリソース量=理論値-現在使用帯域÷理論値である。現在使用している帯域は、所定時間例えば、1分間のデータ転送速度の平均値などを用いることができる。また、二次記憶装置124のデータ転送速度の実効値または理論値に所定の比率を乗じた値を用いてもよい。そして、データ転送制御サーバ140の空きI/Oリソース量比較処理部1432が、処理完了通知を送信した並列分散処理実行サーバ120の空きI/Oリソース量と、各処理遅延サーバの空きI/Oリソース量を比較し、最も空きI/Oリソース量の差が小さい処理遅延サーバを選択し、当該選択した処理遅延サーバの名称を並列分散処理制御サーバ130へ送信するS112。次に、並列分散処理制御サーバ130の処理割当制御部1331が、データ転送制御サーバ140から受信した処理遅延サーバ名に該当する並列分散処理実行サーバ120に存在する未割当のデータブロックの中から、任意に一つのデータブロックを選択するS113。すなわち、処理割当制御部1331は、データ情報管理テーブル300を参照して、データ転送制御サーバ140から受信した処理遅延サーバ名と一致するデータ配置サーバ名302のうち、割当状態303が未割当のデータブロックIDから、ひとつのデータブロックIDを任意に選択する。この選択はデータブロックIDの昇順やラウンドロビンなど周知または公知の手法で選択すればよい。そして、並列分散処理制御サーバ130のデータ情報管理部1332が、データ情報管理テーブル300において選択したデータブロックの割当状態303を割当済みに更新する。また、処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120に、上記割当済みに更新したデータブロックの処理要求を送信し、計算機システム10は処理を図2AのステップS103に戻すS114。なお、並列分散処理実行サーバ120に対して送信されるデータブロックの処理要求は、選択したデータブロックのデータブロックIDと、前記データブロックの処理を実行するタスクIDとを含む。なお、ステップS107~S114の詳細は後述する。<データ情報管理テーブル>図3は、本発明の第1の実施形態を示し、データ情報管理テーブル300の一例を示す図である。データ情報管理テーブル300は、入力データ1241を規定サイズで分割した複数のデータブロックで構成され、各データブロックを管理するための属性情報として、各データブロックを識別するためのデータブロックID301と、データ配置サーバ名302と、データブロックの処理が既に割り当てられたか否かを示す割当状態303と、データブロックの処理が終了したか否かを示す処理状態304からひとつのエントリが構成される。割当状態303の欄には、割当済み、未割当等の情報が格納され、処理状態304の欄には、処理済み、未処理等の情報が格納される。データ配置サーバ名302には、並列分散処理実行サーバ120の名称または識別子が格納される。図示の例では、サーバAが図1Aの並列分散処理実行サーバ120-1に相当し、サーバBが並列分散処理実行サーバ120-1に相当するものとする。<S107~S114の詳細並列分散処理実行サーバへの処理の割り当て>図4A、図4Bは、図2AのS116の割当処理を示し、図2BのステップS107からS114の並列分散処理実行サーバ120への処理の割り当てに関する詳細な手順を示すフローチャートである。なお、図4A、図4Bの処理は、並列分散処理制御サーバ130上の処理割当制御部1331、データ情報管理部1332、タスク管理部1334が行なう処理である。まず、並列分散処理実行サーバ120から処理完了通知としてタスクIDを受信した処理割当制御部1331が、タスク管理部1334にタスク管理テーブル600の更新を要求する。また、処理割当制御部1331が、データ情報管理部1332に未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを要求するS201。次に、タスク管理テーブル600の更新要求を受信したタスク管理部1334が、タスク管理テーブル600で該当するタスクの実行状態603を待ちに更新し、処理データブロックID604をNULL値に更新するS202。続いて、データ情報管理部1332が、データ情報管理テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120の処理済みデータ率を算出して、データ転送制御サーバ140に処理済みデータ率と並列分散処理実行サーバ120の名称を送信する。処理済みデータ率を受信したデータ転送制御サーバ140は、受信したサーバ名の処理状況管理テーブル800の処理済みデータ率802を更新するS203。なお、処理済みデータ率の算出方法は、データ情報管理部1332が、並列分散処理実行サーバ120に格納されているデータブロック数に対する、データ情報管理テーブル300の処理状態304が処理済みのデータブロック数の比率を求めることで行われる。すなわち、処理済みデータ率は、並列分散処理実行サーバ120が格納している全データブロックに対して、現時点までに処理が完了したデータブロックの割合であり、以下の式1で算出されるものである。処理済みデータ率=格納されている処理済みのデータブロック数÷格納されているデータブロック数1そして、データ情報管理部1332が、データ情報管理テーブル300を参照し、割当状態303が未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを作成し、処理割当制御部1331に送信するS204。未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを受信した処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120がリストに存在するか否かを判定するS205。ステップS205の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120がリストに存在する場合S205→Yesは、処理割当制御部1331が、データ情報管理テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120に保存されている未割当のデータブロックの中から、任意に一つのデータブロックを選択するS206。ステップS205の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120がリストに存在しない場合S205→Noは図4Bの処理に進む。処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120のタスクが次に処理するデータブロックを持つ並列分散処理実行サーバ120のサーバ名をデータ転送制御サーバ140に要求するS207。なお、サーバ名の要求は、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストとを含む。次に、データ転送制御サーバ140が、処理完了通知を送信した並列分散処理実行サーバが次に処理するデータブロックを格納している並列分散処理実行サーバ120を抽出し、並列分散処理制御サーバ130の処理割当制御部1331に送信するS208。このステップS208の処理の詳細については後述する。続いて、処理割当制御部1331が、データ転送制御サーバ140から受信したサーバ名に該当する並列分散処理実行サーバ120に格納されている未割当のデータブロックの中から、任意に一つのデータブロックを選択するS209。そして、処理割当制御部1331は、上記受信したサーバ名に該当する並列分散処理実行サーバ120に、処理完了通知を送信した並列分散処理実行サーバ120へ、選択したデータブロックの転送を実行するよう要求するS210。なお、データブロックの転送要求は、選択したデータブロックのデータブロックIDと、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名とを含む。そして、処理割当制御部1331は、データ情報管理部1332にデータ情報管理テーブル300の更新を要求し、タスク管理部1334にタスク管理テーブル600の更新を要求するS211。なお、データ情報管理テーブル300の更新要求は、選択したデータブロックのデータブロックID301とデータ配置サーバ名302を含み、タスク管理テーブル600の更新要求は、並列分散処理実行サーバ120から処理完了通知として受信したタスクID601と、選択したデータブロックの処理データブロックID604を含む。データ情報管理部1332は、受信した更新要求に含まれるデータブロックID301に該当するデータブロックの割当状態303を割当済みに更新し、タスク管理部1334は、受信したタスクIDに該当するタスク管理テーブル600のタスクの実行状態603を実行中に更新し、受信したデータブロックIDで処理対象データブロックID604を更新するS212。そして、処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120に、選択したデータブロックの処理実行要求を送信するS213。なお、処理実行要求は、選択したデータブロックのデータブロックIDと、処理を割り当てるタスクIDとを含む。<処理実行サーバ管理テーブル>図5は、本発明の第1の実施形態を示し、処理実行サーバ管理テーブル500の一例を示す図である。処理実行サーバ管理テーブル500は、並列分散処理実行サーバ120を管理する属性情報として、各並列分散処理実行サーバ120を識別するためのサーバ名501の情報を有する。<タスク管理テーブル>図6は、本発明の第1の実施形態を示し、タスク管理部1334が管理するタスク管理テーブル600の一例を示す図である。タスク管理テーブル600は、並列分散処理実行サーバ120の各タスクを管理する属性情報として、各タスクを識別する識別子を格納するタスクID601と、タスクが存在する並列分散処理実行サーバ120の名称または識別子を格納するサーバ名602と、タスクが処理を実行中であるか否かを示す実行状態603と、タスクが処理しているデータブロックの識別子を示す処理データブロックID604から一つのエントリを構成する。実行状態603の欄には、実行中、待ち等の情報が格納される。<S208の詳細データブロックの転送元サーバの決定>図7A、図7Bは、図4BのステップS208の詳細な処理手順を示すフローチャートである。なお、図7A、図7Bの処理は、データ転送制御サーバ140上の処理遅延サーバ抽出処理部1431、空きリソース量比較処理部1432、処理状況管理部1433が行なう処理である。上記ステップS208は、データ転送制御サーバ140が、処理完了通知を送信した並列分散処理実行サーバが次に処理するデータブロックを格納している並列分散処理実行サーバ120を抽出し、並列分散処理制御サーバ130の処理割当制御部1331にデータブロックを有するサーバ名を送信する処理である。まず、データ転送制御サーバ140の処理遅延サーバ抽出処理部1431が、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名のリストとを、並列分散処理制御サーバ130から受信するS301。このサーバ名のリストは、図4BのステップS207で処理割当制御部1331が生成した未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストである。次に、処理遅延サーバ抽出処理部1431は、処理状況管理部1433に、未割当のデータブロックをもつ並列分散処理実行サーバ120の処理済みデータ率を要求するS302。なお、処理済みデータ率の要求は、データ転送制御サーバ140の処理遅延サーバ抽出処理部1431が図4AのステップS203で受信した、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを含む。続いて、サーバ名リストを受信した処理状況管理部1433は、処理状況管理テーブル800を参照し、サーバ名リストに含まれる並列分散処理実行サーバ120の処理済みデータ率を抽出し、処理遅延サーバ抽出処理部1431に処理済みデータ率を送信するS303。そして、処理遅延サーバ抽出処理部1431は、処理遅延閾値テーブル1000を参照して、処理遅延閾値1001を取得する。処理遅延サーバ抽出処理部1431は、並列分散処理制御サーバ130から受信したサーバ名リストに含まれる並列分散処理実行サーバ120の処理済みデータ率と、処理遅延閾値1001を比較して、処理遅延閾値1001より処理済みデータ率が小さく、処理の進捗が遅い並列分散処理実行サーバ120を抽出するS304。なお、処理遅延閾値1001は、並列分散処理処理実行以前に予めユーザや管理者などが設定しておく。続いて、処理遅延サーバ抽出処理部1431が、ステップS304で抽出した処理の進捗の遅い並列分散処理実行サーバ120が存在するか否かを判定するS305。ステップS305の判定の結果、処理の進捗の遅い並列分散処理実行サーバ120が存在する場合S305→Yesは、処理遅延サーバ抽出処理部1431が、処理の進捗の遅い並列分散処理実行サーバ120のサーバ数が1台であるか判定する図7BのS308。ステップS308の判定の結果、処理の進捗の遅い並列分散処理実行サーバ120が1台の場合S308→Yes、処理遅延サーバ抽出処理部1431が、並列分散処理制御サーバ130に、処理完了通知を送信した並列分散処理実行サーバ120が次に処理するデータブロックの転送元サーバとして、処理の進捗の遅い並列分散処理実行サーバのサーバ名を送信するS315。ステップS308の判定の結果、処理の進捗の遅い並列分散処理実行サーバ120が複数ある場合S308→No、処理遅延サーバ抽出処理部1431が、処理状況管理部1433に、処理の進捗の遅い並列分散処理実行サーバ120の実行多重度を要求するS309。並列分散処理実行サーバ120の実行多重度は、同時に実行するタスクの数を示す。そして、ステップS309では、実行多重度を各並列分散処理実行サーバ120の単位時間当たりの処理性能として扱う。なお、実行多重度の要求は、処理の進捗の遅い並列分散処理実行サーバ120のサーバ名リストを含む。また、単位時間当たりの処理性能を示す値として、CPUの処理能力やメモリの容量等のハードウェアスペックを示す値を利用しても構わない。続いて、処理の進捗の遅いサーバの実行多重度を受信した処理状況管理部1433が、処理状況管理テーブル800を参照し、サーバ名リストに含まれる各並列分散処理実行サーバ120の実行多重度803を取得して、処理遅延サーバ抽出処理部1431に送信するS311。そして、処理遅延サーバ抽出処理部1431は、処理の進捗の遅い並列分散処理実行サーバ120の中で、実行多重度の最も少ない並列分散処理実行サーバ120を、処理遅延サーバとして抽出するS311。なお、単位時間当たりの処理性能として、CPUの処理能力やメモリ容量等のハードウェアスペックを利用した場合、処理遅延サーバ抽出処理部1431は、処理の進捗の遅い並列分散処理実行サーバ120の中で、ハードウェアスペックの最も低い並列分散処理実行サーバ120を処理遅延サーバとして抽出する。図9は、処理状況管理テーブルの変形例を示し、図8に示した処理状況管理テーブル800の実行多重度803に代わって、ハードウェアスペック903~905を用いるようにしたものである。図9の処理状況管理テーブル800は、図8に示した処理状況管理テーブル800の実行多重度803に代わって、並列分散処理実行サーバ120のCPU122の種別を格納するCPU種別903と、CPU122のコア数904と、主記憶装置123の種別を格納するメモリ種別905からひとつのエントリを構成する。処理状況管理テーブル800において、CPU種別903=CPU1は高スペック高性能のCPU122であることを示し、CPU2は低スペックのCPUを示し、メモリ種別のメモリ1は高スペック高性能のメモリ、メモリ2は低スペックのメモリを示しており、ハードウェアスペックの最も低い並列分散処理実行サーバ120は、1コアのCPU2と、低スペックのメモリ2と、を備えるサーバDである。なおCPU種別903に代わって、CPU122の動作クロック数とコア数及びキャッシュの容量でCPUの処理能力を表すようにしてもよい。続いて、処理遅延サーバ抽出処理部1431は、ステップS311で抽出した処理遅延サーバ数が1台であるか否かを判定するS312。ステップS312の判定の結果、処理遅延サーバ数が1台の場合S312→Yes、ステップS315の処理に進み、処理遅延サーバのサーバ名を並列分散処理制御サーバ130に送信するS315。ステップS312の判定の結果、処理遅延サーバ数が複数である場合S312→No、処理遅延サーバ抽出処理部1431が、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、処理遅延サーバのサーバ名リストを空きI/Oリソース量比較処理部1432に送信するS313。続いて、空きI/Oリソース量比較処理部1432が、リソース使用量管理サーバ150に、処理完了通知を送信した並列分散処理実行サーバ120と、処理遅延サーバの空きI/Oリソース量を要求するS314。なお、空きI/Oリソース量の要求は、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、処理遅延サーバのサーバ名リストを含む。ステップS305の結果、処理の進捗の遅い並列分散処理実行サーバ120が存在しない場合S305→No、処理遅延サーバ抽出処理部1431が、空きI/Oリソース量比較処理部1432に、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを送信するS306。そして、空きI/Oリソース量比較処理部1432が、リソース使用量管理サーバ150に処理完了通知を送信した並列分散処理実行サーバ120と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを送信し、各並列分散処理実行サーバ120の空きI/Oリソース量を要求するS307。なお、空きI/Oリソース量の要求は、処理完了通知を送信した並列分散処理実行サーバのサーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストを含む。次に、リソース使用量管理サーバ150が、データ転送制御サーバ140から受信した処理完了通知を送信した並列分散処理実行サーバ120と、サーバ名リストに含まれる並列分散処理実行サーバ120の空きI/Oリソース量を、空きI/Oリソース量比較処理部1432に送信する図7AのS316。なお、空きI/Oリソース量は、ネットワークI/Oの空きリソース量と、ディスクI/Oの空きリソース量とを含む。ステップS316の処理の詳細については後述する。そして、空きI/Oリソース量比較処理部1432が、処理完了通知を送信した並列分散処理実行サーバ120の空きI/Oリソース量と、各並列分散処理実行サーバ120の空きI/Oリソース量を比較し、最も空きI/Oリソース量の差が小さい並列分散処理実行サーバ120を、処理完了通知を送信した並列分散処理実行サーバ120が次に処理するデータブロックの転送元サーバとして抽出する。空きI/Oリソース量比較処理部1432は、抽出した並列分散処理実行サーバ120のサーバ名を並列分散処理制御サーバ130へ送信するS317。なお、空きI/Oリソース量の比較方法は、空きI/Oリソース量比較処理部1432が、ディスクI/Oの空きリソース量の差の絶対値と、ネットワークI/Oの空きリソース量の差の絶対値を求め、各空きI/Oリソース量の差の絶対値が大きい方の値を求めることで行なわれる。すなわち、空きI/Oリソース量の差は以下の式2で算出されるものである。空きI/Oリソース量の差=max{|ディスクI/Oの空きリソース量の差|、|ネットワークI/Oの空きリソース量の差|}2なお、処理完了通知を送信した並列分散処理実行サーバ120が次に処理するデータブロックの転送元サーバ抽出のバリエーションとして、処理完了通知を送信した並列分散処理実行サーバの空きI/Oリソース量を、可能な限りデータブロックの転送に利用するために、ディスクI/OとネットワークI/Oの空きリソース量が、処理完了通知を送信した並列分散処理実行サーバ120より多い空きI/Oリソース量を持つ並列分散処理実行サーバ120の中から、最も空きI/Oリソース量の差が小さい並列分散処理実行サーバを抽出してもよい。<処理状況管理テーブル>図8は、本発明の第1の実施形態を示し、処理状況管理テーブル800の一例を示す図である。図9は、処理状況管理テーブルの変形例を示し、図8に示した処理状況管理テーブル800の実行多重度803に代わって、ハードウェアスペック903~905を用いるようにしたものである。ここで、図8は並列分散処理実行サーバ120の単位時間当たりの処理性能として、実行多重度を利用した場合を示し、図9は並列分散処理実行サーバの単位時間当たりの処理性能として、CPUやメモリ等のハードウェアスペックを利用した場合を示している。処理状況管理テーブル800は、各並列分散処理実行サーバ120の処理の進捗状況を管理する属性情報として、各並列分散処理実行サーバ120を識別するためのサーバ名801、各並列分散処理実行サーバに格納されているデータブロックに対する、処理済みのデータブロックの割合を示す処理済みデータ率802、各並列分散処理実行サーバ120の単位時間当たりの処理性能を示す実行多重度803の各情報を有する。図9の処理状況管理テーブル800は、前記処理状況管理テーブル800から、実行多重度803が削除され、各並列分散処理実行サーバ120の単位時間当たりの処理性能としてCPUのスペック名称を示すCPU種別903、CPUコア数904、メモリのスペック名称を示すメモリ種別904の各情報が追加されている。<処理遅延閾値管理テーブル>図10は、本発明の第1の実施形態を示し、処理遅延閾値管理テーブルの一例を示す図である。処理遅延閾値管理テーブル1000は、各並列分散処理実行サーバ120の処理の進捗が遅延しているか判定するための基準を管理する属性情報として、並列分散処理実行サーバ120の処理済みデータ率が小さく、処理が遅れているか否かを判定するための閾値の情報を有する。<S316の詳細空きI/Oリソース量の取得>図11は、図7AのステップS316の詳細な処理手順を示すフローチャートである。なお、図11の処理は、リソース使用量管理サーバ150上のリソース使用量管理部1531が行なう処理である。まず、リソース使用量管理部1531が、処理完了通知を送信した並列分散処理実行サーバ120のサーバ名と、サーバ名リストを、データ転送制御サーバ140から受信するS401。続いて、リソース使用量管理部1531が、リソース使用量管理テーブル1200を参照し、各並列分散処理実行サーバ120のネットワークI/Oの空きI/Oリソース量と、ディスクI/Oの空きI/Oリソース量を算出し、データ転送制御サーバ140へ送信するS316。<リソース使用量管理テーブル>図12は、本発明の第1の実施形態を示し、リソース使用量管理テーブル1200の一例を示す図である。リソース使用量管理テーブル1200は、各並列分散処理実行サーバ120のI/Oリソース使用量を管理する属性情報として、各並列分散処理実行サーバ120を識別するためのサーバ名1201、ネットワークI/O使用量1202、ディスクI/O使用量1203の各情報を有する。なお、リソース使用量管理テーブル1200では、ネットワークI/O使用量、ディスクI/O使用量を各並列分散処理実行サーバ全体のI/O帯域幅に対する使用している帯域幅の割合で示したが、使用しているI/O帯域幅をそのまま利用しても構わない。<データブロックの転送元となる並列分散処理実行サーバ選択の概念図>次に、図13を参照して、本発明の第1の実施形態を示し、データブロックの転送元となる並列分散処理実行サーバ120を選択する処理の概念を説明する。図13は、並列分散処理実行サーバが4台の並列分散処理実行サーバ120-1~120-4が稼動する環境下で、並列分散処理実行サーバAからタスクA3の処理完了通知を並列分散処理制御サーバ130が受信した際の、データブロックの転送元サーバ選択の例を示すものである。なお、並列分散処理実行サーバ120-1は並列分散処理実行サーバAで示し、並列分散処理実行サーバ120-2~120-4を並列分散処理実行サーバB~Dで示す。図13において、データ情報管理テーブル300から、未処理のデータブロックがまだ存在し、処理完了通知を送信した並列分散処理実行サーバAに格納されているデータブロック1~10は全て割当済みであることを示している。処理割当制御部1331は、データ情報管理部1332から未割当のデータブロックを持つ並列分散処理実行サーバ120のサーバ名リストサーバB、サーバC、サーバDを取得し、処理完了通知を送信した並列分散処理実行サーバAに未割当のデータブロックが存在しないことを確認する。続いて、処理割当制御部1331は、データ転送制御サーバ140に処理完了通知を送信した並列分散処理実行サーバ120のサーバ名サーバAと、未割当のデータブロック20、30、40を持つ並列分散処理実行サーバ120のサーバ名リストサーバB、サーバC、サーバDを送信する。次に、データ転送制御サーバ140の処理遅延サーバ抽出処理部1431が、受信したサーバ名リストに含まれる並列分散処理実行サーバB、C、Dの処理済みデータ率を、処理状況管理部1433から取得する。そして、処理遅延閾値管理テーブル1000に格納されている閾値と、各並列分散処理実行サーバB、C、Dの処理済みデータ率をそれぞれ比較する。比較の結果、処理の進捗の遅い並列分散処理実行サーバ120として、閾値50よりも処理済みデータ率が小さい並列分散処理実行サーバC、Dが抽出され、抽出した並列分散処理実行サーバ120のサーバ名リストサーバC、サーバDを処理状況管理部1433に送信し、各並列分散処理実行サーバ120の実行多重度を得る。そして、各並列分散処理実行サーバ120の実行多重度を比較し、並列分散処理実行サーバC、Dともに実行多重度は1であるため、並列分散処理実行サーバC、Dを処理遅延サーバとして抽出し、処理完了通知を送信した並列分散処理実行サーバAのサーバ名と共に、空きI/Oリソース量比較処理部1432に送信する。続いて、空きI/Oリソース量比較処理部1432は、処理完了通知を送信した並列分散処理実行サーバAと、処理遅延サーバB、Cの空きI/Oリソース量をリソース使用量管理サーバ150から取得する。そして、処理完了通知を送信した並列分散処理実行サーバAと、各処理遅延サーバC、Dの空きI/Oリソース量を比較する。比較の結果、並列分散処理実行サーバAと並列分散処理実行サーバCの空きI/Oリソース量の差は30、並列分散処理実行サーバAと並列分散処理実行サーバDの空きI/Oリソース量の差は5となり、空きI/Oリソース量比較処理部1432は、空きI/Oリソース量の差の小さい並列分散処理実行サーバDをデータブロックの転送元サーバとして選択する。続いて、空きI/Oリソース量比較処理部1423は、並列分散処理制御サーバ130に、選択した並列分散処理実行サーバ120のサーバ名サーバDを送信する。サーバ名サーバDを受信した並列分散処理制御サーバ130の処理割当制御部1331は、並列分散処理実行サーバDに格納されている未割当のデータブロックの中からデータ40を選択し、並列分散処理実行サーバDに対して、データ40の並列分散処理実行サーバAへの転送を要求する。そして、処理割当制御部1331は、並列分散処理実行サーバAに対して、転送したデータブロックデータ40のタスクA3での処理実行要求を送信する。<まとめ>以上のように第1の実施形態によれば、並列分散処理の実行中に利用可能なコンピュータリソースが変化し、並列分散処理の実行多重度が変化する並列分散処理システムであっても、ある並列分散処理実行サーバに他の並列分散処理実行サーバ上に格納されたデータブロックの処理を割り当てる際に、各並列分散処理実行サーバの処理の進捗状況と、単位時間当たりのデータ処理能力から、処理の進捗の遅いサーバを抽出し、抽出した並列分散処理実行サーバと、データブロックの処理を割り当てる並列分散処理実行サーバの空きI/Oリソース量を比較し、最も差の小さい並列分散処理実行サーバをデータブロックの転送元サーバとして選択することで、データ転送回数を削減し、データ転送時間を短縮して、並列分散処理全体を効率的に実行することができる。なお、上記第1実施形態では、上記図7BのステップS311の処理で、最も性能の低い並列分散処理実行サーバ120を選択する例として、実行多重度を用いたが、上述のようにハードウェアスペックの値を用いて、最もハードウェアスペックの低い並列分散処理実行サーバ120を最も処理性能の低い計算機として選択することができる。また、単位時間当たりの処理性能として、CPUの処理能力やメモリ容量等のハードウェアスペックを利用した場合、処理遅延サーバ抽出処理部1431は、処理の進捗の遅い並列分散処理実行サーバ120の中で、ハードウェアスペックの最も低い並列分散処理実行サーバ120を処理遅延サーバとして抽出するようにしてもよい。また、処理遅延サーバの判定は、上述の処理済みデータ率と閾値1001に限定されるものではなく、処理タスクの実行開始から所定時間を経過しても処理完了通知を送信していない並列分散処理実行サーバ120を処理遅延サーバとしてもよい。処理遅延サーバの判定は、この他、所定時間経過後のデータブロックの未処理の比率と閾値を比較するようにしても良い。また、空きI/Oリソース量を実使用量と理論値の比率で表す例を示したが、ネットワークインターフェース121や二次記憶装置124の実効値や、理論値に所定比率を乗じた値を用いることができる。すなわち、ネットワークI/Oの空きI/Oリソース量=実効値-現在使用帯域÷実効値である。また、二次記憶装置124で利用可能なデータ転送速度MByte/secも、ディスクI/Oの空きI/Oリソース量=実効値-現在使用帯域÷実効値とすることができる。なお、実効値に代わって、理論値に所定の比率を乗じた値を用いてもよい。<第2実施形態>次に、図14A~図18を参照して、本発明の第2の実施形態について説明する。前述の第1の実施形態では、各並列分散処理実行サーバ120は1台の物理サーバとして稼動し、ある並列分散処理実行サーバ120に、他の並列分散処理実行サーバ120に格納されたデータブロックの処理を割り当てる際に、各並列分散処理実行サーバ120の処理の進捗率と、空きリソース量を基に、データブロックの転送元サーバを選択していた。しかし、物理サーバ上で複数の並列分散処理実行サーバ120が仮想サーバとして実行される場合、同一物理サーバ上で稼動する並列分散処理実行サーバ120間のデータ転送は、ネットワーク100を経由したデータの転送を必要とせず、物理サーバの内部での通信となるため、異なる物理サーバ上で稼動する並列分散処理実行サーバ120間のデータ転送よりも、データブロックの転送時間が短くなる可能性がある。そこで、第2の実施形態では、物理サーバ上に複数の並列分散処理実行サーバ120が仮想サーバとして稼動する環境下において、ある並列分散処理実行サーバ120に、他の並列分散処理実行サーバ120に格納されたデータブロックの処理を割り当てる際に、同一物理サーバ上で稼動する並列分散処理実行サーバ120からデータブロックの転送元サーバを選択することを優先した上で、各並列分散処理実行サーバ120の処理の進捗率と、空きリソース量を基に、データブロックの転送元サーバとして選択するものである。<システム構成>図14A、図14Bは、本発明の第2の実施形態を示し、計算機システム20の一例を示すブロック図である。図14A、図14Bに示す計算機システム20には、並列分散処理実行サーバ120は、並列分散処理を実行する仮想サーバとして物理サーバ210-1で実行される。物理サーバ210-1~210-nの総称は物理サーバ210とする。物理サーバ210は、物理リソース220上で実行される仮想化部230が複数の仮想サーバを提供し、各仮想サーバは並列分散処理実行サーバ120として実行される。仮想化部230は、物理リソース220を複数の並列分散処理実行サーバ120に割り当てるハイパーバイザやVMMVirtualMachineMonitorで構成される。物理リソース220は、CPU122、主記憶装置123、ネットワークインターフェース121、二次記憶装置124から構成される。仮想化部230は、各並列分散処理実行サーバ120に対して、仮想または論理CPU122v、仮想主記憶装置123v、仮想ネットワークインターフェース121v、仮想化された二次記憶装置124vを提供する。各並列分散処理実行サーバ120の機能要素は前記第1実施形態と同様である。また、並列分散処理制御サーバ130、データ転送制御サーバ140及びリソース使用量管理サーバ150も前記第1実施形態同様に構成される。ただし、処理実行サーバ管理テーブル1700と、処理状況管理テーブル1800は、前記第1実施形態の構成に、物理サーバ210と仮想サーバの対応関係が加えられた点が、前記第1実施形態との相違点である。<全体処理>図15A、図15Bは、本発明の第2の実施形態を示し、並列分散処理実行方法に関する全体処理の手順を示すフローチャートである。なお、図15A、図15Bは、第1実施形態の図2A、図2Bに対応する処理である。まず、クライアント装置110から並列分散処理制御サーバ130へ、処理対象データ入力データ1241のデータロードの要求が送信される。データロードの要求を受信した並列分散処理制御サーバ130は、ロード対象データを規定のサイズのデータブロック入力データ1241に分割し、複数の並列分散処理実行サーバ120に分散してロードするS501。次に、クライアント装置110から並列分散処理制御サーバ130へ、並列分散処理実行要求が送信される。並列分散処理実行要求を受信した並列分散処理制御サーバ130の処理割当制御部1331は、各並列分散処理実行サーバ120に、ロードされたデータブロックの処理実行要求を送信し、データ情報管理部1332がデータ情報管理テーブル300の処理対象となるデータブロックの割当状態303を割当済みに更新するS502。なお、処理割当制御部1331が送信するデータブロックの処理実行要求は、処理対象のデータブロックのデータブロックIDと、各データブロックの処理を実行するタスクのタスクIDとを含む。なお、ここでタスクは、処理対象のデータブロックを入力データ1241として所定の処理を実行するプログラムである。続いて、処理実行要求を受信した並列分散処理実行サーバ120のユーザ定義処理実行部1231は、各タスクにおいて処理実行要求で指定されたデータブロックを入力データ1241として所定の処理を実行するS503。すなわち、並列分散処理実行サーバ120のユーザ定義処理実行部1231は、処理実行要求で指定されたタスクを起動し、各タスクに対して処理実行要求で指定されたデータブロックを入力データ1241として割り当てて処理を実行させる。並列分散処理実行サーバ120は、割り当てられたタスクに入力データ1241として指定されたデータブロックの処理が完了すると、並列分散処理実行サーバ120から並列分散処理制御サーバ130へ、処理完了通知として処理の完了したタスクのタスクIDを送信するS504。そして、処理完了通知を受信した並列分散処理制御サーバ130のデータ情報管理部1332が、データ情報管理テーブル300の該当するデータブロックID301の処理状態304を処理済みに更新するS505。続いて、並列分散処理制御サーバ130の処理割当制御部1331がデータ情報管理テーブル300を参照し、全てのデータブロックの処理状態304が処理済みであるか否かを判定するS506。ステップS506の判定の結果、全てのデータブロックが処理済みである場合S506→Yes、並列分散処理制御サーバ130の処理割当制御部1331が、クライアント装置110に並列分散処理完了通知を送信し、計算機システム10は処理を終了する。ステップS506の判定の結果、何れかのデータブロックが処理済みでない場合S506→No、並列分散処理制御サーバ130は、データ管理情報テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120に、割当状態403が未割当のデータブロックが存在するか否かを判定するS507。ステップS507の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120に未割当のデータブロックが存在する場合S507→Yes、並列分散処理制御サーバ130の処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120に存在する未割当のデータブロックの中から、任意に一つのデータブロックを選択するS508。ステップS507の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120に未割当のデータブロックが存在しない場合S507→Noは、図15Bに進む。図15Bでは、並列分散処理制御サーバ130がデータ情報管理テーブル300を参照して、処理完了通知を送信した並列分散処理実行サーバ120が稼動する物理サーバ210上の他の並列分散処理実行サーバ120に、割当状態303が未割当のデータブロックが存在するか否かを判定するS509。つまり、同一の物理サーバ210上で未割当のデータブロックを有する仮想サーバを抽出する。ステップS509の判定の結果、未割当のデータブロックが存在する場合S509→Yes、並列分散処理制御サーバ130は、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名と、前記並列分散処理実行サーバ120と同一の物理サーバ上で稼動し、未割当のデータブロックを格納している並列分散処理実行サーバ120の仮想サーバ名リストをデータ転送制御サーバ140に送信するS510。ステップS509の判定の結果、同一の物理サーバ210上で未割当のデータブロックが存在しない場合S509→No、並列分散処理制御サーバ130の処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストをデータ転送制御サーバ140に送信するS511。次に、データ転送制御サーバ140の処理遅延サーバ抽出処理部1431が、並列分散処理制御サーバ130から受信した未割当のデータブロックを持つ仮想サーバ名リストに含まれる並列分散処理実行サーバ120の中から、処理遅延サーバを抽出し、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名と、抽出したサーバの仮想サーバ名リストと、をリソース使用量管理サーバ150へ送信し、各並列分散処理実行サーバ120の空きI/Oリソース量を要求するS512。続いて、リソース使用量管理サーバ150のリソース使用量管理部1531が、リソース使用量管理テーブル1200を参照し、処理完了通知を送信した並列分散処理実行サーバ120と、仮想サーバ名リストに含まれる並列分散処理実行サーバ120の空きI/Oリソース量を転送制御サーバ140へ送信するS513。なお、空きI/Oリソース量は、前記第1実施形態と同様であり、ネットワークI/Oの空きI/Oリソース量と、ディスクI/Oの空きI/Oリソース量とを含む。そして、データ転送制御サーバ140の空きI/Oリソース量比較処理部1432が、処理完了通知を送信した並列分散処理実行サーバ120と、各処理遅延サーバの空きI/Oリソース量を比較し、最も空きI/Oリソース量の差が小さい処理遅延サーバを抽出し、抽出した前記処理遅延サーバの仮想サーバ名を並列分散処理制御サーバ130へ送信するS514。次に、並列分散処理制御サーバ130の処理割当制御部1331が、データ転送制御サーバ140から受信した仮想サーバ名に該当する並列分散処理実行サーバ120に存在する未割当のデータブロックの中から、任意に一つのデータブロックを選択するS515。そして、並列分散処理制御サーバ130のデータ情報管理部1332が、データ情報管理テーブル300において選択したデータブロックの割当状態303を割当済みに更新する。また、処理割当制御部1331が、処理完了通知を送信した並列分散処理実行サーバ120に、選択したデータブロックの処理要求を送信し、計算機システム20は処理をステップS503に戻すS516。なお、データブロックの処理要求は、選択したデータブロックのデータブロックIDと、前記データブロックの処理を実行するタスクIDとを含む。なお、ステップS507~S516の詳細については後述する。<S507~S516の詳細並列分散処理実行サーバへの処理の割り当て>図16A、図16Bは、図15A、図15Bに示したステップS507からステップS516の並列分散処理実行サーバ120への処理の割り当てに関する詳細な手順を示すフローチャートである。なお、図16A、図16Bの処理は、並列分散処理制御サーバ130上の処理割当制御部1331、データ情報管理部1332、タスク管理部1334が行なう処理であり、第1実施形態の図4A、図4Bに対応する処理である。まず、並列分散処理実行サーバ120から処理完了通知としてタスクIDを受信した、処理割当制御部1331が、タスク管理部1334にタスク管理テーブル600の更新を要求する。また、処理割当制御部1331がデータ情報管理部1332に未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストを要求するS601。次に、タスク管理テーブル600の更新要求を受信したタスク管理部1334が、タスク管理テーブル600で該当するタスクの実行状態603を待ちに更新し、処理データブロックID604をNULL値更新するS602。続いて、データ情報管理部1332が、データ情報管理テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120の処理済みデータ率を算出して、データ転送制御サーバ140に処理済みデータ率と並列分散処理実行サーバ120の名称を送信する。処理済みデータ率を受信した転送制御サーバ140は、受信した仮想サーバ名について処理状況管理テーブル1800から処理済みデータ率1803を更新するS603。なお、処理済みデータ率の算出方法は、前記第1実施形態と同様であり、データ情報管理部1332が、並列分散処理実行サーバ120に格納されているデータブロック数に対する、データ情報管理テーブル300の処理状態304が処理済みのデータブロック数の比率を求めることで行われる。すなわち、処理済みデータ率は、並列分散処理実行サーバ120が格納している全データブロックに対して、現時点までに処理が完了したデータブロックの割合であり、前記第1実施形態に示した式1で算出されるものである。そして、データ情報管理部1332が、データ情報管理テーブル300を参照し、割当状態303が未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストを作成し、処理割当制御部1331に送信するS604。未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストを受信した処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120が仮想サーバ名リストに含まれるか否かを判定するS605。ステップS605の判定の結果、処理完了通知を送信した並列分散処理実行サーバ120がリストに存在する場合S605→Yes、処理割当制御部1331が、データ情報管理テーブル300を参照し、処理完了通知を送信した並列分散処理実行サーバ120に保存されている未割当のデータブロックの中から、任意に一つのデータブロックを選択するS606。ステップS605の結果、処理完了通知を送信した並列分散処理実行サーバ120がリストに存在しない場合S605→Noは、図16Bの処理へ進む。図16Bにおいて、処理割当制御部1331は、処理実行サーバ管理部1333に処理完了通知を送信した並列分散処理実行サーバ120と同一物理サーバ上で稼動する並列分散処理実行サーバ120の仮想サーバ名リストを要求するS607。なお、仮想サーバ名リストの要求は、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名を含む。続いて、処理実行サーバ管理部1333は、受信した仮想サーバ名に該当する並列分散処理実行サーバ120と同一の物理サーバ上の他の並列分散処理実行サーバ120の仮想サーバ名リストを処理割当制御部1331に送信するS608。そして、処理割当制御部1331が受信した仮想サーバ名リスト内に、処理完了通知を送信した並列分散処理実行サーバ120が稼動する物理サーバ210上の他の並列分散処理実行サーバ120が存在するか否かを判定するS609。ステップS609の判定の結果、他の並列分散処理実行サーバ120が存在する場合S609→Yes、処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名と、同一物理サーバ上で稼動し、未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストをデータ転送制御サーバ140に送信する。次に処理するデータブロックを保持する並列分散処理実行サーバ120を要求するS611。ステップS609の結果、他の並列分散処理実行サーバ120が存在しない場合S609→No、処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120のタスクが次に処理するデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名を要求するS610。なお、仮想サーバ名の要求は、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名と、未割当のデータブロックを持つ並列分散処理実行サーバ120の仮想サーバ名リストとを含む。データ転送制御サーバ140が、処理完了通知を送信した並列分散処理実行サーバが次に処理するデータブロックを格納している並列分散処理実行サーバ120を抽出し、並列分散処理制御サーバ130に送信するS612。なお、ステップS612の処理は前記第1実施形態の図7A、図7Bの処理に対応する処理である。続いて、処理割当制御部1331が、受信した仮想サーバ名に該当する並列分散処理実行サーバ120に格納されている未割当のデータブロックの中から、任意に一つのデータブロックを選択するS613。そして、処理割当制御部1331は、データ転送制御サーバ140から受信した仮想サーバ名に該当する並列分散処理実行サーバ120に対して、処理完了通知を送信した並列分散処理実行サーバ120へ、上記ステップS613で選択したデータブロックを転送するよう要求するS614。なお、データブロックの転送要求は、選択したデータブロックのデータブロックIDと、処理完了通知を送信した並列分散処理実行サーバ120の仮想サーバ名とを含む。そして、処理割当制御部1331は、図16Aの処理に戻って、データ情報管理部1332にデータ情報管理テーブル300の更新を要求し、タスク管理部1334にタスク管理テーブル600の更新を要求するS615。なお、データ情報管理テーブル300の更新要求は、選択したデータブロックのデータブロックID301を含み、タスク管理テーブル600の更新要求は、並列分散処理実行サーバ120から処理完了通知として受信したタスクIDと、選択したデータブロックのデータブロックID601を含む。データ情報管理部1332は、受信した更新要求に含まれるデータブロックIDに該当するデータブロックに対応するデータ情報管理テーブル300の割当状態303を割当済みに更新する。タスク管理部1334は、受信したタスクIDに該当するタスク管理テーブル600のタスクの実行状態603を実行中に更新し、受信したデータブロックIDで処理対象データブロックID604を更新するS616。そして、処理割当制御部1331は、処理完了通知を送信した並列分散処理実行サーバ120に、選択したデータブロックの処理実行要求を送信するS617。なお、処理実行要求は、選択したデータブロックのデータブロックIDと、処理を割り当てるタスクIDとを含む。<処理実行サーバ管理テーブル>図17は、本発明の第2の実施形態を示し、処理実行サーバ管理テーブル1700の一例を示す図である。なお、図17は、第1実施形態の図5に示した処理実行サーバ管理テーブル500に対応するテーブルである。処理実行サーバ管理テーブル1700は、物理サーバ210と並列分散処理実行サーバ120を管理する属性情報として、各物理サーバ210を識別するための物理サーバ名1701と、仮想サーバで構成された各並列分散処理実行サーバ120を識別するための仮想サーバ名1702からひとつのエントリが構成される。<処理状況管理テーブル>図18は、本発明の第2の実施形態を示し、処理状況管理テーブル1800の他の例を示す図である。なお、図18は、第1実施形態の図8に示した処理状況管理テーブル800に対応するテーブルである。処理状況管理テーブル1800は、各並列分散処理実行サーバ120の処理の進捗状況を管理する属性情報として、各物理サーバ210を識別するための物理サーバ名1801、各並列分散処理実行サーバ120を識別するための仮想サーバ名1802、各並列分散処理実行サーバに格納されているデータブロックに対する処理済みのデータブロックの割合を示す処理済みデータ率1803、各並列分散処理実行サーバの単位時間当たりの処理性能を示す実行多重度1804からひとつのエントリが構成される。なお、処理状況管理テーブル1800には、CPUやメモリ等のハードウェアスペックの属性情報が示されていないが、これらのハードウェアスペックを含むようにしても良い。<まとめ>以上のように、第2の実施形態によれば、複数の並列分散処理実行サーバ120が仮想サーバとして物理サーバ210上で稼動する環境下において、処理実行中に利用可能なコンピュータリソースが変化し、並列分散処理の実行多重度が変化する並列分散処理システムであっても、ある並列分散処理実行サーバ120に他の並列分散処理実行サーバ120上に格納されたデータブロック40の処理を割り当てる際に、同一の物理サーバ210上で稼動する並列分散処理実行サーバ120から優先して、各並列分散処理実行サーバ120の処理の進捗状況と、単位時間当たりのデータ処理能力と、空きI/Oリソース量と、を基にデータブロックの転送元サーバとして選択することで、並列分散処理全体を効率的に実行することが出来る。