"複数のノードを並列に用いて第1のデータ処理を行い、前記第1のデータ処理の結果に対して第2のデータ処理を行うシステムで実行されるデータ処理方法であって、過去に実行された前記第1のデータ処理の結果を記憶する記憶装置を参照して、データを、前記記憶装置に記憶された結果を利用可能な第1のセグメントと、前記記憶装置に記憶された結果を利用可能でない複数の第2のセグメントとに分割し、前記複数の第2のセグメントを前記複数のノードに割り当て、前記複数のノードを並列に用いて前記複数の第2のセグメントに対して前記第1のデータ処理を行い、前記記憶装置に記憶された前記第1のセグメントに対応する過去の結果と、前記複数のノードを用いて得られた前記複数の第2のセグメントに対応する結果とに対して、前記第2のデータ処理を行う、ことを特徴とするデータ処理方法。前記複数の第2のセグメントに対する前記第1のデータ処理の結果を、前記記憶装置に格納する、請求の範囲第1項記載のデータ処理方法。前記データのうち、過去に前記第1のデータ処理を実行していない第1の部分と、過去に前記第1のデータ処理を実行しているが利用可能な結果が前記記憶装置に記憶されていない第2の部分とが、異なる第2のセグメントに属するように前記データを分割し、前記第1の部分に対する前記第1のデータ処理の結果と前記第2の部分に対する前記第1のデータ処理の結果とを分けて、前記記憶装置に格納する、請求の範囲第2項記載のデータ処理方法。前記記憶装置に記憶された、前記データの少なくとも一部と他のデータとを含む集合に対して過去に実行された前記第1のデータ処理の結果は、前記データを処理するために利用可能でないと判定する、請求の範囲第1項記載のデータ処理方法。前記第2のデータ処理は、前記第1のセグメントに対する前記第1のデータ処理の結果と前記複数の第2のセグメントに対する前記第1のデータ処理の結果とを集計する処理を含む、請求の範囲第1項記載のデータ処理方法。複数のノードを並列に用いて第1のデータ処理を行い、前記第1のデータ処理の結果に対して第2のデータ処理を行う分散処理システムであって、過去に実行された前記第1のデータ処理の結果を記憶する記憶装置と、前記記憶装置を参照して、データを、前記記憶装置に記憶された結果を利用可能な第1のセグメントと、前記記憶装置に記憶された結果を利用可能でない複数の第2のセグメントとに分割する分割手段と、前記複数の第2のセグメントを前記複数のノードに割り当て、前記複数のノードを並列に用いて前記複数の第2のセグメントに対して前記第1のデータ処理を実行させ、前記記憶装置に記憶された前記第1のセグメントに対応する過去の結果と、前記複数のノードを用いて得られた前記複数の第2のセグメントに対応する結果とに対して、前記第2のデータ処理を実行するよう制御する制御手段と、を有することを特徴とする分散処理システム。複数のノードを並列に用いて第1のデータ処理を行い、前記第1のデータ処理の結果に対して第2のデータ処理を行うシステムの制御に用いられるプログラムであって、コンピュータに、過去に実行された前記第1のデータ処理の結果を記憶する記憶装置を参照して、データを、前記記憶装置に記憶された結果を利用可能な第1のセグメントと、前記記憶装置に記憶された結果を利用可能でない複数の第2のセグメントとに分割し、前記複数の第2のセグメントを前記複数のノードに割り当て、前記複数のノードを並列に用いて前記複数の第2のセグメントに対して前記第1のデータ処理を実行させ、前記記憶装置に記憶された前記第1のセグメントに対応する過去の結果と、前記複数のノードを用いて得られた前記複数の第2のセグメントに対応する結果とに対して、前記第2のデータ処理を実行するよう制御する、処理を実行させることを特徴とするプログラム。上記のように、分散処理システムには、データを分割し複数のノードを用いて第1段階のデータ処理を行い、第1段階のデータ処理の結果に対して第2段階のデータ処理を行うものがある。しかし、与えられたデータが自動的に分割され並列処理化される場合、従来の分散処理システムでは、毎回、与えられたデータ全体を対象として第1段階のデータ処理から実行されていた。このため、過去のデータ処理の結果が活用されていなかった。一側面では、本発明は、データ処理を効率化できるデータ処理方法、分散処理システムおよびプログラムを提供することを目的とする。一実施態様では、複数のノードを並列に用いて第1のデータ処理を行い、第1のデータ処理の結果に対して第2のデータ処理を行うシステムが実行するデータ処理方法が提供される。このデータ処理方法では、過去に実行された第1のデータ処理の結果を記憶する記憶装置を参照して、データを、記憶装置に記憶された結果を利用可能な第1のセグメントと、記憶装置に記憶された結果を利用可能でない複数の第2のセグメントとに分割する。複数の第2のセグメントを複数のノードに割り当て、複数のノードを並列に用いて複数の第2のセグメントに対して第1のデータ処理を行う。記憶装置に記憶された第1のセグメントに対応する過去の結果と、複数のノードを用いて得られた複数の第2のセグメントに対応する結果とに対して、第2のデータ処理を行う。また、一実施態様では、複数のノードを並列に用いて第1のデータ処理を行い、第1のデータ処理の結果に対して第2のデータ処理を行う分散処理システムが提供される。この分散処理システムは、記憶装置と分割手段と制御手段とを有する。記憶装置は、過去に実行された第1のデータ処理の結果を記憶する。分割手段は、記憶装置を参照して、データを、記憶装置に記憶された結果を利用可能な第1のセグメントと、記憶装置に記憶された結果を利用可能でない複数の第2のセグメントとに分割する。制御手段は、複数の第2のセグメントを複数のノードに割り当て、複数のノードを並列に用いて複数の第2のセグメントに対して第1のデータ処理を実行させ、記憶装置に記憶された第1のセグメントに対応する過去の結果と、複数のノードを用いて得られた複数の第2のセグメントに対応する結果とに対して、第2のデータ処理を実行するよう制御する。また、一実施態様では、コンピュータが実行するプログラムであって、複数のノードを並列に用いて第1のデータ処理を行い、第1のデータ処理の結果に対して第2のデータ処理を行うシステムの制御に用いられるプログラムが提供される。一実施態様によれば、データ処理を効率化できる。本発明の上記および他の目的、特徴および利点は本発明の例として好ましい実施の形態を表す添付の図面と関連した以下の説明により明らかになるであろう。以下、本実施の形態を図面を参照して説明する。[第1の実施の形態]図1は、第1の実施の形態の分散処理システムを示す図である。第1の実施の形態の分散処理システムは、複数のノードを並列に用いて第1のデータ処理を行い、第1のデータ処理の結果に対して第2のデータ処理を行う。例えば、MapReduceでは、第1のデータ処理としてMapフェーズを、第2のデータ処理としてReduceフェーズを行う。第1の実施の形態の分散処理システムは、記憶装置1、情報処理装置2,3およびノード4,4aを含む。記憶装置1、情報処理装置2,3およびノード4,4aは、ネットワークで接続されている。ノード4,4aは、入力されたデータを並列に処理するためのノードである。情報処理装置2,3およびノード4,4aは、CPUCentralProcessingUnitなどのプロセッサとRAMRandomAccessMemoryなどのメモリとを備えてもよく、メモリに記憶されたプログラムをプロセッサが実行するコンピュータであってもよい。記憶装置1は、ネットワークを介してアクセス可能なデータストレージである。例えば、記憶装置1はHDDHardDiskDriveやSSDSolidStateDriveなどのデバイスにデータを格納する。記憶装置1は、過去に実行された第1のデータ処理の結果を記憶する。情報処理装置2は、分割手段2aを有する。分割手段2aは、記憶装置1を参照して、データD1を、記憶装置1に記憶された結果を利用可能な第1のセグメントSG1と、記憶装置1に記憶された結果を利用可能でない第2のセグメントSG2,SG3とに分割する。情報処理装置3は、制御手段3aを有する。制御手段3aは、第2のセグメントSG2,SG3をノード4,4aに割り当て、ノード4,4aを並列に用いて第2のセグメントSG2,SG3に対して第1のデータ処理を実行させる。制御手段3aは、記憶装置1に記憶された第1のセグメントSG1に対応する過去の結果SG1aと、ノード4,4aを用いて得られた第2のセグメントSG2,SG3に対応する結果SG2a,SG3aとに対して、第2のデータ処理を実行するよう制御する。第2のデータ処理の対象となるのは、データD1を並列処理した結果であるデータD2である。データD2には、セグメントSG1,SG2,SG3を処理した結果SG1a,SG2a,SG3aが含まれる。制御手段3aは、例えばノード4,4aを並列に用いてデータD2に対する第2のデータ処理を行う。第1の実施の形態の分散処理システムによれば、分割手段2aにより、記憶装置1が参照され、データD1が、記憶装置に記憶された結果を利用可能な第1のセグメントSG1と、記憶装置1に記憶された結果を利用可能でない第2のセグメントSG2,SG3とに分割される。制御手段3aにより、第2のセグメントSG2,SG3がノード4,4aに割り当てられ、ノード4,4aが並行に用いられて、第2のセグメントSG2,SG3に対して第1のデータ処理が実行される。制御手段3aにより、記憶装置1に記憶された第1のセグメントSG1に対応する過去の結果SG1aと、ノード4,4aを用いて得られた第2のセグメントSG2,SG3に対応する結果SG2a,SG3aとに対して、第2のデータ処理が実行されるよう制御される。これにより、データ処理を効率化できる。具体的には、過去の処理結果を流用可能なセグメントSG1以外のセグメントSG2,SG3をノード4,4aに分散処理させるので、データD1の全てを処理せずに済む。既に実行された処理を重複して実行しないよう制御することで、処理効率を向上し、処理の高速化を図れる。また、ノード4,4aの負荷を軽減できる。[第2の実施の形態]図2は、第2の実施の形態の分散処理システムを示す図である。第2の実施の形態の分散処理システムは、入力されたデータを分割して複数のノードに分散処理させる。第2の実施の形態の分散処理システムでは、電子商取引において、ユーザによる商品の評価値を受け付け、その評価値に基づいて当該ユーザへ勧める商品を抽出する。具体的には、当該ユーザと他のユーザとの間の商品ごとの評価値の相関から、当該ユーザが高評価を付けると予測される商品を抽出してお勧め商品とする。第2の実施の形態の分散処理システムは、多数のユーザおよび商品に係る大量の評価値データを分散処理することで、お勧め商品の抽出処理を高速に実行する。なお、以下では、第2の実施の形態の分散処理システムが出力する、各商品の各ユーザによる予測評価値の情報を、リコメンド情報と称することがある。ここで、第2の実施の形態の分散処理システムでは、分散処理にMapReduceを用いるものとする。MapReduceを利用するためのフレームワークとして、例えば、Hadoopが知られている。ただし、他の手法により分散処理を行うシステムでもよい。他の分散処理の手法としては、例えば、MPIMessagePassingInterfaceやOpenMP登録商標などが考えられる。また、例えば、並列化コンパイラによって生成した実行ファイルを用いて分散処理するシステムでもよい。第2の実施の形態の分散処理システムは、ECElectronicCommerceサーバ20、クライアント30、実行制御サーバ100、処理解析サーバ200、処理変換サーバ300、データ分析サーバ400、登録サーバ500、入力情報管理サーバ600、中間情報管理サーバ700、結果情報管理サーバ800およびデータ処理サーバ900,900a,を含む。各サーバは、ネットワーク10に接続されている。ネットワーク10は、例えばLANLocalAreaNetworkである。また、ネットワーク10は、インターネット図示を省略に接続されている。ECサーバ20は、電子商取引による商品販売を制御するサーバコンピュータである。ECサーバ20は、ユーザの商品に対する評価値を受け付ける。ユーザは、例えばインターネットを介して通信可能な端末装置図示を省略を操作して、ECサーバ20に商品ごとの評価値を送信できる。ECサーバ20は、入力された評価値をユーザを示す情報例えば、ユーザ名に対応付けて、入力情報管理サーバ600に格納する。クライアント30は、分散処理システムを管理する管理者が操作するクライアントコンピュータである。管理者は、クライアント30を操作して、リコメンド情報を生成するためのソースプログラムを実行制御サーバ100に送信する。ソースプログラムは、例えば、Java登録商標で記述される。また、管理者は、クライアント30を操作して、処理対象とする入力情報を実行制御サーバ100に指示する。実行制御サーバ100は、分散処理システム全体を制御するサーバコンピュータである。実行制御サーバ100は、クライアント30から受信したソースプログラムや、指示された入力情報に基づいて、他のサーバと連携し、分散処理の実行を制御する。具体的には、実行制御サーバ100は、ソースプログラムを処理解析サーバ200に送信して、並列処理が可能な箇所の解析を依頼する。実行制御サーバ100は、並列処理対象の入力情報につき、処理変換サーバ300に処理変換を依頼する。処理変換では、既存の処理結果を流用した処理を組み立てて分散処理を効率的に行えるようにする。実行制御サーバ100は、処理変換サーバ300による変換結果に基づいて、データ処理サーバ900,900a,による分散処理を制御する。具体的には、実行制御サーバ100は、データ処理サーバ900,900a,のうち、アイドル状態のサーバにMap処理またはReduce処理をアサインする。実行制御サーバ100のような制御を行うノードを、マスターmasterと呼ぶことがある。処理解析サーバ200は、ソースプログラムの処理ロジックを解析するサーバコンピュータである。処理解析サーバ200は、実行制御サーバ100から受信した処理解析の依頼に応じて、ソースプログラムを解析し、並列処理が可能な箇所を特定する。処理解析サーバ200は、解析結果を実行制御サーバ100に応答する。処理変換サーバ300は、実行制御サーバ100から受信した処理変換依頼に応じて、並列処理の処理変換を行うサーバコンピュータである。具体的には、処理変換サーバ300は、入力情報のうち、過去の計算結果を流用可能な部分セグメントについては、当該過去の計算結果を流用するように処理を変換する。処理変換サーバ300は、入力情報について当該流用可能な部分を特定するための分析をデータ分析サーバ400に依頼する。処理変換サーバ300は、処理変換の結果を実行制御サーバ100に応答する。データ分析サーバ400は、処理変換サーバ300から受信した分析依頼に応じて、入力情報の分析を行うサーバコンピュータである。具体的には、データ分析サーバ400は、中間情報管理サーバ700を参照して、入力情報に含まれる部分に対する過去の計算結果を検索する。データ分析サーバ400は、入力情報を次のように分類する。すなわち、入力情報を、1新規に計算する部分と2既に処理されたことのある部分とに分類する。更に、2既に処理されたことのある部分を、2-1流用可能な処理結果がない部分と2-2流用可能な処理結果がある部分とに分類する。データ分析サーバ400は、当該分析結果を処理変換サーバ300に応答する。登録サーバ500は、データの登録処理を行うサーバコンピュータである。登録サーバ500は、データ処理サーバ900,900a,による分散処理に関するデータを取得して、中間情報管理サーバ700および結果情報管理サーバ800に当該データを格納する。入力情報管理サーバ600は、入力情報を管理するサーバコンピュータである。入力情報は、入力された日付、ユーザを示す情報および当該ユーザによる商品ごとの評価値を含む。中間情報管理サーバ700は、中間情報を管理するサーバコンピュータである。中間情報は、分散処理の過程で生成された情報である。中間情報は、Map処理の結果を含む。結果情報管理サーバ800は、結果情報を管理するサーバコンピュータである。結果情報は、分散処理の結果として生成された情報である。結果情報は、Reduce処理の結果や最終的に生成されたリコメンド情報を含む。データ処理サーバ900,900a,は、データを並列処理するサーバコンピュータである。データ処理サーバ900,900a,は、実行制御サーバ100からアサインされたMap処理またはReduce処理を実行する。このようにデータの並列処理を担当するノードを、ワーカーworkerあるいはスレーブslaveと呼ぶことがある。図3は、実行制御サーバのハードウェア例を示す図である。実行制御サーバ100は、CPU101、RAM102、HDD103、画像信号処理部104、入力信号処理部105、ディスクドライブ106および通信部107を有する。各ユニットが実行制御サーバ100のバスに接続されている。他のサーバやクライアント30も、実行制御サーバ100と同様のハードウェアを用いて実装できる。CPU101は、実行制御サーバ100の情報処理を制御するプロセッサである。CPU101は、HDD103に記憶されているプログラムやデータの少なくとも一部を読み出し、RAM102に展開してプログラムを実行する。なお、実行制御サーバ100は、複数のプロセッサを設けて、プログラムを分散して実行してもよい。RAM102は、CPU101が実行するプログラムや処理に用いるデータを一時的に記憶する揮発性メモリである。なお、実行制御サーバ100は、RAM以外の種類のメモリを備えてもよく、複数個のメモリを備えていてもよい。HDD103は、OSOperatingSystemプログラムやアプリケーションプログラムなどのプログラムおよびデータを記憶する不揮発性の記憶装置である。HDD103は、CPU101の命令に従って、内蔵の磁気ディスクに対してデータの読み書きを行う。なお、実行制御サーバ100は、HDD以外の種類の不揮発性の記憶装置例えば、SSDなどを備えてもよく、複数の記憶装置を備えていてもよい。画像信号処理部104は、CPU101の命令に従って、実行制御サーバ100に接続されたディスプレイ11に画像を出力する。ディスプレイ11としては、例えば、CRTCathodeRayTubeディスプレイや液晶ディスプレイを用いることができる。入力信号処理部105は、実行制御サーバ100に接続された入力デバイス12から入力信号を取得し、CPU101に出力する。入力デバイス12としては、例えば、マウスやタッチパネルなどのポインティングデバイス、キーボードなどを用いることができる。ディスクドライブ106は、記録媒体13に記録されたプログラムやデータを読み取る駆動装置である。記録媒体13として、例えば、フレキシブルディスクFD:FlexibleDiskやHDDなどの磁気ディスク、CDCompactDiscやDVDDigitalVersatileDiscなどの光ディスク、光磁気ディスクMO:MagnetoOpticaldiskを使用できる。ディスクドライブ106は、例えば、CPU101の命令に従って、記録媒体13から読み取ったプログラムやデータをRAM102またはHDD103に格納する。通信部107は、ネットワーク10を介して他のサーバと通信を行う通信インタフェースである。通信部107は、有線通信インタフェースでもよいし、無線通信インタフェースでもよい。図4は、分散処理システムのソフトウェア例を示す図である。図4に示すユニットの一部または全部は、各サーバが実行するプログラムのモジュールであってもよい。また、図4に示すユニットの一部または全部は、FPGAFieldProgrammableGateArrayやASICApplicationSpecificIntegratedCircuitなどの電子回路であってもよい。データ処理サーバ900a,もデータ処理サーバ900と同様のユニットを用いて実装できる。ECサーバ20は、ECAPPApplication21を有する。ECAPP21は、電子商取引の機能を実現するアプリケーションである。ECサーバ20はWebサーバとして機能する。ECAPP21は、当該Webサーバ上のWebアプリケーションとして、ユーザが操作する端末装置から利用可能である。クライアント30は、ブラウザ31を有する。ブラウザ31は、実行制御サーバ100にWebアクセスするためのWebブラウザである実行制御サーバ100は、Webサーバとしても機能する。管理者は、ブラウザ31を操作して、実行制御サーバ100に処理依頼を行える。処理依頼と共に、処理内容を記述したソースプログラムを実行制御サーバ100に送信したり、処理対象とする入力情報あるいは入力情報の範囲を指定したりすることができる。なお、クライアント30から実行制御サーバ100への処理依頼は、例えば、CLICommandLineInterfaceを用いて行ってもよい。実行制御サーバ100は、実行制御部110を有する。実行制御部110は、分散処理の実行を制御する。実行制御部110は、クライアント30から受信した処理依頼に応じて、リコメンド情報の生成を開始する。実行制御部110は、ソースプログラムを処理解析サーバ200に送信して並列化可能な箇所の解析を依頼する。実行制御部110は、ソースプログラムのうち、並列化可能な箇所を指定して、処理変換サーバ300に処理変換を依頼する。実行制御部110は、変換結果に基づいて、Map処理やReduce処理をデータ処理サーバ900,900a,にアサインする。そして、実行制御部110は、分散処理により生成されたリコメンド情報をクライアント30に応答する。処理解析サーバ200は、処理解析部210を有する。処理解析部210は、実行制御部110から取得したソースプログラムを解析して、データの並列処理が可能な箇所を特定する。データの並列処理が可能な箇所は、例えばデータ並列性の有無により特定できる。データ並列性とは、1入力されたデータを分割して別々に処理した場合も分割しないで処理した場合と結果が同一になる、2一回の処理が他の処理前後の処理に依存しない、という特性である。データ並列性に関しては、例えば次の文献に記載されている。データ並列性に関する文献DavidCullerandJaswinderPalSinghParallelComputerArchitectureAHardwareSoftwareApproachAug1998処理解析部210は、例えば、ソースプログラム中に宣言的に記述された所定箇所を検出することで、データの並列処理が可能な箇所を特定できる。具体的には、MapReduceのフレームワークで型決めしたインタフェースを検出することが考えられる。この場合、処理解析部210は、ソースプログラム中から、所定のインタフェースMapperやReducerなどを実装したクラスをデータ並列処理が可能な箇所として特定する。あるいは、例えば、所定のクラスMapperやReducerなどを継承したクラスをデータ並列処理が可能な箇所として特定する。なお、並列化可能箇所が明示的に示されていない場合や、並列化可能箇所が明示的に指定されていない処理の場合、処理解析部210は並列化コンパイラを用いて並列化可能箇所を特定してもよい。並列化コンパイラによる並列化可能箇所の特定は、例えば次の文献に記載されている。並列化コンパイラに関する文献UtpalBanerjeeRudolfEigenmannAlexandruNicolauandDavidAPaduaAutomaticProgramParallelizationProcoftheIEEEVol81No2Feb1993処理変換サーバ300は、処理変換部310を有する。処理変換部310は、処理解析部210により特定された並列処理可能な箇所を、過去の処理結果を流用する処理に変換する。処理変換部310は、入力情報のうち、過去の処理結果を流用可能な部分をデータ分析サーバ400に特定させる。データ分析サーバ400は、データ分析部410を有する。データ分析部410は、入力情報のうち、過去の処理結果を流用可能な部分を特定して、処理変換部310に通知する。登録サーバ500は、処理結果登録部510を有する。処理結果登録部510は、データ処理サーバ900,900a,によるMap処理の結果やReduce処理の結果を取得して、中間情報管理サーバ700および結果情報管理サーバ800に格納する。入力情報管理サーバ600は、入力情報記憶部610を有する。入力情報記憶部610は、入力情報を記憶する。中間情報管理サーバ700は、中間情報記憶部710を有する。中間情報記憶部710は、中間情報を記憶する。結果情報管理サーバ800は、結果情報記憶部810を有する。結果情報記憶部810は、結果情報を記憶する。データ処理サーバ900は、データ処理部910を有する。データ処理部910は、実行制御部110によりMap処理をアサインされると、入力情報の一部を取得してMap処理を実行する。また、データ処理部910は、実行制御部110によりReduce処理をアサインされるとMap処理の結果に対してReduce処理を実行する。Map処理やReduce処理のロジックは、ソースプログラム内に記述される。図5は、分散処理の例を示す図である。第2の実施の形態の分散処理システムでは、リコメンド情報を生成するまでに、第1,第2フェーズの2回のフェーズで、MapReduceを利用する。第1,第2フェーズは、データ処理サーバ900,900a,により実行される。第1,第2フェーズの結果、ユーザ間の評価値の相関を得ることができる。ユーザ間の相関に基づいて、各ユーザが高評価すると予測される商品を抽出し、これをお勧め商品とする。第1フェーズは、ステップS1,S2,S3を含む。ステップS1は、第1のMap処理である。ステップS1では、csv形式のデータが入力される。当該入力データには、ユーザ名、商品を示す情報、評価した日付、評価値が含まれる。ステップS1では、入力データに対して、キーkeyとバリューvalueとのペアが生成される。具体的には、キーは商品を示す情報である。バリューは当該商品を評価したユーザ名、その評価値および評価した日付のセットである。ここで、処理結果登録部510は、ステップS1で生成されたキーバリューを取得し、中間情報として中間情報記憶部710に格納する。ステップS2は、第1のshufflesort処理である。ステップS2では、ステップS1の出力が入力される。ステップS2では、入力されたキーバリューをキーでソートして出力する。ステップS3は、第1のReduce処理である。ステップS3では、ステップS2の出力が入力となる。ステップS3では、入力されたデータをマージして新たなキーバリューを出力する。具体的には、キーは商品を示す情報である。バリューは商品ごとの全ユーザの評価値のセットである。第2フェーズは、ステップS4,S5,S6を含む。ステップS4は、第2のMap処理である。ステップS4では、ステップS3の出力が入力となる。ステップS4では、入力されたデータに基づいて、新たなキーバリューを出力する。具体的には、キーはあるユーザ名と別のユーザ名のセットである。バリューは当該各ユーザによる評価値のセットである。ステップS5は、第2のshufflesort処理である。ステップS5では、ステップS4の出力が入力される。ステップS5では、入力されたキーバリューをキーでソートして出力する。ステップS6は、第2のReduce処理である。ステップS6では、ステップS5の出力が入力となる。ステップS6では、入力されたキーバリューに基づいて、ユーザ間の相関係数を求め、これを評価の類似度Similarityuser1,user2とする。例えば、ユーザTaroおよびJiroの類似度SimilarityT,JTはTaroを、JはJiroを示すは、次の1式により求めることができる。ここで、CovT,Jは、ユーザTaroおよびJiroの評価値の共分散である。σTはTaroの評価値の標準偏差であり、σJはJiroの評価値の標準偏差である。また、TiはTaroの商品Itemに対する評価値であり、JiはJiroのItemに対する評価値である。T,Jの上付きバーはそれぞれTi,Jiの相加平均を示す。このように、第1,第2フェーズを実行して、各ユーザ間の相関係数を算出する。各ユーザ間の相関係数を用いて、例えば、Taroのある商品Itemに対する予測評価値RateT,itemTはTaroを示すを次の2式により求めることができる。ここで、Σは、当該Itemに対して計算対象の期間内に評価を行った全ユーザについての和をとるものである。userの上付きバーは当該期間内の全ユーザによる評価値の相加平均を示す。図6は、評価値テーブルの例を示す図である。評価値テーブル611は、入力情報記憶部610に記憶される。評価値テーブル611は、日付、商品IDIDentifierおよび評価値の項目を含む。日付の項目には、評価値が入力された日付が登録される。商品IDの項目には、評価対象の商品を示す情報が登録される。評価値の項目には、各ユーザによる評価値が登録される。例えば、2011年3月1日に、商品IDItem1の商品に対して、ユーザTaroが入力した評価値1が登録されている。なお、評価値が大きい程、高評価であるとする。なお、評価値テーブル611に登録されるデータは、csv形式など他のデータ形式で保持されてもよい。図7は、中間情報テーブルの例を示す図である。中間情報テーブル711は、中間情報記憶部710に記憶される。中間情報テーブル711は、入力データハッシュ値、データ日付および中間情報の項目を含む。入力データハッシュ値の項目には、ステップS1で入力されたcsv形式のデータのハッシュ値が登録される。データ日付の項目には、入力データに含まれる日付の範囲が登録される。中間情報の項目には、ステップS1で出力された中間情報が登録される。例えば、2011年3月1日~5日に入力された評価値を含む入力情報のハッシュ値123a5cに対して、中間情報データ名結果01-05が登録されている。図8は、第1のMap処理の入出力データ例を示す図である。データD11は、ステップS1第1のMap処理の入力データである。データD11は、評価値テーブル611から管理者が指定した期間の情報をcsv形式で抽出したものである。データD11には、ユーザ名、商品ID、評価日付および評価値の順で情報が記述されている。データD12は、ステップS1の出力データである。データD12は、データD11に含まれる商品IDをキーとしたキーバリューである。バリューは、ユーザ名、評価日付および評価値の各値を-ハイフンで結んだものである。第1のMap処理を記述した第1のMapクラスは、例えば、次のように記述できる。図9は、第1のMapクラスの例を示す図である。プログラム911は、ソースプログラムの一部である。プログラム911では、データD11から一行ずつレコードを取得し、,カンマで区切られた各値を抽出して、データD12を生成する処理が記述されている。第1のMapクラスCollaborativeFiltering_phase1_Mapは、Mapperインタフェースを実装したクラスである。第2の実施の形態の分散処理システムでは、Mapperインタフェースによる型の定義により、第1のMapクラスのメソッドmapによる処理のデータ並列性が保証されている。処理解析部210は、第1のMapクラスがMapperインタフェースを実装したものであることを検出することで、メソッドmapによる処理を並列処理可能な箇所として特定できる。図10は、第1のshufflesort処理の入出力データ例を示す図である。データD12は、ステップS1の出力データであり、ステップS2第1のshufflesort処理の入力データである。データD13は、ステップS2の出力データである。データD13は、データD12のキーバリューをキー商品IDでソートしたものである。図11は、第1のReduce処理の入出力データ例を示す図である。データD13は、ステップS2の出力データであり、ステップS3第1のReduce処理の入力データである。データD14は、ステップS3の出力データである。データD14は、データD13のキーバリューをキー商品IDでマージしたものであり、商品IDに対する全ユーザの評価値を集約したものである。例えば、キー商品IDに対するバリューとして、ユーザと評価値とのセットが評価を行ったユーザ分だけ抽出される。データD14は、商品ID,user1,user1の評価値,user2,user2の評価値,というフォーマットである。なお、同一ユーザが同一商品に複数回評価を行っている場合には、新しい方の評価値を採用する。図12は、第1のReduceクラスの例を示す図である。プログラム912は、ソースプログラムの一部である。プログラム912では、データD13から一行ずつレコードを取得し、バリューに含まれる-ハイフンで区切られた各値をキーごとに抽出して、データD14を生成する処理が記述されている。第1のReduceクラスCollaborativeFiltering_phase1_Reduceは、Reducerインタフェースを実装したクラスである。Reducerインタフェースも、Mapperインタフェースと同様に、その型の定義により、第1のReduceクラスのデータ並列性が保証されている。このため、第1のReduceクラスのメソッドReduceも並列処理可能な箇所である。図13は、第2のMap処理の入出力データ例を示す図である。データD21は、ステップS3の出力データであり、ステップS4第2のMap処理の入力データである。データD22は、ステップS4の出力データである。データD22は、データD21に基づいて生成された新たなキーバリューである。具体的には、データD21の各行について、評価したユーザの組をキーとして抽出し、抽出したユーザによる評価値の組をバリューとしたものである。データD22では、ユーザ名を-ハイフンで結んだ値をキーとしている例えば、Taro-Jiro。また、各ユーザによる評価値を,カンマで区切った値をバリューとしている例えば、1,4。図14は、第2のshufflesort処理の入出力データ例を示す図である。データD22は、ステップS4の出力データであり、ステップS5第2のshufflesort処理の入力データである。データD23は、ステップS5の出力データである。データD23は、データD22のキーバリューをキーユーザ名の組でソートしたものである。図15は、第2のReduce処理の入出力データ例を示す図である。データD23は、ステップS5の出力データであり、ステップS6第2のReduce処理の入力データである。データD24は、ステップS6の出力データであり、データD23に基づいて生成されるデータである。データD24の各行は、データD23に基づいて1式により算出されたユーザ間の類似度相関係数である。図16は、リコメンド情報の例を示す図である。データD24は、ステップS6の出力データである。データD25は、ユーザ間の類似度を用いて2式により算出された各ユーザの各商品に対する予測評価値の一覧であり、お勧め商品を抽出するためのリコメンド情報である。例えば、ユーザTaroに対するお勧め商品を3つ抽出したい場合には、データD25からRateTaro,Itemのうち、大きい方から3つ選択して、該当する商品を特定すればよい。図17は、分散処理制御の例を示すフローチャートである。以下、図17に示す処理をステップ番号に沿って説明する。ステップS11実行制御部110は、処理対象のソースプログラムおよび処理対象の入力情報範囲日付範囲を含む処理依頼をクライアント30から受信する。例えば、管理者は、ブラウザ31を操作して、実行制御部110に送信する情報あるいは、当該情報のパスを入力できる。実行制御部110は、当該ソースプログラムにつき処理解析部210に処理解析を依頼する。ソースプログラムには、MapReduce処理の第1フェーズに関するプログラム911,912が含まれる。処理解析部210は、プログラム911に含まれる第1のMapクラスがMapperインタフェースを実装していることを検出すると、当該箇所をデータ並列化可能箇所として特定する。また、処理解析部210は、プログラム912に含まれる第1のReduceクラスがReducerインタフェースを実装していることを検出すると、当該箇所をデータ並列化可能箇所として特定する。処理解析部210は、特定した箇所を実行制御部110に応答する。実行制御部110は、ソースプログラムで記述された処理のうち、データ並列化可能箇所まで処理を進める。ステップS12実行制御部110は、データ並列化可能箇所のうち、中間情報の流用可能性のある箇所に関して、処理変換部310に処理変換を依頼する。中間情報の流用可能性のある第1のMap処理では、当該処理の過去の処理結果中間情報が中間情報記憶部710に記憶された中間情報テーブル711に蓄積されている。処理変換部310は、第1のReduce処理に対する第1のMap処理が中間情報を生成する処理であるため、当該第1のMap処理に関して、処理変換を開始する。処理変換部310は、データ分析部410に入力情報の分析を依頼する。データ分析部410は、中間情報テーブル711を参照して、入力情報のうち、処理済のデータ部分を特定する。詳細は後述する。ステップS13データ分析部410は、ステップS12で特定した処理済のデータ部分のうち、中間情報を流用可能なデータ部分を特定する。詳細は後述する。データ分析部410は、入力情報のうち、新たに計算すべきデータ部分、中間情報を流用できないデータ部分および中間情報を流用可能なデータ部分を示す情報を処理変換部310に応答する。ステップS14処理変換部310は、中間情報を流用可能なデータ部分については、中間情報テーブル711に登録された中間情報を流用した処理を行うように、第1フェーズの処理を変換する。詳細は後述する。処理変換部310は、変換結果を実行制御部110に応答する。ステップS15実行制御部110は、処理変換部310による変換結果に基づいて、データ処理サーバ900,900a,に分散処理を依頼する。例えば、データ処理部910は、第1のMap処理、第1のshufflesort処理、第1のReduce処理、第2のMap処理、とアサインされた処理を順次実行する。実行制御部110からデータ処理部910に処理対象として指定されるデータは、クライアント30から指定された入力情報のうち、新たに計算すべきデータ部分および処理済だが中間情報を流用できないデータ部分となる。また、データ処理部910は、各データ部分に対する第1のMap処理の結果、新規に得た中間情報を処理結果登録部510に出力する。更に、データ処理部910は、第1のMap処理において、中間情報を流用可能なデータ部分については、当該データ部分に対応する中間情報を中間情報テーブル711から取得して後続の処理を行う。そして、データ処理部910は、ユーザ間の類似度の情報データD24やリコメンド情報データD25を生成して、処理結果登録部510に出力する。データ処理サーバ900a,のデータ処理部も同様である。ステップS16処理結果登録部510は、データ処理サーバ900,900a,から取得した中間情報を1つにまとめて中間情報テーブル711に登録する。このとき、当該中間情報に含まれるデータの日付範囲も登録する。また、入力情報のうち当該日付範囲に対応する部分のデータのハッシュ値をインデックスとして登録する。また、処理結果登録部510は、データ処理サーバ900,900a,から取得したユーザ間の類似度の情報やリコメンド情報を結果情報記憶部810に格納する。このようにして、第2の実施の形態の分散処理システムでは、第1のMap処理において中間情報を流用可能な処理については、中間情報を流用した処理を行う。また、中間情報を流用できない箇所について新たに第1のMap処理を行うと、その結果得られた中間情報を中間情報テーブル711に登録する。再びリコメンド情報を生成するときにその結果を流用できるようにするためである。次に、上記ステップS12の処理を説明する。図18は、処理済データ特定処理の例を示すフローチャートである。以下、図18に示す処理をステップ番号に沿って説明する。ステップS21データ分析部410は、処理変換部310から指示された入力情報を読み込む。データ分析部410は、処理変換部310から入力情報を取得してもよいし、処理変換部310から指定された日付範囲の入力情報を、入力情報記憶部610に記憶された評価値テーブル611から取得してもよい。処理変換部310は、csv形式のデータとして入力情報を取得する。ステップS22データ分析部410は、入力情報から1行を読み込む。ステップS23データ分析部410は、読み込んだ行に含まれる日付を取得する。ステップS24データ分析部410は、中間情報テーブル711に含まれる中間情報の日付範囲データ日付の項目を取得する。ステップS25データ分析部410は、入力情報の当該行のデータが処理済であるか否かを判定する。処理済である場合、処理をステップS26に進める。処理済でない場合、処理をステップS27に進める。データ分析部410は、読み込んだ行のデータを処理済であるか否かを、当該行に含まれる日付が、中間情報テーブル711に含まれる何れかのデータ日付に含まれるか否かにより判定できる。含まれる場合、処理済である。含まれない場合、処理済でない。ステップS26データ分析部410は、ステップS22で読み込んだ行を処理済データとしてファイル出力する。そして、処理をステップS28に進める。ステップS27データ分析部410は、ステップS22で読み込んだ行を新規データとしてファイル出力する。そして、処理をステップS28に進める。ステップS28データ分析部410は、ステップS22で読み込んだ行が入力情報の最後の行であるか否かを判定する。最後の行でない場合、処理をステップS22に進める。最後の行である場合、処理を終了する。このようにして、データ分析部410は、入力情報を、新規に計算する部分および処理済の部分に分けてファイル出力する。データ分析部410は、処理済の部分について、中間情報を流用可能な部分を更に特定する。以下、図17のステップS13の処理を説明する。図19は、流用可能データ特定処理の例を示すフローチャートである。以下、図19に示す処理をステップ番号に沿って説明する。ステップS31データ分析部410は、処理済データ特定処理で出力した処理済データを読み込む。ステップS32データ分析部410は、処理済データから1行を読み込む。ステップS33データ分析部410は、読み込んだ1行に含まれる日付を集合aの要素として取得する。集合aは、評価値の登録日付の集合である。ステップS34データ分析部410は、ステップS32で読み込んだ行が処理済データの最後の行であるか否かを判定する。最後の行でない場合、処理をステップS32に進める。最後の行である場合、ステップS35に進める。例えば、処理済データに含まれる日付が、2011/03/03~2011/03/06の範囲の各日付であれば、集合a={2011/03/03,2011/03/04,2011/03/05,2011/03/06}となる。ステップS35データ分析部410は、中間情報テーブル711を参照して、中間情報に含まれる日付のべき集合listを作成する。例えば、中間情報テーブル711に対して、集合list={{φ},{2011/03/01~05},{2011/03/02~05},,{2011/03/01~05,2011/03/02~05},}を得る。ステップS36データ分析部410は、集合listから、空集合{φ}および重複する要素を除外する。ステップS37データ分析部410は、集合listの各要素の和集合が集合aの部分集合にならないものを除外する。ステップS38データ分析部410は、集合listのうち最大の和集合になる要素を流用可能箇所とする。ここで、最大の和集合とは、要素である日付の個数が最大となる和集合を示す。ステップS39データ分析部410は、集合listのうちから流用可能箇所として特定した要素の和集合に含まれる日付を処理変換部310に応答する。このようにして、データ分析部410は、処理済データのうちから、中間情報の流用が可能な部分を特定する。中間情報の流用ができない部分は、第1のMap処理の対象に含める部分となる。すなわち、入力情報のうち、図18のステップS27で新規データとしてファイル出力された部分および、中間情報の流用ができない部分が第1のMap処理の対象の部分である。次に、図17のステップS14の処理を説明する。図20は、処理変換処理の例を示すフローチャートである。以下、図20に示す処理をステップ番号に沿って説明する。ステップS41処理変換部310は、入力情報を読み込む。ステップS42処理変換部310は、データ分析部410の分析結果に基づいて、入力情報に新規に計算する部分があるか否かを判定する。新規に計算する部分がある場合、処理をステップS43に進める。新規に計算する部分がない場合、処理をステップS44に進める。ステップS43処理変換部310は、入力情報のうち、新規に計算する部分および計算のロジック第1のMap処理のmapメソッドなどを実行制御部110に通知して処理を依頼する。実行制御部110は、処理変換部310からの通知に基づき、データ処理サーバ900,900a,に当該データ部分に対する第1のMap処理を依頼する。ステップS44処理変換部310は、データ分析部410の分析結果に基づいて、入力情報のうち、処理済だが中間情報を流用できない部分があるか否かを判定する。処理済だが中間情報を流用できない部分がある場合、処理をステップS45に進める。処理済だが中間情報を流用できない部分がない場合、処理をステップS46に進める。ステップS45処理変換部310は、ステップS43と同様に、入力情報のうち、処理済だが中間情報を流用できない部分および計算のロジック第1のMap処理のmapメソッドなどを実行制御部110に通知して処理を依頼する。実行制御部110は、処理変換部310からの通知に基づき、データ処理サーバ900,900a,に当該データ部分に対する第1のMap処理を依頼する。ステップS46処理変換部310は、データ分析部410の分析結果に基づいて、入力情報のうち、中間情報を流用可能な部分があるか否かを判定する。中間情報を流用可能な部分がある場合、処理をステップS47に進める。中間情報を流用可能な部分がない場合、処理を終了する。ステップS47処理変換部310は、入力情報のうち、中間情報を流用可能な部分を実行制御部110に通知して、第1のMap処理の後続の処理を依頼する。実行制御部110は、処理変換部310からの通知に基づき、データ処理サーバ900,900a,に当該データ部分に対しては第1のMap処理を行わず、既存の中間情報を取得するよう依頼する。この場合、図17のステップS15において、データ処理部910は、通知された中間情報を中間情報記憶部710から取得して、後続の第1のshufflesort処理や第1のReduce処理を実行することになる。ステップS43,S44の何れかまたは両方を実行している場合には、データ処理部910は、通知された中間情報と、ステップS43,S44で新たに得た中間情報とに基づいて、後続の処理を実行する。このようにして、処理変換部310は、入力情報を3種類に区分して、実行制御部110に処理依頼する。実行制御部110は、データ処理サーバ900,900a,を制御して、入力情報の各区分に応じた分散処理を実行させる。図21は、分散処理制御の例を示すシーケンス図である。以下、図21に示す処理をステップ番号に沿って説明する。ステップST101クライアント30は、ソースプログラムや入力情報の日付範囲を指定した、リコメンド情報生成の処理依頼を実行制御サーバ100に送信する。実行制御サーバ100は、当該処理依頼を受信する。ステップST102実行制御サーバ100は、ソースプログラムを指定して、処理解析依頼を処理解析サーバ200に送信する。処理解析サーバ200は、当該処理解析依頼を受信する。ステップST103処理解析サーバ200は、解析結果を実行制御サーバ100に応答する。実行制御サーバ100は、処理解析結果を受信する。当該解析結果では、MapReduce処理の第1フェーズのステップS1第1のMap処理およびステップS3第1のReduce処理がデータ並列化可能箇所として特定されている。ステップST104実行制御サーバ100は、データ並列化可能箇所のうち、中間情報の流用可能性のある第1のMap処理について、処理の変換依頼を処理変換サーバ300に送信する。処理変換サーバ300は、当該変換依頼を受信する。ステップST105処理変換サーバ300は、入力情報および中間情報の分析依頼をデータ分析サーバ400に送信する。データ分析サーバ400は、当該分析依頼を受信する。ステップST106データ分析サーバ400は、中間情報管理サーバ700に中間情報を要求する。中間情報管理サーバ700は、当該要求を受信する。ステップST107中間情報管理サーバ700は、中間情報記憶部710に記憶された中間情報テーブル711を参照して、中間情報を応答する。データ分析サーバ400は、中間情報の応答を受信する。ステップST108データ分析サーバ400は、入力情報の日付範囲と中間情報とに基づいて、入力情報を、第1のMap処理の計算対象とする箇所および中間情報を流用可能な箇所に区分する。データ分析サーバ400は、その分析結果を処理変換サーバ300に応答する。処理変換サーバ300は、分析結果の応答を受信する。ステップST109処理変換サーバ300は、当該分析結果に基づいて、入力情報を分割し、中間情報を流用する処理に変換する。処理変換サーバ300は、実行制御サーバ100に変換結果を応答する。図22は、分散処理制御の例を示すシーケンス図である。以下、図22に示す処理をステップ番号に沿って説明する。なお、図22では、データ処理サーバ900のみを図示しているが、データ処理サーバ900a,も、データ処理サーバ900と同様の処理を行う。ステップST111実行制御サーバ100は、処理変換サーバ300による処理変換結果に応じて、第1のMap処理を含む処理依頼をデータ処理サーバ900,900a,に送信する。処理依頼では、第1のMap処理において、新たに計算対象とする入力情報の日付範囲が指定される。また、実行制御サーバ100は、入力情報のうち中間情報を流用可能な箇所を示す日付範囲の情報をデータ処理サーバ900,900a,に送信する。流用可能な箇所を示す情報は、データ処理サーバ900,900a,のうちの何れか1つに送信することが考えられる。また、流用可能な箇所を分割して、分割した各部分を複数のデータ処理サーバに通知してもよい。ステップST112データ処理サーバ900,900a,は、新たに計算対象とする入力情報を、入力情報管理サーバ600に要求する。日付範囲を指定することで、新たに計算対象とする箇所を要求する。入力情報管理サーバ600は、当該要求を受信する。ステップST113入力情報管理サーバ600は、入力情報記憶部610に記憶された入力情報を参照して、指定された日付範囲のデータを応答する。データ処理サーバ900は、当該応答を受信する。ステップST114データ処理サーバ900,900a,は、中間情報の流用可能な箇所について、中間情報管理サーバ700に中間情報を要求する。中間情報管理サーバ700は、当該要求を受信する。ステップST115中間情報管理サーバ700は、中間情報記憶部710に記憶された中間情報テーブル711を参照して、要求された日付範囲の中間情報を取得し応答する。データ処理サーバ900,900a,は、当該応答を受信する。ステップST116データ処理サーバ900,900a,は、MapReduceの第1フェーズを実行する。第1のMap処理では、中間情報を流用可能な箇所は処理対象外となり、当該箇所についてはステップST115で取得した中間情報を第1のMap処理の処理結果として、後続の処理を実行する。データ処理サーバ900,900a,は、入力情報のうち、第1のMap処理の計算を新たに行った箇所について中間情報を取得すると、登録サーバ500に送信する。登録サーバ500は、データ処理サーバ900,900a,から受信した中間情報を中間情報管理サーバ700に格納する。ステップST117データ処理サーバ900,900a,は、MapReduceの第2フェーズを実行する。そして、データ処理サーバ900,900a,は、ユーザ間の類似度の情報を得ると、当該情報に基づいて、リコメンド情報を生成する。更に、データ処理サーバ900,900a,は、これらの情報を登録サーバ500に送信する。登録サーバ500は、データ処理サーバ900,900a,から受信したユーザ間の類似度の情報やリコメンド情報を結果情報管理サーバ800に格納する。ステップST118データ処理サーバ900,900a,は、リコメンド情報を生成する処理が完了した旨を実行制御サーバ100に応答する。実行制御サーバ100は、当該応答を受信する。ステップST119実行制御サーバ100は、指示された処理が完了した旨をクライアント30に応答する。クライアント30は、当該応答を受信する。管理者は、クライアント30を操作して、結果情報管理サーバ800に格納されたリコメンド情報を参照し、ユーザの商品評価の傾向を把握することができる。また、ECサーバ20は、リコメンド情報に基づき、高評価が予測される商品をユーザごとに抽出し、当該商品をお勧め商品としてユーザが操作する端末装置のブラウザ上に表示させることもできる。このようにして、分散処理システムによる分散処理制御が行われる。以下、リコメンド情報の具体的な生成処理を説明する。ある日のリコメンド情報を生成するために、その日の直前の5日間の評価値のログを処理する場合を想定する。以下の説明において、1回目の処理は、2011年3月6日のリコメンド情報を生成する処理である。2回目の処理は、2011年3月7日のリコメンド情報を生成する処理である。3回目の処理は、2011年3月8日のリコメンド情報を生成する処理である。また、1回目の処理の直前では、中間情報テーブル711に何れの中間情報も登録されていないとする。図23は、1回目の処理の例を示す図である。1回目の処理の入力情報は、データD11である。データD11には、2011年3月1日に入力された評価値の情報以下、単に1日のログのように称するを含む1~5日のログが含まれる。データD11を入力して、第1のMap処理を実行した結果、データD12が出力される。データD12は、1~5日のログから新規に生成された中間情報である。登録サーバ500は、データ処理サーバ900,900a,による第1のMap処理の結果を収集してデータD12を生成し、データD12に含まれる日付範囲と共に中間情報テーブル711に登録する。なお、図23では、中間情報テーブル711のハッシュ値の項目の図示を省略している以下、同様。以降、第1のshufflesort処理、第1のReduce処理、第2のMap処理、と順次実行され、データD25が生成される。データD25は、1~5日のログから生成された6日のリコメンド情報である。図24は、2回目の処理の入力情報の例を示す図である。2回目の処理の入力情報は、データD11aである。データD11aには、2~6日のログが含まれる。この時点では、中間情報テーブル711には、1~5日のログに関する中間情報データD12が登録されている。データD11aの各日付のログは、図18,19で示した処理において次のように扱われる。12~5日のログは、既に処理されたことのある処理済データである。1~5日のログを含むデータD11を処理した結果、データD12を既に得ているからである。また、2~5日のログは、中間情報を流用可能な部分ではない。中間情報であるデータD12に含まれる日付の集合1~5日が、2~5日の日付の集合の部分集合ではないからである。26日のログは、新規に追加されたデータであり、新規に計算するデータである。6日のログの処理結果を含む中間情報は未取得だからである。図25は、2回目の処理の例を示す図である。第1のMap処理に入力されたデータD11aは次のように処理される。12~5日のログによって、中間情報であるデータD12aが生成される。登録サーバ500は、データ処理サーバ900,900a,による2~5日のログに対する第1のMap処理の結果を収集してデータD12aを生成し、データD12aに含まれる日付範囲と共に中間情報テーブル711に登録する。26日のログによって、中間情報であるデータD12bが生成される。登録サーバ500は、データ処理サーバ900,900a,による6日のログに対する第1のMap処理の結果を収集してデータD12bを生成し、データD12bに含まれる日付範囲と共に中間情報テーブル711に登録する。以降、第1のshufflesort処理、第1のReduce処理、第2のMap処理、と順次実行され、データD25aが生成される。データD25aは、2~6日のログから生成された7日のリコメンド情報である。図26は、3回目の処理の入力情報の例を示す図である。3回目の処理の入力情報は、データD11bである。データD11bには、3~7日のログが含まれる。この時点では、中間情報テーブル711には、1~5日のログに関する中間情報データD12、2~5日のログに関する中間情報データD12aおよび6日のログに関する中間情報データD12bが登録されている。データD11bの各日付のログは、図18,19で示した処理において次のように扱われる。13~5日のログは、既に処理されたことのある処理済データである。1~5日のログを含むデータD11を処理した結果、データD12を既に得ているからである。あるいは、2~5日のログを含むデータD11aを処理した結果、データD12aを既に得ているからである。また、3~5日のログは、中間情報を流用可能な部分ではない。中間情報であるデータD12,D12aに含まれる日付の集合1~5日およびデータD12,D12a,D12bに含まれる日付の和集合が、3~6日の日付の集合の部分集合ではないからである。26日のログは、既に処理されたことのある処理済データである。6日のログを含むデータD11aを処理した結果、データD12bを既に得ているからである。また、6日のログは、中間情報を流用可能な部分である。中間情報であるデータD12bに含まれる日付の集合6日が3~6日の日付の集合の部分集合だからである。37日のログは、新規に追加されたデータであり、新規に計算するデータである。データD12,D12a,D12bは、何れも7日のログの処理結果を含まないからである。図27は、3回目の処理の例を示す図である。第1のMap処理に入力されたデータD11bは次のように処理される。13~5日のログによって、中間情報であるデータD12cが生成される。登録サーバ500は、データ処理サーバ900,900a,による3~5日のログに対する第1のMap処理の結果を収集してデータD12cを生成し、データD12cに含まれる日付範囲と共に中間情報テーブル711に登録する。26日のログに関しては、中間情報テーブル711から流用可能な既存の中間情報であるデータD12bを取得する。37日のログによって、中間情報であるデータD12dが生成される。登録サーバ500は、データ処理サーバ900,900a,による7日のログに対する第1のMap処理の結果を収集してデータD12dを生成し、データD12dに含まれる日付範囲と共に中間情報テーブル711に登録する。以降、第1のshufflesort処理、第1のReduce処理、第2のMap処理、と順次実行され、データD25bが生成される。データD25bは、3~7日のログから生成された8日のリコメンド情報である。このように、第2の実施の形態の分散処理システムでは、入力情報のうち、過去に取得した中間情報を利用可能な部分を特定する。そして、入力情報の内、当該部分に関しては過去の中間情報を取得し、それ以外の部分を分散処理の対象とする。これにより、データ処理を効率化できる。具体的には、過去に取得した中間情報を流用可能な部分以外の部分をデータ処理サーバ900,900a,に分散処理させるので、入力情報の全てを処理せずに済む。既に実行された処理を重複して実行しないことで、処理効率を向上し、処理の高速化を図れる。また、データ処理サーバ900,900a,が処理するデータの量を低減できるので、データ処理サーバ900,900a,の負荷を軽減できる。なお、第2の実施の形態の分散処理システムでは、中間情報の管理単位を入力データの日付範囲の単位とする場合を例示したが他の単位で管理してもよい。例えば、中間情報の管理単位を日単位に統一してもよい。その場合、例えば、処理結果登録部510は、中間情報を日単位に分割して、中間情報記憶部710に記憶された中間情報テーブル711に登録する。このようすれば、中間情報の流用可能性を向上できる。また、第2の実施の形態の分散処理システムでは、評価値が入力された日付に基づいて、既処理の入力情報か、流用可能な中間情報があるか、を判断するものとしたが、他の指標により判断してもよい。例えば、年、月、週および時間帯などで判断してもよい。また、第2の実施の形態の分散処理システムでは、入力されるデータが日付時間によって区分できる場合を例示したが、これに限らない。例えば、入力されるデータが場所座標値などで区分できる場合にも上記方法を利用できる。より具体的には、ある領域内の各地点に関する入力データに対し、当該領域に含まれる一部領域の各地点のデータについて、過去の計算結果を流用できるかを第2の実施の形態と同様にして判断し、後段の処理を効率的に制御することもできる。更に、第2の実施の形態の分散処理システムでは、複数のサーバ装置で処理を分担するものとしたが、複数のサーバ装置の機能の少なくとも一部を一台のサーバ装置に集約してもよい。例えば、実行制御サーバ100、処理解析サーバ200、処理変換サーバ300およびデータ分析サーバ400の機能を一台のサーバ装置に実装してもよい。上記については単に本発明の原理を示すものである。更に、多数の変形や変更が当業者にとって可能であり、本発明は上記に示し、説明した正確な構成および応用例に限定されるものではなく、対応する全ての変形例および均等物は、添付の請求項およびその均等物による本発明の範囲とみなされる。"