(57)【要約】  ノード間でのデータの転送を削減する。  システムは、複数のノードを用いて、入力データに対して第１の処理を行い、第１の処理の結果に対して第２の処理を行う。セグメント（＃１）と過去に第１の処理が行われたセグメント（＃２）とを含む入力データが指定されると、システムは、ノード（２０）と、過去に行われたセグメント（＃２）に対する第１の処理の結果の少なくとも一部を記憶するノード（２０ａ）とを選択する。選択されたノード（２０）は、セグメント（＃１）に対して第１の処理を行う。選択されたノード（２０ａ）は、ノード（２０）から転送されるセグメント（＃１）に対する第１の処理の結果の少なくとも一部と、ノード（２０ａ）に記憶されたセグメント（＃２）に対する第１の処理の結果の少なくとも一部とに対して、第２の処理を行う。
2014020735.tif 000002
【特許請求の範囲】
【請求項１】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムが実行するデータ処理方法であって、
第１のセグメントと過去に前記第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、前記複数のノードの中から、第１のノードと、過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択し、
前記第１のノードを用いて、前記第１のセグメントに対して前記第１の処理を行い、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を転送し、
前記第２のノードを用いて、前記第１のノードから転送された前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部と、前記第２のノードに記憶された過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部とに対して、前記第２の処理を行う、
データ処理方法。
【請求項２】
選択される前記第２のノードは、過去に、前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を取得して前記第２の処理を行ったノードである、請求の範囲第１項記載のデータ処理方法。
【請求項３】
前記第２のノードは、過去に行われた前記第２のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、所定のキーを含むレコードを記憶しており、
前記第１のノードから前記第２のノードへは、前記第１のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、前記所定のキーを含むレコードを転送する、
請求の範囲第１項または第２項記載のデータ処理方法。
【請求項４】
前記第２のノードに、前記第１のノードから転送された前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を、前記第２の処理を行ってから少なくとも所定時間経過するまで消去せずに記憶しておく、請求の範囲第１項乃至第３項の何れか一項に記載のデータ処理方法。
【請求項５】
過去に指定された入力データに含まれるセグメントと、過去に行われた前記第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を、前記システムが備える記憶装置に格納して管理し、
前記第１および第２のノードは、前記記憶装置を参照して選択される、
請求の範囲第１項乃至第４項の何れか一項に記載のデータ処理方法。
【請求項６】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムの制御に用いられる情報処理装置であって、
入力データに含まれるセグメントと、過去に行われた前記第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を記憶する記憶部と、
第１のセグメントと過去に前記第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、前記記憶部を参照して、前記複数のノードの中から、第１のノードと、過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択し、
前記第１のノードに、前記第１のセグメントに対して前記第１の処理を行わせ、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部が転送されるよう制御し、
前記第２のノードに、前記第１のノードから転送された前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部と、前記第２のノードに記憶された過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部とに対して、前記第２の処理を行わせる制御部と、
を有する情報処理装置。
【請求項７】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムを制御するためのプログラムであって、
コンピュータに、
第１のセグメントと過去に前記第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、前記複数のノードの中から、第１のノードと、過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択し、
前記第１のノードに、前記第１のセグメントに対して前記第１の処理を行わせ、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部が転送されるよう制御し、
前記第２のノードに、前記第１のノードから転送された前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部と、前記第２のノードに記憶された過去に行われた前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部とに対して、前記第２の処理を行わせる、
処理を実行させるプログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明はデータ処理方法、情報処理装置およびプログラムに関する。
【背景技術】
【０００２】
現在、ネットワークに接続された複数のノード（例えば、複数のコンピュータ）を並列に動作させてデータ処理を行う並列データ処理システムが利用されている。並列データ処理システムは、例えば、データを分割して複数のノードに分散して割り当て、ノード間で独立にデータ処理を行うことで、データ処理を高速化する。並列データ処理システムは、例えば、サーバ装置のアクセスログの解析など、大量のデータを処理するときに利用される。並列データ処理システムは、いわゆるクラウドコンピューティングのシステムとして実現されることがある。並列データ処理システムに実行させるプログラムの作成を支援するため、ＭａｐＲｅｄｕｃｅなどのフレームワークが提案されている。【０００３】
ＭａｐＲｅｄｕｃｅで定義されるデータ処理は、ＭａｐタスクとＲｅｄｕｃｅタスクという２種類のタスクを含む。ＭａｐＲｅｄｕｃｅでは、まず、入力データが複数の部分集合に分割され、入力データの部分集合毎にＭａｐタスクが起動される。Ｍａｐタスク間には依存関係がないため、複数のＭａｐタスクは並列化可能である。次に、複数のＭａｐタスクが出力した中間データに含まれるレコードを、キーに応じて分類することで、中間データの集合が複数の部分集合に分割される。このとき、Ｍａｐタスクを行ったノードとＲｅｄｕｃｅタスクを行うノードの間で、中間データのレコードが転送され得る。そして、中間データの部分集合毎にＲｅｄｕｃｅタスクが起動される。Ｒｅｄｕｃｅタスクは、例えば、同じキーをもつ複数のレコードの値（バリュー）を集計する。Ｒｅｄｕｃｅタスク間には依存関係がないため、複数のＲｅｄｕｃｅタスクは並列化可能である。【０００４】
なお、複数のスレーブノードと複数のスイッチの間の接続関係を確認し、接続関係に基づいてスレーブノードをグルーピングし、１つのデータ集合から分割された複数のデータブロックが同じグループに配置されるよう制御する分散処理システムが提案されている。また、処理前後のデータ量の変化を確認し、データ量が減少する場合には分散度を高く設定しデータ量が増加する場合には分散度を低く設定することで、ノード間のトラフィックを考慮してデータ処理を高速化する分散処理システムが提案されている。
【先行技術文献】
【特許文献】
【０００５】
【特許文献１】特開２０１０−２４４４６９号公報
【特許文献２】特開２０１０−２４４４７０号公報
【非特許文献】
【０００６】
【非特許文献１】Jeffrey Dean and Sanjay Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters", Proc. of the 6th Symposium on Operating Systems Design and Implementation, pp.137-150, Dec. 2004
【発明の概要】
【発明が解決しようとする課題】
【０００７】
上記のように、複数のノードを用いて、入力データに対して第１段階の処理を行い、第１段階の処理の結果に対して第２段階の処理を行う情報処理システムが考えられる。ここで、今回処理する入力データの中に、過去に処理した入力データと共通する部分が含まれている場合には、その共通部分に対応する過去の第１段階の処理の結果を再利用できることが好ましい。しかし、再利用しようとする第１段階の処理の結果がどこに保存されているかを考慮せずにデータ処理を開始してしまうと、第２段階の処理を行うノード宛てのデータ転送が多く発生し、通信のオーバヘッドが大きくなるという問題がある。【０００８】
一側面では、本発明は、ノード間でのデータの転送を削減できるデータ処理方法、情報処理装置およびプログラムを提供することを目的とする。
【課題を解決するための手段】
【０００９】
一側面では、複数のノードを用いて、入力データに対して第１の処理を行い、第１の処理の結果に対して第２の処理を行うシステムが実行するデータ処理方法が提供される。第１のセグメントと過去に第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、複数のノードの中から、第１のノードと、過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択する。第１のノードを用いて、第１のセグメントに対して第１の処理を行い、第１のノードから第２のノードに、第１のセグメントに対する第１の処理の結果の少なくとも一部を転送する。第２のノードを用いて、第１のノードから転送された第１のセグメントに対する第１の処理の結果の少なくとも一部と、第２のノードに記憶された過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部とに対して、第２の処理を行う。【００１０】
また、一側面では、複数のノードを用いて、入力データに対して第１の処理を行い、第１の処理の結果に対して第２の処理を行うシステムの制御に用いられる、記憶部と制御部とを有する情報処理装置が提供される。記憶部は、入力データに含まれるセグメントと、過去に行われた第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を記憶する。制御部は、第１のセグメントと過去に第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、記憶部を参照して、複数のノードの中から、第１のノードと、過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択する。制御部は、第１のノードに、第１のセグメントに対して第１の処理を行わせ、第１のノードから第２のノードに、第１のセグメントに対する第１の処理の結果の少なくとも一部が転送されるよう制御する。制御部は、第２のノードに、第１のノードから転送された第１のセグメントに対する第１の処理の結果の少なくとも一部と、第２のノードに記憶された過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部とに対して、第２の処理を行わせる。【００１１】
また、一側面では、複数のノードを用いて、入力データに対して第１の処理を行い、第１の処理の結果に対して第２の処理を行うシステムを制御するためのプログラムが提供される。プログラムを実行するコンピュータは、第１のセグメントと過去に第１の処理が行われた第２のセグメントとを含む入力データが指定されたとき、複数のノードの中から、第１のノードと、過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部を記憶する第２のノードとを選択する。第１のノードに、第１のセグメントに対して第１の処理を行わせ、第１のノードから第２のノードに、第１のセグメントに対する第１の処理の結果の少なくとも一部が転送されるよう制御する。第２のノードに、第１のノードから転送された第１のセグメントに対する第１の処理の結果の少なくとも一部と、第２のノードに記憶された過去に行われた第２のセグメントに対する第１の処理の結果の少なくとも一部とに対して、第２の処理を行わせる。
【発明の効果】
【００１２】
一側面では、ノード間でのデータの転送を削減することができる。
本発明の上記および他の目的、特徴および利点は本発明の例として好ましい実施の形態を表す添付の図面と関連した以下の説明により明らかになるであろう。
【図面の簡単な説明】
【００１３】
【図１】第１の実施の形態の情報処理システムを示す図である。
【図２】第２の実施の形態の情報処理システムを示す図である。
【図３】マスタノードのハードウェア例を示すブロック図である。
【図４】ＭａｐＲｅｄｕｃｅ処理の流れの第１の例を示す図である。
【図５】ＭａｐＲｅｄｕｃｅ処理の流れの第２の例を示す図である。
【図６】マスタノードの機能例を示すブロック図である。
【図７】スレーブノードの機能例を示すブロック図である。
【図８】ジョブリストの例を示す図である。
【図９】タスクリストの例を示す図である。
【図１０】Ｍａｐ管理テーブルとＲｅｄｕｃｅ管理テーブルの例を示す図である。
【図１１】スレーブノードへ送信するＭａｐタスク通知の例を示す図である。
【図１２】マスタ制御の手順例を示すフローチャートである。
【図１３】Ｍａｐ情報補完の手順例を示すフローチャートである。
【図１４】Ｒｅｄｕｃｅ情報補完の手順例を示すフローチャートである。
【図１５】タスク完了処理の手順例を示すフローチャートである。
【図１６】タスク割当の手順例を示すフローチャートである。
【図１７】スレーブ制御の手順例を示すフローチャートである。
【図１８】中間データ取得の手順例を示すフローチャートである。
【図１９】管理テーブル更新の手順例を示すフローチャートである。
【図２０】ＭａｐＲｅｄｕｃｅ処理のシーケンス例を示す図である。
【発明を実施するための形態】
【００１４】
以下、本実施の形態を図面を参照して説明する。
［第１の実施の形態］
図１は、第１の実施の形態の情報処理システムを示す図である。第１の実施の形態の情報処理システムは、複数のノードを用いて、入力データに対して第１の処理を行い、第１の処理の結果に対して第２の処理を行う。並列データ処理のフレームワークであるＭａｐＲｅｄｕｃｅを利用する場合、Ｍａｐタスクの処理が第１の処理の一例であり、Ｒｅｄｕｃｅタスクの処理が第２の処理の一例である。この情報処理システムは、情報処理装置１０と、ノード２０，２０ａを含む複数のノードとを備える。情報処理装置１０および複数のノードは、有線ＬＡＮ（Local Area Network）などのネットワークに接続されている。
【００１５】
情報処理装置１０は、複数のノードに第１および第２の処理を割り当てる管理用のコンピュータである。情報処理装置１０を、マスタノードと呼んでもよい。情報処理装置１０は、記憶部１１および制御部１２を有する。記憶部１１は、過去に処理された入力データに含まれるセグメントと、過去に行われた第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を記憶する。制御部１２は、入力データが指定されると、記憶部１１に記憶された情報を参照して、再利用できる第１の処理の結果を判定し、複数のノードの中から、第１の処理を行うノードと第２の処理を行うノードを選択する。
【００１６】
ノード２０，２０ａを含む複数のノードそれぞれは、情報処理装置１０からの指示に応じて、第１および第２の処理の少なくとも一方を実行するコンピュータである。各ノードを、スレーブノードと呼んでもよい。ノード２０は演算部２１を有し、ノード２０ａは演算部２１ａおよび記憶部２２ａを有する。演算部２１，２１ａは、第１の処理または第２の処理を行う。例えば、演算部２１が第１の処理を行い、演算部２１ａが演算部２１による第１の処理の結果を取得して第２の処理を行う。記憶部２２ａは、過去に行われた第１の処理の結果の少なくとも一部を記憶する。ノード２０も記憶部を備えてよい。
【００１７】
記憶部１１，２２ａは、ＲＡＭ（Random Access Memory）などの揮発性メモリでもよいし、ＨＤＤ（Hard Disk Drive）やフラッシュメモリなどの不揮発性記憶装置でもよい。制御部１２および演算部２１，２１ａは、ＣＰＵ（Central Processing Unit）やＤＳＰ（Digital Signal Processor）などのプロセッサでもよいし、ＡＳＩＣ（Application Specific Integrated Circuit）やＦＰＧＡ（Field Programmable Gate Array）などのその他の電子回路であってもよい。プロセッサは、例えば、メモリに記憶されたプログラムを実行する。プロセッサは、プログラムの命令を実行するための演算器やレジスタの他に、データ処理のための専用の電子回路を含んでいてもよい。
【００１８】
ここで、セグメント＃１，＃２を含む複数のセグメントに分割可能な入力データが指定された場合を考える。セグメント＃２は、過去に第１の処理が行われたことのある、入力データの部分集合である。セグメント＃１は、過去に第１の処理が行われたことのない、入力データの部分集合であってもよい。また、記憶部２２ａに、セグメント＃２に対する第１の処理の結果の少なくとも一部（結果＃１−２）が記憶されているとする。
【００１９】
この場合、制御部１２は、複数のノードの中からノード２０（第１のノード）を選択する。また、制御部１２は、記憶部１１に記憶された情報を参照して、複数のノードの中から、結果＃１−２を記憶するノード２０ａ（第２のノード）を検索して選択する。制御部１２は、選択したノード２０に、セグメント＃１に対する第１の処理を行うよう指示し、選択したノード２０ａに、第２の処理を行うよう指示する。セグメント＃２に対する第１の処理は、結果＃１−２を再利用することで省略し得る。
【００２０】
すると、演算部２１は、セグメント＃１に対して第１の処理を行う。セグメント＃１に対する第１の処理の結果の少なくとも一部（結果＃１−１）が、ノード２０からノード２０ａに転送される。演算部２１ａは、ノード２０から転送された結果＃１−１と記憶部２２ａに記憶された結果＃１−２とをマージして第２の処理を行う。
【００２１】
なお、記憶部２２ａに記憶される結果＃１−２は、セグメント＃２に対する第１の処理の結果に含まれるレコードのうち、所定のキーをもつレコードの集合であってもよい。また、ノード２０からノード２０ａに転送される結果＃１−１は、セグメント＃１に対する第２の処理の結果に含まれるレコードのうち、所定のキーをもつレコードの集合であってもよい。第２の処理では、例えば、同じキーをもつ複数のレコードの値（バリュー）が集計されて、当該キーに関する第２の処理の結果（結果＃２）が生成される。また、ノード２０ａは、過去に結果＃１−２に対して第２の処理を行ったノードであってもよい。ノード２０ａは、ノード２０から受信した結果＃１−１を記憶部２２ａに格納してもよい。
【００２２】
第１の実施の形態の情報処理システムによれば、過去に行われたセグメント＃２に対する第１の処理の結果の少なくとも一部が再利用され、セグメント＃２に対する第１の処理が省略され得る。よって、データ処理の計算量を削減できる。また、セグメント＃２に対する第１の処理の結果の少なくとも一部を記憶しているノード２０ａに、第２の処理が割り当てられる。よって、再利用する第１の処理の結果の転送を削減でき、データ処理を効率化できると共に、ネットワークの負荷を低減することができる。
【００２３】
［第２の実施の形態］
図２は、第２の実施の形態の情報処理システムを示す図である。第２の実施の形態の情報処理システムは、ＭａｐＲｅｄｕｃｅを利用してデータ処理を並列化する。ＭａｐＲｅｄｕｃｅを実装したソフトウェアとしては、例えば、Ｈａｄｏｏｐが挙げられる。この情報処理システムは、業務サーバ４１、データベース（ＤＢ：Database）サーバ４２、管理ＤＢサーバ４３、端末装置４４、マスタノード１００およびスレーブノード２００，２００ａ，２００ｂ，２００ｃを含む。上記の各装置はネットワーク３０に接続されている。
【００２４】
業務サーバ４１は、電子商取引などの業務に用いられるサーバコンピュータである。業務サーバ４１は、ネットワーク３０または他のネットワークを介して、ユーザが操作するクライアントコンピュータ（図示せず）からアクセスを受け付け、アプリケーションソフトウェアにより所定の情報処理を実行する。そして、業務サーバ４１は、情報処理の実行状況を示すログデータを生成し、ＤＢサーバ４２にログデータを格納する。
【００２５】
ＤＢサーバ４２および管理ＤＢサーバ４３は、データを記憶し、他のコンピュータからのアクセスに応じてデータの検索や更新を行うサーバコンピュータである。ＤＢサーバ４２に記憶されたデータ（例えば、業務サーバ４１が生成したログデータ）は、スレーブノード２００，２００ａ，２００ｂ，２００ｃにより解析される入力データとして使用され得る。管理ＤＢサーバ４３には、スレーブノード２００，２００ａ，２００ｂ，２００ｃにより実行されるデータ解析を制御するための管理情報が記憶される。なお、ＤＢサーバ４２と管理ＤＢサーバ４３とを統合して１つのＤＢサーバとしてもよい。
【００２６】
端末装置４４は、ユーザ（情報処理システムの管理者を含む）が操作するクライアントコンピュータである。端末装置４４は、ユーザの操作に応じて、ＤＢサーバ４２やスレーブノード２００，２００ａ，２００ｂ，２００ｃに記憶されたデータの解析を開始するためのコマンドを、マスタノード１００に送信する。コマンドでは、解析するデータが含まれるファイルや、処理手順を定義したプログラムのファイルが指定される。プログラムのファイルは、例えば、端末装置４４からマスタノード１００にアップロードしておく。
【００２７】
マスタノード１００は、スレーブノード２００，２００ａ，２００ｂ，２００ｃを制御して並列データ処理を実現するサーバコンピュータである。マスタノード１００は、端末装置４４からコマンドを受け付けると、入力データを複数のセグメントに分割し、入力データのセグメントを処理して中間データを生成する複数のＭａｐタスクを定義する。また、マスタノード１００は、中間データを集計する１またはそれ以上のＲｅｄｕｃｅタスクを定義する。そして、マスタノード１００は、ＭａｐタスクおよびＲｅｄｕｃｅタスクを、スレーブノード２００，２００ａ，２００ｂ，２００ｃに分散して割り当てる。なお、コマンドで指定されたプログラムのファイルは、例えば、マスタノード１００によって、スレーブノード２００，２００ａ，２００ｂ，２００ｃに配置される。
【００２８】
スレーブノード２００，２００ａ，２００ｂ，２００ｃは、マスタノード１００からの指示に応じて、ＭａｐタスクとＲｅｄｕｃｅタスクの少なくとも一方を実行するサーバコンピュータである。１つのスレーブノードがＭａｐタスクとＲｅｄｕｃｅタスクの両方を実行することもある。複数のＭａｐタスクは互いに独立しているため並列に実行でき、複数のＲｅｄｕｃｅタスクは互いに独立しているため並列に実行できる。Ｍａｐタスクを行うノードからＲｅｄｕｃｅタスクを行うノードに、中間データが転送されることがある。
【００２９】
なお、マスタノード１００は、第１の実施の形態で説明した情報処理装置１０の一例である。また、スレーブノード２００，２００ａ，２００ｂ，２００ｃそれぞれは、第１の実施の形態で説明したノード２０またはノード２０ａの一例である。
【００３０】
図３は、マスタノードのハードウェア例を示すブロック図である。マスタノード１００は、ＣＰＵ１０１、ＲＡＭ１０２、ＨＤＤ１０３、画像信号処理部１０４、入力信号処理部１０５、ディスクドライブ１０６および通信インタフェース１０７を有する。上記の各ユニットは、マスタノード１００が備えるバス１０８に接続されている。
【００３１】
ＣＰＵ１０１は、プログラムの命令を実行する演算器を含むプロセッサである。ＣＰＵ１０１は、ＨＤＤ１０３に記憶されているプログラムやデータの少なくとも一部をＲＡＭ１０２にロードし、プログラムを実行する。なお、ＣＰＵ１０１は複数のプロセッサコアを備えてもよく、マスタノード１００は複数のプロセッサを備えてもよく、以下で説明する処理を複数のプロセッサまたはプロセッサコアを用いて並列実行してもよい。
【００３２】
ＲＡＭ１０２は、ＣＰＵ１０１が実行するプログラムや計算に用いられるデータを一時的に記憶する揮発性メモリである。なお、マスタノード１００は、ＲＡＭ以外の種類のメモリを備えてもよく、複数の揮発性メモリを備えてもよい。
【００３３】
ＨＤＤ１０３は、ＯＳ（Operating System）やファームウェアやアプリケーションソフトウェアなどのソフトウェアのプログラムおよびデータを記憶する不揮発性の記憶装置である。なお、マスタノード１００は、フラッシュメモリやＳＳＤ（Solid State Drive）などの他の種類の記憶装置を備えてもよく、複数の不揮発性の記憶装置を備えてもよい。
【００３４】
画像信号処理部１０４は、ＣＰＵ１０１からの命令に従って、マスタノード１００に接続されたディスプレイ５１に画像を出力する。ディスプレイ５１としては、ＣＲＴ（Cathode Ray Tube）ディスプレイや液晶ディスプレイなどを用いることができる。
【００３５】
入力信号処理部１０５は、マスタノード１００に接続された入力デバイス５２から入力信号を取得し、ＣＰＵ１０１に通知する。入力デバイス５２としては、マウスやタッチパネルなどのポインティングデバイス、キーボードなどを用いることができる。
【００３６】
ディスクドライブ１０６は、記録媒体５３に記録されたプログラムやデータを読み取る駆動装置である。記録媒体５３として、例えば、フレキシブルディスク（ＦＤ：Flexible Disk）やＨＤＤなどの磁気ディスク、ＣＤ（Compact Disc）やＤＶＤ（Digital Versatile Disc）などの光ディスク、光磁気ディスク（ＭＯ：Magneto-Optical disk）を使用できる。ディスクドライブ１０６は、ＣＰＵ１０１からの命令に従って、記録媒体５３から読み取ったプログラムやデータをＲＡＭ１０２またはＨＤＤ１０３に格納する。
【００３７】
通信インタフェース１０７は、ネットワーク３０を介して他のコンピュータ（例えば、端末装置４４やスレーブノード２００，２００ａ，２００ｂ，２００ｃ）と通信を行うインタフェースである。通信インタフェース１０７は、有線網に接続する有線インタフェースでもよいし、無線網に接続する無線インタフェースでもよい。
【００３８】
ただし、マスタノード１００は、ディスクドライブ１０６を備えていなくてもよく、専ら他のコンピュータからアクセスされる場合には、画像信号処理部１０４や入力信号処理部１０５を備えていなくてもよい。業務サーバ４１、ＤＢサーバ４２、管理ＤＢサーバ４３、端末装置４４およびスレーブノード２００，２００ａ，２００ｂ，２００ｃも、マスタノード１００と同様のハードウェアを用いて実現できる。なお、ＣＰＵ１０１は、第１の実施の形態で説明した制御部１２の一例であり、ＲＡＭ１０２またはＨＤＤ１０３は、第１の実施の形態で説明した記憶部１１の一例である。
【００３９】
図４は、ＭａｐＲｅｄｕｃｅ処理の流れの第１の例を示す図である。ＭａｐＲｅｄｕｃｅで規定されるデータ処理手順には、入力データの分割、Ｍａｐフェーズ、中間データの分類とマージ（Ｓｈｕｆｆｌｅ＆Ｓｏｒｔ）およびＲｅｄｕｃｅフェーズが含まれる。
【００４０】
入力データの分割では、入力データが複数のセグメントに分割される。図４の例では、入力データとしての文字列がセグメント＃１〜＃３に分割されている。
Ｍａｐフェーズでは、入力データのセグメント毎にＭａｐタスクが起動される。図４の例では、セグメント＃１を処理するＭａｐタスク＃１−１と、セグメント＃２を処理するＭａｐタスク＃１−２と、セグメント＃３を処理するＭａｐタスク＃１−３とが起動されている。複数のＭａｐタスクは互いに独立に実行される。Ｍａｐタスクで行われるＭａｐ処理の手順は、プログラムによってユーザが定義することができる。図４の例では、Ｍａｐ処理として、各単語が文字列の中に何回出現するかをカウントしている。各Ｍａｐタスクは、Ｍａｐ処理の結果として、１またはそれ以上のレコードを含む中間データを生成する。中間データのレコードは、キーとバリューを組にしたキー・バリュー形式で表される。図４の例では、各レコードは、単語を表すキーと、その単語の出現数を表すバリューとを含む。入力データのセグメントと中間データとは、１対１に対応付けることができる。
【００４１】
Ｓｈｕｆｆｌｅ＆Ｓｏｒｔでは、複数のＭａｐタスクで生成された中間データに含まれるレコードが、キーに応じて分類されてマージされる。すなわち、レコードのキーから当該レコードを担当するＲｅｄｕｃｅタスクが判定され、同じキーをもつレコードが集められてマージされる。キーからＲｅｄｕｃｅタスクを判定する方法としては、各Ｒｅｄｕｃｅタスクにハッシュ値としての番号を割り当て、キーのハッシュ値を算出して判定する方法が考えられる。ただし、キーからＲｅｄｕｃｅタスクを判定する関数をユーザが定義してもよい。図４の例では、Ａｐｐｌｅ，Ｈｅｌｌｏをキーにもつレコードが１箇所に集められ、また、ｉｓ，Ｒｅｄをキーにもつレコードが１箇所に集められている。レコードのマージでは、同じキーをもつレコードのバリューが、リスト形式に纏められる。
【００４２】
Ｒｅｄｕｃｅフェーズでは、Ｓｈｕｆｆｌｅ＆Ｓｏｒｔを通じて形成された中間データのセグメント（同じＲｅｄｕｃｅタスクが担当するレコードの集合）毎に、Ｒｅｄｕｃｅタスクが起動される。図４の例では、Ａｐｐｌｅ，Ｈｅｌｌｏをキーにもつレコードを処理するＲｅｄｕｃｅタスク＃１−１と、ｉｓ，Ｒｅｄをキーにもつレコードを処理するＲｅｄｕｃｅタスク＃１−２とが起動されている。複数のＲｅｄｕｃｅタスクは互いに独立に実行される。Ｒｅｄｕｃｅタスクで行われるＲｅｄｕｃｅ処理の手順は、プログラムによってユーザが定義することができる。図４の例では、Ｒｅｄｕｃｅ処理として、リスト形式で列挙された単語の出現数を合計している。各Ｒｅｄｕｃｅタスクは、Ｒｅｄｕｃｅ処理の結果として、キー・バリュー形式のレコードを含む出力データを生成する。
【００４３】
ＭａｐタスクおよびＲｅｄｕｃｅタスクは、スレーブノード２００，２００ａ，２００ｂ，２００ｃに分散して割り当てることができる。例えば、Ｍａｐタスク＃１−２がスレーブノード２００に割り当てられ、Ｒｅｄｕｃｅタスク＃１−１がスレーブノード２００ａに割り当てられる。この場合、Ｍａｐタスク＃１−２が生成した中間データに含まれるレコードのうち、Ａｐｐｌｅ，Ｈｅｌｌｏをキーにもつレコードが、スレーブノード２００からスレーブノード２００ａに転送されることになる。
【００４４】
図５は、ＭａｐＲｅｄｕｃｅ処理の流れの第２の例を示す図である。ここでは、図４に示したＭａｐＲｅｄｕｃｅ処理の後に、図５に示すＭａｐＲｅｄｕｃｅ処理が実行される場合を考える。図５の例では、入力データがセグメント＃２〜＃４に分割されている。セグメント＃２，＃３は、図４に示したものと同一である。すなわち、図５で処理される入力データの一部は、図４で処理された入力データと重複している。
【００４５】
Ｍａｐフェーズでは、セグメント＃２を処理するＭａｐタスク＃２−１と、セグメント＃３を処理するＭａｐタスク＃２−２と、セグメント＃４を処理するＭａｐタスク＃２−３とが起動されている。Ｒｅｄｕｃｅフェーズでは、図４の場合と同様に、Ａｐｐｌｅ，Ｈｅｌｌｏをキーにもつレコードを処理するＲｅｄｕｃｅタスク＃２−１と、ｉｓ，Ｒｅｄをキーにもつレコードを処理するＲｅｄｕｃｅタスク＃２−２とが起動されている。
【００４６】
ここで、図５の入力データには、セグメント＃１が含まれておらずセグメント＃４が含まれている点で、図４の入力データと異なる。このため、Ａｐｐｌｅ，Ｈｅｌｌｏの出現数を示すＲｅｄｕｃｅタスク＃２−１の結果は、図４に示したＲｅｄｕｃｅタスク＃１−１の結果と異なる。また、ｉｓ，Ｒｅｄの出現数を示すＲｅｄｕｃｅタスク＃２−２の結果は、図４に示したＲｅｄｕｃｅタスク＃１−２の結果と異なる。
【００４７】
一方、入力データのセグメントとＭａｐタスクの結果である中間データとは、１対１に対応する。このため、セグメント＃２を処理するＭａｐタスク＃２−１の結果は、図４に示したＭａｐタスク＃１−２の結果と同じである。また、セグメント＃３を処理するＭａｐタスク＃２−２の結果は、図４に示したＭａｐタスク＃１−３の結果と同じである。すなわち、セグメント＃２，＃３に対応する中間データについては、再利用が可能である。
【００４８】
ここで、Ｒｅｄｕｃｅタスク＃１−１を実行したノードに、Ｍａｐタスク＃１−２，＃１−３から収集した中間データを保存しておき、そのノードにＲｅｄｕｃｅタスク＃２−１を実行させれば、中間データを再利用するにあたってノード間での中間データの転送を抑制できる。同様に、Ｒｅｄｕｃｅタスク＃１−２を実行したノードに、Ｍａｐタスク＃１−３から収集した中間データを保存しておき、そのノードにＲｅｄｕｃｅタスク＃２−２を実行させれば、ノード間での中間データの転送を抑制できる。そこで、マスタノード１００は、中間データを再利用できるようにし、中間データの転送が少なくなるようにスレーブノード２００，２００ａ，２００ｂ，２００ｃにＲｅｄｕｃｅタスクを割り振る。
【００４９】
図６は、マスタノードの機能例を示すブロック図である。マスタノード１００は、定義記憶部１１０、タスク情報記憶部１２０、再利用情報記憶部１３０、ジョブ発行部１４１、ジョブトラッカー１４２、ジョブ分割部１４３およびバックアップ部１４４を有する。定義記憶部１１０、タスク情報記憶部１２０および再利用情報記憶部１３０は、例えば、ＲＡＭ１０２またはＨＤＤ１０３に確保された記憶領域として実現される。ジョブ発行部１４１、ジョブトラッカー１４２、ジョブ分割部１４３およびバックアップ部１４４は、例えば、ＣＰＵ１０１に実行させるプログラムのモジュールとして実装される。
【００５０】
定義記憶部１１０は、Ｍａｐ定義１１１、Ｒｅｄｕｃｅ定義１１２および分割定義１１３を記憶する。Ｍａｐ定義１１１は、Ｍａｐ処理を定義する。Ｒｅｄｕｃｅ定義１１２は、Ｒｅｄｕｃｅ処理を定義する。分割定義１１３は、入力データの分割方法を定義する。Ｍａｐ定義１１１、Ｒｅｄｕｃｅ定義１１２および分割定義１１３は、例えば、プログラムのモジュール（オブジェクト指向プログラムのクラスなど）である。
【００５１】
タスク情報記憶部１２０は、ジョブリスト１２１、タスクリスト１２２および通知バッファ１２３を記憶する。ジョブリスト１２１は、ひと纏まりのＭａｐＲｅｄｕｃｅ処理を示すジョブの一覧を示す情報である。タスクリスト１２２は、ジョブ毎に定義されたＭａｐタスクとＲｅｄｕｃｅタスクの一覧を示す情報である。通知バッファ１２３は、マスタノード１００からスレーブノード２００，２００ａ，２００ｂ，２００ｃに送信する通知（メッセージ）を一時的に格納しておく記憶領域である。何れかのスレーブノードからハートビートとしての通知が受信されたときに、通知バッファ１２３に格納された当該スレーブノード宛ての通知が、応答として当該スレーブノードに送信される。
【００５２】
再利用情報記憶部１３０は、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２を記憶する。Ｍａｐ管理テーブル１３１は、過去にＭａｐタスクを実行したノードおよび当該ノードに保存されている中間データを示す情報を格納する。Ｒｅｄｕｃｅ管理テーブル１３２は、過去にＲｅｄｕｃｅタスクを実行したノードおよび当該ノードに保存されている中間データを示す情報を格納する。Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２に基づいて、過去に生成された中間データが再利用される。
【００５３】
ジョブ発行部１４１は、端末装置４４からコマンドを受け付けると、ジョブトラッカー１４２に、ＭａｐＲｅｄｕｃｅで使用するＭａｐ定義１１１、Ｒｅｄｕｃｅ定義１１２、分割定義１１３および入力データを指定して、新たなジョブを登録するよう要求する。また、ジョブ発行部１４１は、ジョブトラッカー１４２からジョブの完了が報告されると、端末装置４４に対してジョブ完了を示すメッセージを送信する。
【００５４】
ジョブトラッカー１４２は、ジョブおよびタスク（ＭａｐタスクとＲｅｄｕｃｅタスクを含む）を管理する。ジョブトラッカー１４２は、ジョブ発行部１４１から新たなジョブの登録を要求されると、ジョブ分割部１４３を呼び出すことで、入力データを複数のセグメントに分割する。そして、ジョブトラッカー１４２は、そのジョブを実現するためのＭａｐタスクとＲｅｄｕｃｅタスクを定義してタスクリスト１２２に登録すると共に、ジョブリスト１２１を更新する。このとき、ジョブトラッカー１４２は、Ｍａｐ管理テーブル１３１を参照して、中間データを再利用することで省略できるＭａｐタスクを判定する。
【００５５】
ＭａｐタスクおよびＲｅｄｕｃｅタスクが定義されると、ジョブトラッカー１４２は、スレーブノード２００，２００ａ，２００ｂ，２００ｃのリソースの空き状況に応じて、各タスク（省略されるＭａｐタスクを除く）を何れかのスレーブノードに割り当てる。このとき、ジョブトラッカー１４２は、Ｒｅｄｕｃｅ管理テーブル１３２に従って、各Ｒｅｄｕｃｅタスクを、そのＲｅｄｕｃｅタスクで再利用することのできるＲｅｄｕｃｅ用の中間データが保存されたスレーブノードに優先的に割り当てるようにする。ＭａｐタスクおよびＲｅｄｕｃｅタスクが完了すると、ジョブトラッカー１４２は、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２に中間データに関する情報を登録する。
【００５６】
なお、ジョブトラッカー１４２は、スレーブノード２００，２００ａ，２００ｂ，２００ｃに送信する通知を生成したときは、通知バッファ１２３に通知を格納する。ジョブトラッカー１４２は、何れかのスレーブノードからハートビートを受け付けると、ハートビートに対する応答として、通知バッファ１２３に格納されている当該スレーブノード宛ての通知を送信する。また、ジョブトラッカー１４２は、Ｍａｐタスクを何れかのスレーブノードに割り当てたときに、当該スレーブノードにＭａｐ定義１１１を配置してもよい。また、ジョブトラッカー１４２は、Ｒｅｄｕｃｅタスクを何れかのスレーブノードに割り当てたときに、当該スレーブノードにＲｅｄｕｃｅ定義１１２を配置してもよい。
【００５７】
ジョブ分割部１４３は、ジョブトラッカー１４２から呼び出されると、分割定義１１３に定義された分割方法に従って、入力データを複数のセグメントに分割する。入力データの中に過去にＭａｐ処理が行われた部分が含まれている場合、過去にＭａｐ処理が行われた部分とそれ以外の部分とが異なるセグメントに属するように分割することが好ましい。なお、指定される入力データは、ＤＢサーバ４２に記憶されていることもあるし、スレーブノード２００，２００ａ，２００ｂ，２００ｃに記憶されていることもある。
【００５８】
バックアップ部１４４は、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２を、ネットワーク３０を介して管理ＤＢサーバ４３にバックアップする。バックアップ部１４４によるバックアップは、定期的に行ってもよいし、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２が更新されたときに行ってもよい。
【００５９】
図７は、スレーブノードの機能例を示すブロック図である。スレーブノード２００は、Ｍａｐ結果記憶部２１１、Ｒｅｄｕｃｅ入力記憶部２１２、Ｒｅｄｕｃｅ結果記憶部２１３、タスクトラッカー２２１、Ｍａｐ実行部２２２およびＲｅｄｕｃｅ実行部２２３を有する。Ｍａｐ結果記憶部２１１、Ｒｅｄｕｃｅ入力記憶部２１２およびＲｅｄｕｃｅ結果記憶部２１３は、例えば、ＲＡＭまたはＨＤＤに確保された記憶領域として実現される。タスクトラッカー２２１、Ｍａｐ実行部２２２およびＲｅｄｕｃｅ実行部２２３は、例えば、ＣＰＵに実行させるプログラムのモジュールとして実装される。スレーブノード２００ａ，２００ｂ，２００ｃも、スレーブノード２００と同様の機能を有する。
【００６０】
Ｍａｐ結果記憶部２１１は、スレーブノード２００で実行されたＭａｐタスクの結果としての中間データを記憶する。Ｍａｐ結果記憶部２１１では、複数のＭａｐタスクの結果が、ディレクトリ分けされて管理される。ディレクトリのパス名は、例えば、／ジョブＩＤ／ＭａｐタスクのタスクＩＤ／ｏｕｔのように定義される。
【００６１】
Ｒｅｄｕｃｅ入力記憶部２１２は、スレーブノード２００がＲｅｄｕｃｅタスクを実行するにあたって、Ｍａｐタスクを実行したノードから収集された中間データを記憶する。Ｒｅｄｕｃｅ入力記憶部２１２では、複数のＲｅｄｕｃｅタスクに関する中間データが、ディレクトリ分けされて管理される。ディレクトリのパス名は、例えば、／ジョブＩＤ／ＲｅｄｕｃｅタスクのタスクＩＤ／ｉｎのように定義される。
【００６２】
Ｒｅｄｕｃｅ結果記憶部２１３は、スレーブノード２００で実行されたＲｅｄｕｃｅタスクの結果としての出力データを記憶する。Ｒｅｄｕｃｅ結果記憶部２１３に記憶された出力データは、以降に実行されるジョブの入力データとして利用することができる。
【００６３】
タスクトラッカー２２１は、スレーブノード２００に割り当てられたタスク（ＭａｐタスクとＲｅｄｕｃｅタスクを含む）を管理する。スレーブノード２００には、並列に実行可能なＭａｐタスクの上限数とＲｅｄｕｃｅタスクの上限数が設定されている。実行中のＭａｐタスクの数またはＲｅｄｕｃｅタスクの数が上限に達していない場合、タスクトラッカー２２１は、マスタノード１００にタスク要求通知を送信する。タスクトラッカー２２１は、タスク要求通知に応じてマスタノード１００からＭａｐタスクが割り当てられると、Ｍａｐ実行部２２２を呼び出し、タスク要求通知に応じてＲｅｄｕｃｅタスクが割り当てられると、Ｒｅｄｕｃｅ実行部２２３を呼び出す。何れかのタスクが完了すると、タスクトラッカー２２１は、スレーブノード２００にタスク完了通知を送信する。
【００６４】
また、タスクトラッカー２２１は、Ｍａｐタスクが完了した後、Ｒｅｄｕｃｅタスクを実行する他のスレーブノードから転送要求があると、Ｍａｐ結果記憶部２１１に記憶された中間データの少なくとも一部を送信する。また、タスクトラッカー２２１は、スレーブノード２００にＲｅｄｕｃｅタスクが割り当てられると、Ｍａｐタスクを実行した他のスレーブノードに転送要求を行い、受信した中間データをＲｅｄｕｃｅ入力記憶部２１２に格納する。タスクトラッカー２２１は、収集した中間データをマージする。
【００６５】
Ｍａｐ実行部２２２は、タスクトラッカー２２１から呼び出されると、Ｍａｐ定義１１１で定義されたＭａｐ処理を実行する。Ｍａｐ実行部２２２は、Ｍａｐタスクにより生成された中間データを、Ｍａｐ結果記憶部２１１に格納する。このとき、Ｍａｐ実行部２２２は、キー・バリュー形式の複数のレコードをキーに基づいてソートし、同じＲｅｄｕｃｅタスクに振り分けられるレコードの集合毎にファイルを作成する。ジョブＩＤとＭａｐタスクのタスクＩＤとによって特定されるディレクトリに、転送先となるＲｅｄｕｃｅタスクに応じた番号が付された１またはそれ以上のファイルが格納されることになる。
【００６６】
Ｒｅｄｕｃｅ実行部２２３は、タスクトラッカー２２１から呼び出されると、Ｒｅｄｕｃｅ定義１１２で定義されたＲｅｄｕｃｅ処理を実行する。Ｒｅｄｕｃｅ実行部２２３は、Ｒｅｄｕｃｅタスクにより生成された出力データを、Ｒｅｄｕｃｅ結果記憶部２１３に格納する。なお、Ｒｅｄｕｃｅ入力記憶部２１２では、ジョブＩＤとＲｅｄｕｃｅタスクのタスクＩＤとによって特定されるディレクトリに、転送元のＭａｐタスクのタスクＩＤが付された１またはそれ以上のファイルが格納される。これらのファイルに含まれるキー・バリュー形式のレコードは、キーに基づいてソートされてマージされる。
【００６７】
図８は、ジョブリストの例を示す図である。ジョブリスト１２１は、ジョブＩＤ、Ｍａｐタスク数およびＲｅｄｕｃｅタスク数の項目を含む。ジョブＩＤの項目には、ジョブトラッカー１４２が各ジョブに対して付与する識別番号が登録される。Ｍａｐタスク数の項目には、ジョブＩＤが示すジョブに関して、ジョブトラッカー１４２が定義したＭａｐタスクの数が登録される。Ｒｅｄｕｃｅタスク数の項目には、ジョブＩＤが示すジョブに関して、ジョブトラッカー１４２が定義したＲｅｄｕｃｅタスクの数が登録される。
【００６８】
図９は、タスクリストの例を示す図である。タスクリスト１２２は、ＭａｐタスクやＲｅｄｕｃｅタスクの進行状況に応じて、ジョブトラッカー１４２によって順次更新されていく。タスクリスト１２２は、ジョブＩＤ、種別、タスクＩＤ、Ｍａｐ情報、Ｒｅｄｕｃｅ番号、データノード、状態、割当ノードおよび中間データパスの項目を含む。
【００６９】
ジョブＩＤの項目には、ジョブリスト１２１と同様のジョブの識別番号が登録される。種別の項目には、タスクの種別として「Ｍａｐ」または「Ｒｅｄｕｃｅ」が登録される。タスクＩＤの項目には、ジョブトラッカー１４２が各タスクに対して付与する識別子が登録される。タスクＩＤには、例えば、ジョブＩＤと、タスクの種別を示す記号（ｍまたはｒ）と、ジョブ内でのＭａｐタスクまたはＲｅｄｕｃｅタスクを示す番号が含まれる。
【００７０】
Ｍａｐ情報の項目には、入力データのセグメントの識別情報と、Ｍａｐ定義１１１の識別情報が登録される。セグメントの識別情報には、例えば、ファイルの名前と、当該ファイル内でのセグメントの先頭位置を示すアドレスと、セグメントのサイズが含まれる。Ｍａｐ定義１１１の識別情報には、例えば、プログラムのモジュールとしてのクラスの名前が含まれる。Ｒｅｄｕｃｅ番号の項目には、ジョブ内で各Ｒｅｄｕｃｅタスクに一意に割り当てられた番号が登録される。Ｒｅｄｕｃｅ番号は、中間データのレコードがもつキーにハッシュ関数を適用したときに算出されるハッシュ値であってもよい。
【００７１】
データノードの項目には、Ｍａｐタスクについては、Ｍａｐ処理に用いる入力データを記憶しているスレーブノードまたはＤＢサーバ４２の識別子が登録される。また、データノードの項目には、Ｒｅｄｕｃｅタスクについては、Ｒｅｄｕｃｅ入力としての中間データ（１またはそれ以上のＭａｐタスクから収集された中間データ）を記憶しているスレーブノードの識別子が登録される。Ｒｅｄｕｃｅ入力としての中間データを再利用しない場合は、データノードの項目が空欄となる。入力データまたは中間データを記憶するスレーブノードが複数存在する場合もある。なお、図９において、Ｎｏｄｅ１はスレーブノード２００を示し、Ｎｏｄｅ２はスレーブノード２００ａを示し、Ｎｏｄｅ３はスレーブノード２００ｂを示し、Ｎｏｄｅ４はスレーブノード２００ｃを示している。
【００７２】
状態の項目には、タスクの状態として「未割当」、「実行中」、「完了」の何れか１つが登録される。「未割当」は、タスクを実行するスレーブノードが決定されていない状態である。「実行中」は、タスクが何れかのスレーブノードに割り当てられた後であって、当該スレーブノードにおいてタスクが未だ終了していない状態である。「完了」は、タスクが正常終了した状態である。割当ノードの項目には、タスクを割り当てたスレーブノードの識別子が登録される。未割当のタスクについては、割当ノードの項目が空欄となる。
【００７３】
中間データパスの項目には、Ｍａｐタスクについては、Ｍａｐタスクが実行されたスレーブノードにおいて、Ｍａｐ結果としての中間データが記憶されているディレクトリのパスが登録される。未割当または実行中のＭａｐタスクについては、中間データパスの項目が空欄となる。また、中間データパスの項目には、Ｒｅｄｕｃｅタスクについては、Ｒｅｄｕｃｅ入力としての中間データが記憶されるディレクトリのパスが登録される。Ｒｅｄｕｃｅ入力としての中間データを再利用する場合には、データノードの項目が示すスレーブノードにおけるパスが登録される。Ｒｅｄｕｃｅ入力としての中間データを再利用しない場合には、割当ノードの項目が示すスレーブノードにおけるパスが登録される。Ｒｅｄｕｃｅ入力としての中間データを再利用せず、かつ、未割当または実行中であるＲｅｄｕｃｅタスクについては、中間データパスの項目が空欄となる。
【００７４】
図１０は、Ｍａｐ管理テーブルとＲｅｄｕｃｅ管理テーブルの例を示す図である。Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２は、ジョブトラッカー１４２によって管理されると共に、管理ＤＢサーバ４３にバックアップされる。
【００７５】
Ｍａｐ管理テーブル１３１は、入力データ、クラス、中間データ、ジョブＩＤおよび利用履歴の項目を含む。入力データの項目には、タスクリスト１２２のＭａｐ情報と同様の、入力データのセグメントの識別情報が登録される。クラスの項目には、タスクリスト１２２のＭａｐ情報と同様の、Ｍａｐ定義１１１の識別情報が登録される。中間データの項目には、Ｍａｐ結果としての中間データを記憶する、スレーブノードの識別子とディレクトリのパスが登録される。ジョブＩＤの項目には、Ｍａｐタスクが属するジョブの識別番号が登録される。利用履歴の項目には、Ｍａｐ結果としての中間データの再利用状況を示す情報が登録される。利用履歴は、例えば、中間データが最後に参照された日時を含む。
【００７６】
Ｒｅｄｕｃｅ管理テーブル１３２は、ジョブＩＤ、Ｒｅｄｕｃｅ番号、中間データおよび利用履歴の項目を含む。ジョブＩＤの項目には、Ｒｅｄｕｃｅタスクが属するジョブの識別番号が登録される。Ｍａｐ管理テーブル１３１のレコードとＲｅｄｕｃｅ管理テーブル１３２のレコードとは、ジョブＩＤを介して関連付けられることになる。Ｒｅｄｕｃｅ番号の項目には、ジョブ内で各Ｒｅｄｕｃｅタスクに一意に割り当てられた番号が登録される。中間データの項目には、Ｒｅｄｕｃｅ入力としての中間データを記憶する、スレーブノードの識別子およびディレクトリのパスが登録される。利用履歴の項目には、Ｒｅｄｕｃｅ入力としての中間データの再利用状況を示す情報が登録される。
【００７７】
図１１は、スレーブノードへ送信するＭａｐタスク通知の例を示す図である。Ｍａｐタスク通知１２３ａは、何れかのＭａｐタスクが完了したとき、ジョブトラッカー１４２により生成されて通知バッファ１２３に格納される。通知バッファ１２３に格納されたＭａｐタスク通知１２３ａは、完了したＭａｐタスクと同じジョブに属するＲｅｄｕｃｅタスクが割り当てられたスレーブノードに対して送信される。Ｍａｐタスク通知１２３ａは、種別、ジョブＩＤ、宛先タスク、完了タスクおよび中間データの項目を含む。
【００７８】
種別の項目には、Ｍａｐタスク通知１２３ａのメッセージ種別、すなわち、Ｍａｐタスク通知１２３ａが、マスタノード１００から何れかのスレーブノードにＭａｐ完了を報告するためのメッセージであることを示す情報が登録される。ジョブＩＤの項目には、完了したＭａｐタスクが属するジョブの識別番号が登録される。宛先タスクの項目には、Ｍａｐタスク通知１２３ａの宛先となるＲｅｄｕｃｅタスクの識別子が登録される。完了タスクの項目には、完了したＭａｐタスクの識別子が登録される。中間データの項目には、Ｍａｐタスクを実行したスレーブノードの識別子と、当該スレーブノードにおいてＭａｐ結果としての中間データが記録されているディレクトリのパスが登録される。
【００７９】
次に、マスタノード１００とスレーブノード２００が実行する処理を説明する。スレーブノード２００ａ，２００ｂ，２００ｃの処理は、スレーブノード２００と同様である。
図１２は、マスタ制御の手順例を示すフローチャートである。
【００８０】
（ステップＳ１１）ジョブ分割部１４３は、ジョブ発行部１４１からの要求に応じて、入力データを複数のセグメントに分割する。ジョブトラッカー１４２は、入力データの分割結果に応じて、新たなジョブのＭａｐタスクおよびＲｅｄｕｃｅタスクを定義する。そして、ジョブトラッカー１４２は、ジョブリスト１２１にジョブを登録し、タスクリスト１２２にＭａｐタスクおよびＲｅｄｕｃｅタスクを登録する。
【００８１】
（ステップＳ１２）ジョブトラッカー１４２は、再利用情報記憶部１３０に記憶されたＭａｐ管理テーブル１３１を参照して、ステップＳ１１でタスクリスト１２２に追加したＭａｐタスクの情報を補完する。Ｍａｐ情報補完の詳細は後述する。
【００８２】
（ステップＳ１３）ジョブトラッカー１４２は、再利用情報記憶部１３０に記憶されたＲｅｄｕｃｅ管理テーブル１３２を参照して、ステップＳ１１でタスクリスト１２２に追加したＲｅｄｕｃｅタスクの情報を補完する。Ｒｅｄｕｃｅ情報補完の詳細は後述する。
【００８３】
（ステップＳ１４）ジョブトラッカー１４２は、何れかのスレーブノード（例えば、スレーブノード２００）から、ハートビートとして通知を受信する。受信され得る通知の種別には、タスクの割り当ての要求を示すタスク要求通知と、タスクが完了したことを示すタスク完了通知と、自ノード宛ての通知の有無を確認するための確認通知が含まれる。
【００８４】
（ステップＳ１５）ジョブトラッカー１４２は、ステップＳ１４で受信した通知がタスク要求通知であるか判断する。受信した通知がタスク要求通知である場合は処理をステップＳ１６に進め、タスク要求通知でない場合は処理をステップＳ１８に進める。
【００８５】
（ステップＳ１６）ジョブトラッカー１４２は、タスク要求通知を送信したスレーブノードに、未割当のタスクを１つ以上割り当てる。タスク割当の詳細は後述する。
（ステップＳ１７）ジョブトラッカー１４２は、タスク要求通知を送信したスレーブノードに対するタスク割当通知を生成し、通知バッファ１２３に格納する。タスク割当通知には、ステップＳ１６で割り当てたタスクに関するタスクリスト１２２のレコードと、当該タスクが属するジョブに関するジョブリスト１２１のレコードとが含まれる。
【００８６】
（ステップＳ１８）ジョブトラッカー１４２は、ステップＳ１４で受信した通知がタスク完了通知であるか判断する。受信した通知がタスク完了通知である場合は処理をステップＳ２０に進め、タスク完了通知でない場合は処理をステップＳ１９に進める。
【００８７】
（ステップＳ１９）ジョブトラッカー１４２は、ステップＳ１４で受信した通知の送信元のスレーブノードに対して送信すべき通知を、通知バッファ１２３から読み出す。ジョブトラッカー１４２は、通知バッファ１２３から読み出した通知を、ステップＳ１４で受信した通知に対する応答として送信する。そして、処理をステップＳ１４に進める。
【００８８】
（ステップＳ２０）ジョブトラッカー１４２は、中間データが記憶されたディレクトリのパスを示す情報をタスク完了通知から抽出し、タスクリスト１２２に登録する。
（ステップＳ２１）ジョブトラッカー１４２は、タスク完了通知により完了が報告されたタスクについて、所定のタスク完了処理を行う。タスク完了処理の詳細は後述する。
【００８９】
（ステップＳ２２）ジョブトラッカー１４２は、タスクリスト１２２を参照して、タスク完了通知により完了が報告されたタスクの属するジョブについて、全てのタスクが完了したか判断する。全てのタスクが完了した場合は処理をステップＳ２３に進め、完了していないタスクが１つ以上存在する場合は処理をステップＳ１４に進める。
【００９０】
（ステップＳ２３）ジョブトラッカー１４２は、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２を更新する。管理テーブル更新の詳細は後述する。
図１３は、Ｍａｐ情報補完の手順例を示すフローチャートである。図１３のフローチャートが示す処理は、上記のステップＳ１２において実行される。
【００９１】
（ステップＳ１２１）ジョブトラッカー１４２は、上記のステップＳ１１で定義したＭａｐタスクの中で、未選択のＭａｐタスクがあるか判断する。未選択のものがある場合は処理をステップＳ１２２に進め、全て選択済の場合は処理を終了する。
【００９２】
（ステップＳ１２２）ジョブトラッカー１４２は、上記のステップＳ１１で定義したＭａｐタスクの中から、Ｍａｐタスクを１つ選択する。
（ステップＳ１２３）ジョブトラッカー１４２は、Ｍａｐ管理テーブル１３１から、ステップＳ１２２で選択したＭａｐタスクと、入力データおよびＭａｐ処理に用いるクラスが共通するレコードを検索する。なお、選択したＭａｐタスクに関する入力データおよびクラスは、タスクリスト１２２のＭａｐ情報の項目に記載されている。
【００９３】
（ステップＳ１２４）ジョブトラッカー１４２は、ステップＳ１２３で該当するレコードが検索されたか、すなわち、ステップＳ１２２で選択したＭａｐタスクについて再利用可能なＭａｐ結果が存在するか判断する。存在する場合には処理をステップＳ１２５に進め、存在しない場合は処理をステップＳ１２１に進める。
【００９４】
（ステップＳ１２５）ジョブトラッカー１４２は、タスクリスト１２２に含まれる割当ノードおよび中間データパスの項目の情報を補完する。割当ノードおよび中間データパスは、Ｍａｐ管理テーブル１３１の中間データの項目に記載されている。
【００９５】
（ステップＳ１２６）ジョブトラッカー１４２は、後述するタスク完了処理を行い、ステップＳ１２２で選択したＭａｐタスクを、既に完了したものとして扱う。過去に生成された中間データを利用することで、当該Ｍａｐタスクは実行しなくてよい。
【００９６】
（ステップＳ１２７）ジョブトラッカー１４２は、ステップＳ１２３でＭａｐ管理テーブル１３１から検索したレコードの利用履歴を更新する。例えば、ジョブトラッカー１４２は、利用履歴を現在の日時に書き換える。そして、処理をステップＳ１２１に進める。
【００９７】
図１４は、Ｒｅｄｕｃｅ情報補完の手順例を示すフローチャートである。図１４のフローチャートが示す処理は、上記のステップＳ１３において実行される。
（ステップＳ１３１）ジョブトラッカー１４２は、上記のステップＳ１２において完了と判定したＭａｐタスクが１つ以上あるか判断する。完了と判定したＭａｐタスクがある場合は処理をステップＳ１３２に進め、ない場合は処理を終了する。
【００９８】
（ステップＳ１３２）ジョブトラッカー１４２は、上記のステップＳ１２でＭａｐ管理テーブル１３１から検索されたレコードに含まれるジョブＩＤ、すなわち、再利用するＭａｐ結果を生成したジョブのジョブＩＤを確認する。そして、ジョブトラッカー１４２は、Ｒｅｄｕｃｅ管理テーブル１３２から当該ジョブＩＤを含むレコードを検索する。
【００９９】
（ステップＳ１３３）ジョブトラッカー１４２は、上記のステップＳ１１で定義したＲｅｄｕｃｅタスクの中で、未選択のＲｅｄｕｃｅタスクがあるか判断する。未選択のものがある場合は処理をステップＳ１３４に進め、全て選択済の場合は処理を終了する。
【０１００】
（ステップＳ１３４）ジョブトラッカー１４２は、上記のステップＳ１１で定義したＲｅｄｕｃｅタスクの中から、Ｒｅｄｕｃｅタスクを１つ選択する。
（ステップＳ１３５）ジョブトラッカー１４２は、ステップＳ１３２で検索したレコードの中に、ステップＳ１３４で選択したＲｅｄｕｃｅタスクとＲｅｄｕｃｅ番号が共通するものがあるか判断する。すなわち、ジョブトラッカー１４２は、選択したＲｅｄｕｃｅタスクについて、再利用可能なＲｅｄｕｃｅ入力が存在するか判断する。存在する場合は処理をステップＳ１３６に進め、存在しない場合は処理をステップＳ１３３に進める。
【０１０１】
（ステップＳ１３６）ジョブトラッカー１４２は、タスクリスト１２２に含まれる割当ノードおよび中間データパスの項目の情報を補完する。割当ノードおよび中間データパスは、Ｒｅｄｕｃｅ管理テーブル１３２の中間データの項目に記載されている。
【０１０２】
（ステップＳ１３７）ジョブトラッカー１４２は、ステップＳ１３６でタスクリスト１２２を更新するにあたって参照された、Ｒｅｄｕｃｅ管理テーブル１３２のレコードの利用履歴を更新する。例えば、ジョブトラッカー１４２は、利用履歴を現在の日時に書き換える。そして、処理をステップＳ１３３に進める。
【０１０３】
図１５は、タスク完了処理の手順例を示すフローチャートである。図１５のフローチャートが示す処理は、上記のステップＳ２１，Ｓ１２６において実行される。
（ステップＳ２１１）ジョブトラッカー１４２は、タスクリスト１２２において、完了が報告されたタスクまたは完了したとみなしたタスクの状態を「完了」に設定する。
【０１０４】
（ステップＳ２１２）ジョブトラッカー１４２は、ステップＳ２１１で状態を「完了」に設定したタスクの種別がＭａｐであるか判断する。Ｍａｐである場合には処理をステップＳ２１３に進め、Ｒｅｄｕｃｅである場合は処理を終了する。
【０１０５】
（ステップＳ２１３）ジョブトラッカー１４２は、タスクリスト１２２を参照して、ステップＳ２１１で状態を「完了」に設定したＭａｐタスクと同じジョブに属するＲｅｄｕｃｅタスクを探し、未選択のＲｅｄｕｃｅタスクがあるか判断する。未選択のものがある場合は処理をステップＳ２１４に進め、全て選択済の場合は処理を終了する。
【０１０６】
（ステップＳ２１４）ジョブトラッカー１４２は、ステップＳ２１１で状態を「完了」に設定したＭａｐタスクと同じジョブに属するＲｅｄｕｃｅタスクを１つ選択する。
（ステップＳ２１５）ジョブトラッカー１４２は、ステップＳ２１４で選択したＲｅｄｕｃｅタスクに対して送信するＭａｐタスク通知を生成し、通知バッファ１２３に格納する。ここで生成するＭａｐタスク通知には、図１１に示したように、「完了」に設定されたＭａｐタスクの識別子や、タスクリスト１２２に登録された割当ノードおよび中間データパスが含まれる。なお、Ｍａｐタスク通知を生成した時点で、ステップＳ２１４で選択したＲｅｄｕｃｅタスクの状態が「未割当」である可能性がある。その場合、通知バッファ１２３に格納されたＭａｐタスク通知は、当該Ｒｅｄｕｃｅタスクが何れかのスレーブノードに割り当てられてから送信される。そして、処理をステップＳ２１３に進める。
【０１０７】
図１６は、タスク割当の手順例を示すフローチャートである。図１６のフローチャートが示す処理は、上記のステップＳ１６において実行される。
（ステップＳ１６１）ジョブトラッカー１４２は、タスク要求通知を送信したスレーブノードが新たなＭａｐタスクを受入可能か、すなわち、当該スレーブノードで現在実行されているＭａｐタスクの数が上限未満か判断する。受入可能な場合は処理をステップＳ１６２に進め、受入不可の場合は処理をステップＳ１６６に進める。なお、各スレーブノードのＭａｐタスクの上限数は、予めマスタノード１００に登録しておいてもよいし、各スレーブノードがマスタノード１００に通知するようにしてもよい。
【０１０８】
（ステップＳ１６２）ジョブトラッカー１４２は、未割当のＭａｐタスクの中に、タスク要求通知を送信したスレーブノードにとって「ローカルＭａｐタスク」であるものが存在するか判断する。ローカルＭａｐタスクは、入力データのセグメントが当該スレーブノードに記憶されており、入力データの転送を省略できるようなＭａｐタスクである。各ＭａｐタスクがローカルＭａｐタスクか否かは、タスク要求通知を送信したスレーブノードの識別子が、タスクリスト１２２のデータノードの項目に登録されているか否かによって判断できる。ローカルＭａｐタスクがある場合は処理をステップＳ１６３に進め、ローカルＭａｐタスクがない場合は処理をステップＳ１６４に進める。
【０１０９】
（ステップＳ１６３）ジョブトラッカー１４２は、ステップＳ１６２で見つかったローカルＭａｐタスクを１つ、タスク要求通知を送信したスレーブノードに割り当てる。ジョブトラッカー１４２は、タスクリスト１２２において、当該ローカルＭａｐタスクの割当ノードとして当該スレーブノードの識別子を登録し、また、当該ローカルＭａｐタスクの状態を「実行中」に設定する。そして、処理をステップＳ１６１に進める。
【０１１０】
（ステップＳ１６４）ジョブトラッカー１４２は、タスクリスト１２２を参照して、ローカルＭａｐタスク以外の未割当のＭａｐタスクが存在するか判断する。存在する場合は処理をステップＳ１６５に進め、存在しない場合は処理をステップＳ１６６に進める。
【０１１１】
（ステップＳ１６５）ジョブトラッカー１４２は、ステップＳ１６４で見つかったＭａｐタスクを１つ、タスク要求通知を送信したスレーブノードに割り当てる。ジョブトラッカー１４２は、ステップＳ１６３と同様、タスクリスト１２２において、当該Ｍａｐタスクの割当ノードとして当該スレーブノードの識別子を登録し、また、当該Ｍａｐタスクの状態を「実行中」に設定する。そして、処理をステップＳ１６１に進める。
【０１１２】
（ステップＳ１６６）ジョブトラッカー１４２は、タスク要求通知を送信したスレーブノードが新たなＲｅｄｕｃｅタスクを受入可能か、すなわち、当該スレーブノードで現在実行されているＲｅｄｕｃｅタスクの数が上限未満か判断する。受入可能な場合は処理をステップＳ１６７に進め、受入不可の場合は処理を終了する。なお、各スレーブノードのＲｅｄｕｃｅタスクの上限数は、予めマスタノード１００に登録しておいてもよいし、各スレーブノードがマスタノード１００に通知するようにしてもよい。
【０１１３】
（ステップＳ１６７）ジョブトラッカー１４２は、未割当のＲｅｄｕｃｅタスクの中に、タスク要求通知を送信したスレーブノードにとって「ローカルＲｅｄｕｃｅタスク」であるものが存在するか判断する。ローカルＲｅｄｕｃｅタスクは、Ｍａｐタスクから収集したＲｅｄｕｃｅ入力としての中間データが当該スレーブノードに記憶されており、中間データの転送を削減できるようなＲｅｄｕｃｅタスクである。各ＲｅｄｕｃｅタスクがローカルＲｅｄｕｃｅタスクか否かは、タスク要求通知を送信したスレーブノードの識別子が、タスクリスト１２２のデータノードの項目に登録されているか否かによって判断できる。ローカルＲｅｄｕｃｅタスクがある場合は処理をステップＳ１６８に進め、ローカルＲｅｄｕｃｅタスクがない場合は処理をステップＳ１６９に進める。
【０１１４】
（ステップＳ１６８）ジョブトラッカー１４２は、ステップＳ１６７で見つかったローカルＲｅｄｕｃｅタスクを１つ、タスク要求通知を送信したスレーブノードに割り当てる。ジョブトラッカー１４２は、タスクリスト１２２において、当該ローカルＲｅｄｕｃｅタスクの割当ノードとして当該スレーブノードの識別子を登録し、当該ローカルＲｅｄｕｃｅタスクの状態を「実行中」に設定する。そして、処理をステップＳ１６６に進める。
【０１１５】
（ステップＳ１６９）ジョブトラッカー１４２は、タスクリスト１２２を参照して、ローカルＲｅｄｕｃｅタスク以外の未割当のＲｅｄｕｃｅタスクが存在するか判断する。存在する場合は処理をステップＳ１７０に進め、存在しない場合は処理を終了する。
【０１１６】
（ステップＳ１７０）ジョブトラッカー１４２は、ステップＳ１６９で見つかったＲｅｄｕｃｅタスクを１つ、タスク要求通知を送信したスレーブノードに割り当てる。ジョブトラッカー１４２は、ステップＳ１６８と同様、タスクリスト１２２において、当該Ｒｅｄｕｃｅタスクの割当ノードとして当該スレーブノードの識別子を登録し、当該Ｒｅｄｕｃｅタスクの状態を「実行中」に設定する。そして、処理をステップＳ１６６に進める。
【０１１７】
図１７は、スレーブ制御の手順例を示すフローチャートである。
（ステップＳ３１）タスクトラッカー２２１は、マスタノード１００にタスク要求通知を送信する。タスク要求通知には、スレーブノード２００の識別子が含まれる。
【０１１８】
（ステップＳ３２）タスクトラッカー２２１は、ステップＳ３１で送信したタスク要求通知に対する応答として、マスタノード１００からタスク割当通知を受信する。タスク割当通知には、割り当てられたタスク毎に、ジョブリスト１２１の中の何れかの１つのレコードと、タスクリスト１２２の中の何れか１つのレコードとが含まれる。以下のステップＳ３３〜Ｓ３９の処理が、割り当てられたタスク毎に実行される。
【０１１９】
（ステップＳ３３）タスクトラッカー２２１は、スレーブノード２００に割り当てられたタスクの種別がＭａｐであるか判断する。種別がＭａｐである場合は処理をステップＳ３４に進め、種別がＲｅｄｕｃｅである場合は処理をステップＳ３７に進める。
【０１２０】
（ステップＳ３４）タスクトラッカー２２１は、タスク割当通知で指定された入力データのセグメントを読み込む。入力データは、スレーブノード２００に記憶されていることもあるし、他のスレーブノードやＤＢサーバ４２に記憶されていることもある。
【０１２１】
（ステップＳ３５）タスクトラッカー２２１は、Ｍａｐ実行部２２２を呼び出す（例えば、Ｍａｐ処理を行うための新たなプロセスをスレーブノード２００で起動する）。Ｍａｐ実行部２２２は、タスク割当通知で指定されたＭａｐ定義１１１に従って、ステップＳ３４で読み込まれた入力データのセグメントに対してＭａｐ処理を行う。
【０１２２】
（ステップＳ３６）Ｍａｐ実行部２２２は、Ｍａｐ結果としての中間データをＭａｐ結果記憶部２１１に格納する。このとき、Ｍａｐ実行部２２２は、中間データに含まれるキー・バリュー形式のレコードを、キーに基づいてソートし、同じＲｅｄｕｃｅタスクが担当するレコードの集合毎にファイルを生成する。各ファイルの名前として、Ｒｅｄｕｃｅ番号が付与される。生成されたファイルは、ジョブＩＤとＭａｐタスクのタスクＩＤとから特定されるディレクトリに格納される。そして、処理をステップＳ３９に進める。
【０１２３】
（ステップＳ３７）タスクトラッカー２２１は、スレーブノード２００に割り当てられたＲｅｄｕｃｅタスクが担当する中間データを取得する。タスクトラッカー２２１は、取得した中間データをＲｅｄｕｃｅ入力記憶部２１２に格納し、中間データに含まれるレコードをキーに応じてマージする。中間データ取得の詳細は後述する。
【０１２４】
（ステップＳ３８）タスクトラッカー２２１は、Ｒｅｄｕｃｅ実行部２２３を呼び出す（例えば、Ｒｅｄｕｃｅ処理を行うための新たなプロセスをスレーブノード２００で起動する）。Ｒｅｄｕｃｅ実行部２２３は、タスク割当通知で指定されたＲｅｄｕｃｅ定義１１２に従って、ステップＳ３７でレコードがマージされた後の中間データに対してＲｅｄｕｃｅ処理を行う。そして、Ｒｅｄｕｃｅ実行部２２３は、Ｒｅｄｕｃｅ結果として生成された出力データをＲｅｄｕｃｅ結果記憶部２１３に格納する。
【０１２５】
（ステップＳ３９）タスクトラッカー２２１は、タスク完了通知をマスタノード１００に送信する。タスク完了通知には、スレーブノード２００の識別子と、完了したタスクの識別子と、中間データが格納されたディレクトリのパスが含まれる。ディレクトリは、完了したタスクがＭａｐタスクの場合、生成されたＭａｐ結果が格納されたＭａｐ結果記憶部２１１のディレクトリであり、完了したタスクがＲｅｄｕｃｅタスクの場合、収集されたＲｅｄｕｃｅ入力が格納されたＲｅｄｕｃｅ入力記憶部２１２のディレクトリである。
【０１２６】
図１８は、中間データ取得の手順例を示すフローチャートである。図１８のフローチャートが示す処理は、上記のステップＳ３７において実行される。
（ステップＳ３７１）タスクトラッカー２２１は、マスタノード１００からＭａｐタスク通知を受信する。スレーブノード２００にＲｅｄｕｃｅタスクが割り当てられた時点で既に完了しているＭａｐタスクがある場合、当該Ｍａｐタスクに関するＭａｐタスク通知は、例えば、タスク割当通知と併せて受信される。スレーブノード２００にＲｅｄｕｃｅタスクが割り当てられた時点で未だ完了していないＭａｐタスクがある場合、当該Ｍａｐタスクに関するＭａｐタスク通知は、当該Ｍａｐタスクが完了してから受信される。
【０１２７】
（ステップＳ３７２）タスクトラッカー２２１は、ステップＳ３７１で受信したＭａｐタスク通知が、スレーブノード２００で実行中のジョブに関するものか判断する。すなわち、タスクトラッカー２２１は、Ｍａｐタスク通知に含まれるジョブＩＤが、以前に受信したタスク割当通知に含まれるジョブＩＤと一致するか判断する。条件を満たす場合は処理をステップＳ３７３に進め、満たさない場合は処理をステップＳ３７８に進める。
【０１２８】
（ステップＳ３７３）タスクトラッカー２２１は、Ｍａｐタスク通知で指定された中間データのうち、スレーブノード２００に割り当てられたＲｅｄｕｃｅタスクによって処理される中間データが、Ｒｅｄｕｃｅ入力記憶部２１２に既に保存されているか判断する。保存の有無は、Ｒｅｄｕｃｅ入力記憶部２１２に記憶された何れかのファイルの名前（ＭａｐタスクのタスクＩＤ）が、Ｍａｐタスク通知で指定された中間データパスの一部として記載されたＭａｐタスクのタスクＩＤと一致するか否かによって判断される。Ｒｅｄｕｃｅ入力としての中間データが保存されている場合は処理をステップＳ３７４に進め、保存されていない場合は処理をステップＳ３７６に進める。
【０１２９】
（ステップＳ３７４）タスクトラッカー２２１は、ステップＳ３７３で見つけたファイルが格納されているディレクトリ（コピー元）のパスを確認する。また、タスクトラッカー２２１は、割り当てられたＲｅｄｕｃｅタスク用のディレクトリ（コピー先）のパスを、ジョブＩＤとＲｅｄｕｃｅタスクのタスクＩＤから算出する。
【０１３０】
（ステップＳ３７５）タスクトラッカー２２１は、スレーブノード２００内で、ステップＳ３７４で確認したコピー元からコピー先に、中間データのファイルをコピーする。コピーしたファイルの名前としては、Ｍａｐタスク通知で指定された、完了したＭａｐタスクのタスクＩＤを用いる。そして、処理をステップＳ３７８に進める。
【０１３１】
（ステップＳ３７６）タスクトラッカー２２１は、Ｍａｐタスク通知で指定された他のスレーブノードのディレクトリ（コピー元）のパスを確認する。また、タスクトラッカー２２１は、割り当てられたＲｅｄｕｃｅタスク用のディレクトリ（コピー先）のパスを、ジョブＩＤとＲｅｄｕｃｅタスクのタスクＩＤから算出する。
【０１３２】
（ステップＳ３７７）タスクトラッカー２２１は、他のスレーブノードにアクセスし、ステップＳ３７６で確認したコピー元から、割り当てられたＲｅｄｕｃｅタスクの番号が付されたファイルを受信する。そして、タスクトラッカー２２１は、受信したファイルを、ステップＳ３７６で確認したコピー先に格納する。コピーしたファイルの名前としては、Ｍａｐタスク通知で指定された、完了したＭａｐタスクのタスクＩＤを用いる。
【０１３３】
（ステップＳ３７８）タスクトラッカー２２１は、未完了のＭａｐタスクがあるか判断する。未完了のＭａｐタスクの有無は、受信したＭａｐタスク通知の数が、タスク割当通知で指定されたＭａｐタスク数に一致するか否かで判断する。未完了のＭａｐタスクがある場合は処理をステップＳ３７１に進め、ない場合は処理をステップＳ３７９に進める。
【０１３４】
（ステップＳ３７９）タスクトラッカー２２１は、割り当てられたＲｅｄｕｃｅタスク用のディレクトリに記憶された中間データを、キーに応じてマージする。
図１９は、管理テーブル更新の手順例を示すフローチャートである。図１９のフローチャートが示す処理は、上記のステップＳ２３において実行される。
【０１３５】
（ステップＳ２３１）ジョブトラッカー１４２は、Ｍａｐ管理テーブル１３１から古いレコードを検索する。例えば、ジョブトラッカー１４２は、利用履歴として記載された日時から一定期間以上経過しているレコードを、古いレコードとして検索する。
【０１３６】
（ステップＳ２３２）ジョブトラッカー１４２は、ステップＳ２３１で検索されたレコードで指定されているスレーブノード宛ての削除通知を生成し、通知バッファ１２３に格納する。削除通知には、削除すべき中間データを示す情報として、検索されたレコードで指定されている中間データパスの情報が含まれる。
【０１３７】
（ステップＳ２３３）ジョブトラッカー１４２は、ステップＳ２３１で検索されたレコードを、Ｍａｐ管理テーブル１３１から削除する。
（ステップＳ２３４）ジョブトラッカー１４２は、Ｒｅｄｕｃｅ管理テーブル１３２から古いレコードを検索する。例えば、ジョブトラッカー１４２は、利用履歴として記載された日時から一定期間以上経過しているレコードを、古いレコードとして検索する。
【０１３８】
（ステップＳ２３５）ジョブトラッカー１４２は、ステップＳ２３４で検索されたレコードで指定されているスレーブノード宛ての削除通知を生成し、通知バッファ１２３に格納する。削除通知には、削除すべき中間データを示す情報として、検索されたレコードで指定されている中間データパスの情報が含まれる。
【０１３９】
（ステップＳ２３６）ジョブトラッカー１４２は、ステップＳ２３４で検索されたレコードを、Ｒｅｄｕｃｅ管理テーブル１３２から削除する。
（ステップＳ２３７）ジョブトラッカー１４２は、タスクリスト１２２を参照して、今回のジョブを実行することで、Ｍａｐタスクが割り当てられたスレーブノードに保存された中間データに関する情報を、Ｍａｐ管理テーブル１３１に追加する。
【０１４０】
（ステップＳ２３８）ジョブトラッカー１４２は、タスクリスト１２２を参照して、今回のジョブを実行することで、Ｒｅｄｕｃｅタスクが割り当てられたスレーブノードに保存された中間データに関する情報を、Ｒｅｄｕｃｅ管理テーブル１３２に追加する。
【０１４１】
図２０は、ＭａｐＲｅｄｕｃｅ処理のシーケンス例を示す図である。図２０のシーケンス例では、マスタノード１００が、スレーブノード２００にＭａｐタスクを割り当て、スレーブノード２００ａにＲｅｄｕｃｅタスクを割り当てた場合を考えている。
【０１４２】
マスタノード１００は、ＭａｐタスクとＲｅｄｕｃｅタスクを定義し、タスクリスト１２２に登録する（ステップＳ４１）。スレーブノード２００は、タスク要求通知をマスタノード１００に送信する（ステップＳ４２）。同様に、スレーブノード２００ａは、タスク要求通知をマスタノード１００に送信する（ステップＳ４３）。マスタノード１００は、スレーブノード２００にＭａｐタスクを割り当て、Ｍａｐタスクを示すタスク割当通知をスレーブノード２００に送信する（ステップＳ４４）。また、マスタノード１００は、スレーブノード２００ａにＲｅｄｕｃｅタスクを割り当て、Ｒｅｄｕｃｅタスクを示すタスク割当通知をスレーブノード２００ａに送信する（ステップＳ４５）。
【０１４３】
スレーブノード２００は、タスク割当通知に従ってＭａｐタスクを実行する（ステップＳ４６）。そして、Ｍａｐタスクが完了すると、スレーブノード２００は、タスク完了通知をマスタノード１００に送信する（ステップＳ４７）。マスタノード１００は、スレーブノード２００でＭａｐタスクが完了したことを示すＭａｐタスク通知を、Ｒｅｄｕｃｅタスクを割り当てたスレーブノード２００ａに送信する（ステップＳ４８）。スレーブノード２００ａは、Ｍａｐタスク通知を受けて、スレーブノード２００に転送要求を送信する（ステップＳ４９）。スレーブノード２００は、ステップＳ４６で生成した中間データのうち、スレーブノード２００ａのＲｅｄｕｃｅタスクによって処理される中間データを、スレーブノード２００ａに転送する（ステップＳ５０）。
【０１４４】
スレーブノード２００ａは、タスク割当通知に従って、ステップＳ５０で受信した中間データに対してＲｅｄｕｃｅタスクを実行する（ステップＳ５１）。そして、Ｒｅｄｕｃｅタスクが完了すると、スレーブノード２００ａは、タスク完了通知をマスタノード１００に送信する（ステップＳ５２）。マスタノード１００は、ジョブが完了すると、Ｍａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２を更新する（ステップＳ５３）。マスタノード１００は、更新したＭａｐ管理テーブル１３１およびＲｅｄｕｃｅ管理テーブル１３２を、管理ＤＢサーバ４３にバックアップする（ステップＳ５４）。
【０１４５】
第２の実施の形態の情報処理システムによれば、入力データの特定のセグメントに対する中間データが、過去にＭａｐタスクを実行した何れかのスレーブノードに保存されている場合には、そのセグメントに対するＭａｐ処理を省略できる。よって、データ処理の計算量を削減できる。更に、その中間データの少なくとも一部が、過去にＲｅｄｕｃｅタスクを実行した何れかのスレーブノードに保存されている場合には、そのスレーブノードにＲｅｄｕｃｅタスクを割り当てることで、中間データの転送を削減することができる。よって、通信の待ち時間を削減できると共に、ネットワーク３０の負荷を低減できる。
【０１４６】
なお、前述のように、第１の実施の形態の情報処理は、情報処理装置１０やノード２０，２０ａにプログラムを実行させることで実現でき、第２の実施の形態の情報処理は、マスタノード１００やスレーブノード２００，２００ａ，２００ｂ，２００ｃにプログラムを実行させることで実現できる。このようなプログラムは、コンピュータ読み取り可能な記録媒体（例えば、記録媒体５３）に記録しておくことができる。記録媒体としては、例えば、磁気ディスク、光ディスク、光磁気ディスク、半導体メモリなどを使用できる。磁気ディスクには、ＦＤおよびＨＤＤが含まれる。光ディスクには、ＣＤ、ＣＤ−Ｒ（Recordable）／ＲＷ（Rewritable）、ＤＶＤおよびＤＶＤ−Ｒ／ＲＷが含まれる。
【０１４７】
プログラムを流通させる場合、例えば、当該プログラムを記録した可搬記録媒体が提供される。また、プログラムを他のコンピュータの記憶装置に格納しておき、ネットワーク３０経由でプログラムを配布することもできる。コンピュータは、例えば、可搬記録媒体に記録されたプログラムまたは他のコンピュータから受信したプログラムを、記憶装置（例えば、ＨＤＤ１０３）に格納し、当該記憶装置からプログラムを読み込んで実行する。ただし、可搬記録媒体から読み込んだプログラムを直接実行してもよく、他のコンピュータからネットワーク３０を介して受信したプログラムを直接実行してもよい。また、上記の情報処理の少なくとも一部を、ＤＳＰ、ＡＳＩＣ、ＰＬＤ（Programmable Logic Device）等の電子回路で実現することも可能である。
【０１４８】
上記については単に本発明の原理を示すものである。更に、多数の変形や変更が当業者にとって可能であり、本発明は上記に示し、説明した正確な構成および応用例に限定されるものではなく、対応する全ての変形例および均等物は、添付の請求項およびその均等物による本発明の範囲とみなされる。
【符号の説明】
【０１４９】
１０  情報処理装置
１１，２２ａ  記憶部
１２  制御部
２０，２０ａ  ノード
２１，２１ａ  演算部
【図１】
2014020735.tif 000003
【図２】
2014020735.tif 000004
【図３】
2014020735.tif 000005
【図４】
2014020735.tif 000006
【図５】
2014020735.tif 000007
【図６】
2014020735.tif 000008
【図７】
2014020735.tif 000009
【図８】
2014020735.tif 000010
【図９】
2014020735.tif 000011
【図１０】
2014020735.tif 000012
【図１１】
2014020735.tif 000013
【図１２】
2014020735.tif 000014
【図１３】
2014020735.tif 000015
【図１４】
2014020735.tif 000016
【図１５】
2014020735.tif 000017
【図１６】
2014020735.tif 000018
【図１７】
2014020735.tif 000019
【図１８】
2014020735.tif 000020
【図１９】
2014020735.tif 000021
【図２０】
2014020735.tif 000022
【手続補正書】
【提出日】平成27年1月15日(2015.1.15)
【手続補正１】
【補正対象書類名】特許請求の範囲
【補正対象項目名】請求項２
【補正方法】変更
【補正の内容】
【請求項２】
選択される前記第２のノードは、過去に、前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を取得して前記第２の処理を行ったノードである、請求項１記載のデータ処理方法。
【手続補正２】
【補正対象書類名】特許請求の範囲
【補正対象項目名】請求項３
【補正方法】変更
【補正の内容】
【請求項３】
前記第２のノードは、過去に行われた前記第２のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、所定のキーを含むレコードを記憶しており、
前記第１のノードから前記第２のノードへは、前記第１のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、前記所定のキーを含むレコードを転送する、
請求項１または２記載のデータ処理方法。
【手続補正３】
【補正対象書類名】特許請求の範囲
【補正対象項目名】請求項４
【補正方法】変更
【補正の内容】
【請求項４】
前記第２のノードに、前記第１のノードから転送された前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を、前記第２の処理を行ってから少なくとも所定時間経過するまで消去せずに記憶しておく、請求項１乃至３の何れか一項に記載のデータ処理方法。
【手続補正４】
【補正対象書類名】特許請求の範囲
【補正対象項目名】請求項５
【補正方法】変更
【補正の内容】
【請求項５】
過去に指定された入力データに含まれるセグメントと、過去に行われた前記第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を、前記システムが備える記憶装置に格納して管理し、
前記第１および第２のノードは、前記記憶装置を参照して選択される、
請求項１乃至４の何れか一項に記載のデータ処理方法。
【手続補正書】
【提出日】平成28年2月12日(2016.2.12)
【手続補正１】
【補正対象書類名】特許請求の範囲
【補正対象項目名】全文
【補正方法】変更
【補正の内容】
【特許請求の範囲】
【請求項１】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムが実行するデータ処理方法であって、
第２のセグメントを含む第１の入力データが指定されたとき、前記複数のノードの中の第２のノードを用いて、前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第２の中間データをネットワーク経由で取得し、取得した前記第２の中間データに対して前記第２の処理を行うと共に、取得した前記第２の中間データを前記第２のノードに記憶し、
前記第１の入力データの後に、第１のセグメントと前記第２のセグメントとを含む第２の入力データが指定されたとき、前記複数のノードの中から、第１のノードと前記第２のノードとを選択し、
前記第１のノードを用いて、前記第１のセグメントに対して前記第１の処理を行い、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第１の中間データを前記ネットワーク経由で転送し、
前記第２のノードを用いて、前記第１のノードから転送された前記第１の中間データと、前記第２のノードに記憶されている前記第２の中間データとに対して、前記第２の処理を行う、
データ処理方法。
【請求項２】
前記第１の入力データは、第３のセグメントおよび第４のセグメントを更に含み、
前記第１の入力データが指定されたとき、前記第２のノードを用いて、前記第３のセグメントに対応する第３の中間データと前記第４のセグメントに対応する第４の中間データとを前記ネットワーク経由で更に取得し、前記第２、第３および第４の中間データに対して前記第２の処理を行うと共に、少なくとも前記第３の中間データを前記第２のノードに更に記憶し、
前記第２の入力データは、前記第３のセグメントを更に含み、
前記第２の入力データが指定されたとき、前記第２のノードを用いて、前記第１のノードから転送された前記第１の中間データと、前記第２のノードに記憶されている前記第２の中間データおよび前記第３の中間データとに対して、前記第２の処理を行う、
請求項１記載のデータ処理方法。
【請求項３】
前記第１の入力データが指定されたとき、前記第２のノードは、前記第２のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、所定のキーを含むレコードを前記第２の中間データとして取得し、
前記第２の入力データが指定されたとき、前記第１のノードから前記第２のノードへは、前記第１のセグメントに対する前記第１の処理の結果に含まれるレコードのうち、前記所定のキーを含むレコードを前記第１の中間データとして転送する、
請求項１または２記載のデータ処理方法。
【請求項４】
前記第２のノードに、前記第１のノードから転送された前記第１の中間データを、前記第１の中間データおよび前記第２の中間データに対する前記第２の処理を行ってから少なくとも所定時間経過するまで消去せずに記憶しておく、請求項１乃至３の何れか一項に記載のデータ処理方法。
【請求項５】
前記第１の入力データに含まれるセグメントと、前記第１の入力データが指定されたときに行われた前記第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を、前記システムが備える記憶装置に格納して管理し、
前記第２の入力データが指定されたとき、前記第１および第２のノードは、前記記憶装置を参照して選択される、
請求項１乃至４の何れか一項に記載のデータ処理方法。
【請求項６】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムの制御に用いられる情報処理装置であって、
第１の入力データに含まれるセグメントと、前記第１の入力データが指定されたときに行われた前記第１の処理の結果の少なくとも一部を記憶するノードとの対応関係を示す情報を記憶する記憶部と、
前記第１の入力データの後に、第１のセグメントと前記第１の入力データに含まれていた第２のセグメントとを含む第２の入力データが指定されたとき、前記記憶部を参照して、前記複数のノードの中から、第１のノードを選択し、また、前記第１の入力データが指定されたときに前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第２の中間データをネットワーク経由で取得しており、取得した前記第２の中間データに対して前記第２の処理を行うと共に、取得した前記第２の中間データを記憶している第２のノードを選択し、
前記第１のノードに、前記第１のセグメントに対して前記第１の処理を行わせ、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第１の中間データが前記ネットワーク経由で転送されるよう制御し、
前記第２のノードに、前記第１のノードから転送された前記第１の中間データと、前記第２のノードに記憶されている前記第２の中間データとに対して、前記第２の処理を行わせる制御部と、
を有する情報処理装置。
【請求項７】
複数のノードを用いて、入力データに対して第１の処理を行い、前記第１の処理の結果に対して第２の処理を行うシステムを制御するためのプログラムであって、
コンピュータに、
第２のセグメントを含む第１の入力データが指定されたとき、前記複数のノードの中の第２のノードに、前記第２のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第２の中間データをネットワーク経由で取得させ、取得した前記第２の中間データに対して前記第２の処理を行わせると共に、取得した前記第２の中間データを前記第２のノードに記憶させ、
前記第１の入力データの後に、第１のセグメントと前記第２のセグメントとを含む第２の入力データが指定されたとき、前記複数のノードの中から、第１のノードと前記第２のノードとを選択し、
前記第１のノードに、前記第１のセグメントに対して前記第１の処理を行わせ、前記第１のノードから前記第２のノードに、前記第１のセグメントに対する前記第１の処理の結果の少なくとも一部を含む第１の中間データが前記ネットワーク経由で転送されるよう制御し、
前記第２のノードに、前記第１のノードから転送された前記第１の中間データと、前記第２のノードに記憶されている前記第２の中間データとに対して、前記第２の処理を行わせる、
処理を実行させるプログラム。
【国際調査報告】
2014020735.tif 000023
2014020735.tif 000024
2014020735.tif 000025
2014020735.tif 000026