JP2014044677A - 送信制御プログラム、通信ノード、および送信制御方法 - Google Patents

Abstract

【課題】システムにかかる負荷を抑制すること。
【解決手段】送信対象となるデータＸ’１を記憶するノード１０１＃Ａは、データＸ’１と同一内容となるデータＸ’２を記憶するノード１０１＃Ｃを特定する。次に、ノード１０１＃Ａは、第１ノードとなる自ノードと送信先ノードとなるノード１０１＃Ｄの通信に対する影響度ｆ（＃Ａ，＃Ｄ）と、第２ノードとなる他ノードと送信先ノードの通信に対する影響度ｆ（＃Ｃ，＃Ｄ）を比較する。自ノードと送信先ノードの通信に対する影響度が大きいため、ノード１０１＃Ａは、データＸ’１を送信しない。また、ノード１０１＃Ｃは、自ノードと送信先ノードとなるノード１０１＃Ｄの通信に対する影響度ｆ（＃Ｃ，＃Ｄ）と、他ノードと送信先ノードの通信に対する影響度ｆ（＃Ａ，＃Ｄ）を比較する。自ノードと送信先ノードの通信に対する影響度が小さいため、ノード１０１＃Ｃは、データＸ’２を送信する。
【選択図】図１Ｂ

Description

本発明は、送信制御プログラム、通信ノード、および送信制御方法に関する。

従来、データを複製して、ネットワークに含まれる複数のノードが複製したデータを分散して記憶する技術がある。たとえば、ネットワーク内に、更新が可能なコアデータと、読込が可能な複製データを分散配置し、コアデータを利用状況やネットワーク状況に応じて動的に移動する技術がある。また、ファイルサーバのデータの冗長度を維持するため、冗長度の低下したデータと同一のデータを保存し、かつ、送信先ノードからネットワーク距離の最も近いファイルサーバを送信元ノードに設定する技術がある。（たとえば、下記特許文献１、２を参照。）

特開２００３−２５６２５６号公報 特開２００５−１４１５２８号公報

しかしながら、従来技術では、システム内で同一のデータを記憶するノード群からデータの送信元ノードを決める際に、ノード間にて通信することになり、システムにかかる負荷の増大を招いてしまう。

１つの側面では、本発明は、システムにかかる負荷を抑制することを目的とする。

本発明の一側面によれば、システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、複数のノードのうちのデータの送信先となる送信先ノードと複数のノードの各々のノードとの通信がシステムの性能に与える影響度合いを表す影響度を各々のノードに対応して記憶する記憶部を参照して、第１ノードと送信先ノードとの通信がシステムの性能に与える影響度合いを表す影響度と、特定した第２ノードと送信先ノードとの通信がシステムの性能に与える影響度合いを表す影響度と、を比較し、比較結果に基づいて、複数のノードと通信する通信部を制御して、送信先ノードにデータを送信する送信制御プログラム、通信ノード、および送信制御方法が提案される。

本発明の一態様によれば、システムにかかる負荷を抑制することができるという効果を奏する。

図１Ａは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その１）である。 図１Ｂは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その２）である。 図１Ｃは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その３）である。 図２は、分散処理システムのシステム構成例を示す説明図である。 図３は、ノードのハードウェア構成の一例を示すブロック図である。 図４は、分散処理システムのソフトウェア構成例を示す説明図である。 図５は、ＨＤＦＳの記憶内容の一例を示す説明図である。 図６は、ＨＤＦＳによるファイルの記憶方法の一例を示す説明図である。 図７は、ノードの機能構成例を示すブロック図である。 図８は、経路テーブルの記憶内容の一例を示す説明図である。 図９は、ＭａｐＲｅｄｕｃｅ処理の具体例を示す説明図である。 図１０は、Ｍａｐ処理の詳細例を示す説明図である。 図１１は、Ｍａｐ処理結果の送信先ノードの一例を示す説明図である。 図１２は、データＸ’の第１の送信方法の例を示す説明図である。 図１３は、データＸ’の第２の送信方法の例を示す説明図である。 図１４は、データＸ’の第３の送信方法の例を示す説明図である。 図１５は、データＸ’の送信判断の一例を示す説明図である。 図１６は、経路影響度関数ｆの第１の具体例を示す説明図である。 図１７は、経路影響度関数ｆの第２の具体例を示す説明図である。 図１８は、経路影響度関数ｆの第３の具体例を示す説明図である。 図１９は、経路影響度関数ｆの第４の具体例を示す説明図である。 図２０は、経路影響度関数ｆの第５の具体例を示す説明図である。 図２１は、ＭａｐＲｅｄｕｃｅ処理手順の一例を示すフローチャートである。 図２２は、送信判断処理手順の一例を示すフローチャートである。

以下に添付図面を参照して、開示の送信制御プログラム、通信ノード、および送信制御方法の実施の形態を詳細に説明する。また、本実施の形態にかかる通信ノードの例として、分散処理システムに含まれる、分散処理を実行するノードにて説明を行う。

図１Ａは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その１）である。また、図１Ｂは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その２）である。また、図１Ｃは、本実施の形態にかかる分散処理システムの動作例を示す説明図（その３）である。本実施の形態にかかる分散処理システム１００は、分散処理を実行するノード１０１＃Ａ〜１０１＃Ｄと、スイッチ１０２＃１〜１０２＃３を含む。以下、スイッチ１０２を、単に「スイッチ」と呼称する。

本実施の形態での分散処理について、分散処理システム１００がＨａｄｏｏｐを採用した例を用いて説明する。Ｈａｄｏｏｐは、膨大なデータを分散して処理する技術の一つであるＭａｐＲｅｄｕｃｅを実行するソフトウェアである。ＭａｐＲｅｄｕｃｅでは、データを複数に分割し、複数のノードの各々が、分割されたデータを処理対象とするＭａｐ処理を実行する。そして、複数のノードの少なくともいずれかのノードが、Ｍａｐ処理の処理結果を処理対象とするＲｅｄｕｃｅ処理を実行する。

Ｍａｐ処理は、別のＭａｐ処理とは独立したものであり、全てのＭａｐ処理を並列に実行できる処理である。たとえば、Ｍａｐ処理は、分散処理システム１００内の一部のデータを用いて、他の部分のデータを処理対象とする別のＭａｐ処理とは独立して、ＫｅｙＶａｌｕｅの形式にてデータを出力する処理である。ＫｅｙＶａｌｕｅの形式となるデータとは、Ｖａｌｕｅフィールドに格納された任意の保存したい値と、Ｋｅｙフィールドに格納された保存したいデータに対応する一意の標識と、の組である。

Ｒｅｄｕｃｅ処理は、Ｍａｐ処理の処理結果の属性をもとにＭａｐ処理の処理結果を集約した１以上の処理結果を処理対象とする処理である。たとえば、Ｍａｐ処理の処理結果が、ＫｅｙＶａｌｕｅの形式となるデータである場合、Ｒｅｄｕｃｅ処理は、Ｍａｐ処理の処理結果の属性となるＫｅｙフィールドをもとにＭａｐ処理の結果を集約した１つ以上の処理結果を処理対象とする処理である。また、たとえば、Ｒｅｄｕｃｅ処理は、ＶａｌｕｅフィールドをもとにＭａｐ処理結果を集約した１つ以上の処理結果を処理対象とする処理であってもよい。

以下、Ｈａｄｏｏｐにて用いられている用語を用いて、本実施の形態にかかる分散処理システム１００の動作を説明する。「ジョブ」は、Ｈａｄｏｏｐにおける処理単位である。たとえば、文字列の中に含まれる単語の出現数を単語ごとに計数する処理が１つのジョブとなる。「タスク」は、ジョブが分割された処理単位である。タスクは、Ｍａｐ処理を実行するＭａｐタスクと、Ｒｅｄｕｃｅ処理を実行するＲｅｄｕｃｅタスクとの２種類がある。Ｒｅｄｕｃｅタスクは、Ｒｅｄｕｃｅ処理を実行しやすくするため、Ｒｅｄｕｃｅ処理の前に、ＫｅｙフィールドをもとにＭａｐ処理の処理結果を集約するシャッフル＆ソート処理を実行する。

図１Ａは、分散処理システム１００にて、Ｍａｐ処理の終了状態を示している。具体的に、ノード１０１＃Ａは、Ｍａｐ処理の処理対象となるデータＸ１に対して、Ｍａｐ処理を実行してデータＸ’１を出力し、ノード１０１＃Ａの記憶領域にデータＸ’１を記憶する。また、データＸ１と同一内容となるデータＸ２を有するノード１０１＃Ｃも、データＸ’１と同一内容となるデータＸ’２を出力し、ノード１０１＃Ｃの記憶領域にデータＸ’２を記憶する。また、図１Ａにおいて、ノード１０１＃Ｄは、シャッフル＆ソート処理を実行する装置であり、データＸ’１またはデータＸ’２の送信先ノードである。

以下、送信対象となるデータを記憶するノードを「記憶ノード」と呼称する。また、「記憶ノード」のうちの、データを送信するノードを、「送信元ノード」と呼称する。また、データを受信するノードを「送信先ノード」と呼称する。

図１Ａの例では、ノード１０１＃Ａ、１０１＃Ｃが記憶ノードとなり、ノード１０１＃Ｄが送信先ノードとなる。本実施の形態にかかる分散処理システム１００は、ノード１０１＃Ａ、１０１＃Ｃのうち、分散処理システム１００にかかる負荷を抑制しつつ、ネットワークの通信量が小さくなる送信元ノードを決定する。

図１Ａにて、第１ノードとなるノード１０１＃Ａは、データＸ’１と同一内容となるデータＸ’２を記憶するノード１０１＃Ｃを第２ノードとなる他ノードとして特定する。同様に、ノード１０１＃Ｃは、データＸ’２と同一内容となるデータＸ’１を記憶するノード１０１＃Ａを他ノードとして特定する。具体的な特定方法は、図７にて後述する。

図１Ｂは、各記憶ノードが送信元ノードとなった場合の、送信元ノードと送信先ノードの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を示した図である。以下、送信元ノードと送信先ノードの通信が分散処理システム１００の性能に与える影響度合いを、単に、ノード１０１＃Ａとノード１０１＃Ｂの通信に対する影響度のように記載することもある。影響度は、各ノード１０１の記憶領域に記憶されている。

また、影響度は、値が大きいと分散処理システム１００の性能が低下する度合いが大きくなり、値が小さいと分散処理システム１００の性能が低下する度合いが小さくなるものとする。また、影響度は、値が大きいと分散処理システム１００の性能が低下する度合いが小さくなるようにしてもよい。以下、特に記載がない限り、影響度は、値が大きいと分散処理システム１００の性能が低下する度合いが大きくなるものとする。

また、影響度を算出する関数を、経路影響度関数ｆ（送信元ノードの識別情報，送信先ノードの識別情報）と定義する。第１ノードとなるノード１０１＃Ａは、自ノードと送信先ノードの通信に対する影響度ｆ（＃Ａ，＃Ｄ）と、他ノードと送信先ノードの通信に対する影響度ｆ（＃Ｃ，＃Ｄ）を比較する。「＃Ａ」、「＃Ｄ」は、それぞれ、ノード１０１＃Ａとノード１０１＃Ｄの識別情報を示す。以下、「＃ｘ」という記載については、装置＃ｘについての識別情報であるとする。ｆ（＃Ａ，＃Ｄ）がｆ（＃Ｃ，＃Ｄ）より大きいため、ノード１０１＃Ａは、送信元ノードにならず、データＸ’１を送信しない。

同様に、ノード１０１＃Ｃは、自ノードと送信先ノードの通信に対する影響度ｆ（＃Ｃ，＃Ｄ）と、他ノードと送信先ノードの通信に対する影響度ｆ（＃Ａ，＃Ｄ）を比較する。この場合、ｆ（＃Ｃ，＃Ｄ）がｆ（＃Ａ，＃Ｄ）より小さいため、ノード１０１＃Ｃは、送信元ノードになり、データＸ’２を送信する。

図１Ｃは、ノード１０１＃Ｃが、ノード１０１＃ＤにデータＸ’２を送信している状態を示す。図１Ｃで示すように、ボトルネックとなりやすいスイッチ１０２＃１を避けた通信が行われている。このように、同一内容のデータを持つ各ノード１０１が、同一基準で送信先ノードとの通信にかかる負荷が他ノードより低いか判断し、低い場合に自ノードが送信元ノードとなる。これにより、分散処理システム１００は、ノード間通信を行わなくとも分散処理システム１００に対して低負荷な経路でデータを転送できる。以下、図２〜図２２にて、分散処理システム１００の詳細について説明する。

図２は、分散処理システムのシステム構成例を示す説明図である。分散処理システム１００は、ノード１０１＃Ａ〜１０１＃Ｈと、スイッチ１０２＃１〜１０２＃５を含む。

ノード１０１は、分散処理を行う装置である。ノード１０１は、サーバでもよいし、パーソナル・コンピュータでもよい。スイッチ１０２は、通信の中継を行う装置である。たとえば、スイッチ１０２＃２は、ノード１０１＃Ａおよびノード１０１＃Ｂの通信の中継を行う。スイッチ１０２には、たとえば、リピータハブ、スイッチングハブ、ルータなどを採用することができる。また、スイッチ１０２は、リピータハブ、スイッチングハブ、ルータが混在していてもよい。たとえば、スイッチ１０２＃１がルータであり、スイッチ１０２＃５がスイッチングハブでもよい。

ノード１０１＃Ａ〜１０１＃Ｈと、スイッチ１０２＃１〜１０２＃５の接続関係は次の通りである。ノード１０１＃Ａとノード１０１＃Ｂは、スイッチ１０２＃２に接続している。ノード１０１＃Ｃとノード１０１＃Ｄは、スイッチ１０２＃３に接続している。ノード１０１＃Ｅとノード１０１＃Ｆは、スイッチ１０２＃４に接続している。ノード１０１＃Ｇとノード１０１＃Ｈは、スイッチ１０２＃５に接続している。スイッチ１０２＃２〜１０２＃５は、スイッチ１０２＃１に接続している。

このように、分散処理システム１００の接続形態はツリー型であり、スイッチ１０２＃１はスイッチ１０２＃２〜１０２＃５より上流にある。したがって、本実施の形態では、スイッチ１０２＃１を「上流スイッチ」に分類し、スイッチ１０２＃２〜１０２＃５を「下流スイッチ」に分類する。上流スイッチは、下流スイッチの通信を中継するため、通信が集中し易く、ボトルネックになりやすい。

また、分散処理システム１００の接続形態は、スター型、リング型、メッシュ型等であってもよい。また、分散処理システム１００の接続形態は、ツリー型、スター型、リング型、メッシュ型を組み合わせたものであってもよい。また、たとえば、スイッチ１０２＃１は、外部のネットワークに接続しており、外部のネットワークを介して、分散処理システム１００を管理する管理者が操作するパーソナル・コンピュータに接続していてもよい。次に、ノード１０１のハードウェア構成の説明を行う。

（ノード１０１のハードウェア構成例）
図３は、ノードのハードウェア構成の一例を示すブロック図である。図３において、ノード１０１は、Ｃｅｎｔｒａｌ Ｐｒｏｃｅｓｓｉｎｇ Ｕｎｉｔ（ＣＰＵ）３０１と、Ｒｅａｄ‐Ｏｎｌｙ Ｍｅｍｏｒｙ（ＲＯＭ）３０２と、Ｒａｎｄｏｍ Ａｃｃｅｓｓ Ｍｅｍｏｒｙ（ＲＡＭ）３０３と、を含む。また、ノード１０１は、ディスクドライブ３０４と、ディスク３０５と、通信インターフェース３０６と、を含む。また、ＣＰＵ３０１〜通信インターフェース３０６はバス３０７によってそれぞれ接続されている。また、図３では図示していないが、スイッチ１０２も、ノード１０１と同様のハードウェア構成を有する。

ＣＰＵ３０１は、ノード１０１の全体の制御を司る演算処理装置である。ＲＯＭ３０２は、ブートプログラムなどのプログラムを記憶する不揮発性メモリである。ＲＡＭ３０３は、ＣＰＵ３０１のワークエリアとして使用される揮発性メモリである。

ディスクドライブ３０４は、ＣＰＵ３０１の制御にしたがってディスク３０５に対するデータのリードおよびライトを制御する制御装置である。ディスクドライブ３０４には、たとえば、磁気ディスクドライブ、ソリッドステートドライブなどを採用することができる。ディスク３０５は、ディスクドライブ３０４の制御で書き込まれたデータを記憶する不揮発性メモリである。たとえばディスクドライブ３０４が磁気ディスクドライブである場合、ディスク３０５には、磁気ディスクを採用することができる。また、ディスクドライブ３０４がソリッドステートドライブである場合、ディスク３０５には、半導体素子メモリを採用することができる。

通信インターフェース３０６は、ネットワーク３０８と内部のインターフェースを司り、スイッチ１０２からのデータの入出力を制御する制御装置である。具体的に、通信インターフェース３０６は、通信回線を通じてネットワーク３０８となるＬｏｃａｌ Ａｒｅａ Ｎｅｔｗｏｒｋ（ＬＡＮ）、Ｗｉｄｅ Ａｒｅａ Ｎｅｔｗｏｒｋ（ＷＡＮ）、インターネットなどに接続され、ネットワーク３０８を介して他の装置に接続される。通信インターフェース３０６には、たとえば、モデムやＬＡＮアダプタなどを採用することができる。また、ノード１０１は、光ディスクドライブ、光ディスク、キーボード、マウスを有していてもよい。

図４は、分散処理システムのソフトウェア構成例を示す説明図である。分散処理システム１００は、マスタノード４０１と、スレーブノード４０２＃１〜４０２＃Ｎと、Ｈａｄｏｏｐ Ｄｉｓｔｒｉｂｕｔｅｄ Ｆｉｌｅ Ｓｙｓｔｅｍ（ＨＤＦＳ）クライアント４０３と、ジョブクライアント４０４とを含む。Ｎは、ノード１０１の合計数から１引いた数となる。

マスタノード４０１は、図１〜図３で示したノード１０１＃Ａ〜１０１＃Ｈのうちのいずれかのノード１０１である。また、スレーブノード４０２＃１〜４０２＃Ｎは、ノード１０１＃Ａ〜１０１＃Ｈのうちのマスタノード４０１に選択されたノード１０１以外のノード１０１である。また、ＨＤＦＳクライアント４０３とジョブクライアント４０４は、ノード１０１＃Ａ〜１０１＃Ｈのうちのいずれかのノード１０１でもよいし、スイッチ１０２＃１の外部に接続されているパーソナル・コンピュータでもよい。また、ＨＤＦＳクライアント４０３とジョブクライアント４０４は、同一の装置であってもよい。また、マスタノード４０１と、スレーブノード４０２＃１〜４０２＃Ｎを含めてＨａｄｏｏｐクラスタ４０５として定義する。Ｈａｄｏｏｐクラスタ４０５は、ＨＤＦＳクライアント４０３と、ジョブクライアント４０４と、を含んでもよい。

マスタノード４０１は、Ｍａｐ処理と、Ｒｅｄｕｃｅ処理をスレーブノード４０２＃１〜４０２＃Ｎに割り当てる装置である。スレーブノード４０２＃１〜４０２＃Ｎは、割り当てられたＭａｐ処理と、Ｒｅｄｕｃｅ処理を実行する装置である。

ＨＤＦＳクライアント４０３は、Ｈａｄｏｏｐ独自のファイルシステムである、ＨＤＦＳのファイル操作を行う端末である。ジョブクライアント４０４は、Ｍａｐ処理の処理対象となるデータと、実行可能ファイルとなるＭａｐＲｅｄｕｃｅプログラムと、実行ファイルの設定ファイルとを記憶しており、ジョブの実行要求をマスタノード４０１に通知する装置である。

また、マスタノード４０１は、ジョブトラッカー４１１と、ネームノード４１２と、ＨＤＦＳ４１３と、メタデータテーブル４１４とを有する。スレーブノード４０２＃ｘは、タスクトラッカー４２１＃ｘと、データノード４２２＃ｘと、ＨＤＦＳ４２３＃ｘと、Ｍａｐタスク４２４＃ｘと、Ｒｅｄｕｃｅタスク４２５＃ｘを有する。ｘは、１からＮまでの整数のいずれかである。ＨＤＦＳクライアント４０３は、ＨＤＦＳクライアントアプリケーション４３１と、ＨＤＦＳ Ａｐｐｌｉｃａｔｉｏｎ Ｐｒｏｇｒａｍｍｉｎｇ Ｉｎｔｅｒｆａｃｅ（ＡＰＩ）４３２と、を有する。ジョブクライアント４０４は、ＭａｐＲｅｄｕｃｅプログラム４４１と、ＪｏｂＣｏｎｆ４４２とを、有する。

ジョブトラッカー４１１は、実行すべきジョブをジョブクライアント４０４から受け付けた場合、ジョブをＭａｐタスク４２４、Ｒｅｄｕｃｅタスク４２５に分割する。続けて、ジョブトラッカー４１１は、Ｈａｄｏｏｐクラスタ４０５中の利用可能なタスクトラッカー４２１に、Ｍａｐタスク４２４、Ｒｅｄｕｃｅタスク４２５を割り当てる。

ネームノード４１２は、Ｈａｄｏｏｐクラスタ４０５内のファイルの記憶先を制御する。たとえば、ネームノード４１２は、Ｍａｐ処理の対象となるデータが、ＨＤＦＳ４１３、ＨＤＦＳ４２３＃１〜４２３＃Ｎのどこに記憶されるかを決定し、決定されたＨＤＦＳにファイルを送信する。

ＨＤＦＳ４１３、ＨＤＦＳ４２３＃１〜４２３＃Ｎは、ファイルを分散して記憶する記憶領域である。メタデータテーブル４１４は、ＨＤＦＳ４１３、ＨＤＦＳ４２３＃１〜４２３＃Ｎに記憶しているファイルの位置を記憶する記憶領域である。メタデータテーブル４１４を用いた具体的なファイルの記憶方法としては、図６にて後述する。

タスクトラッカー４２１は、ジョブトラッカー４１１から割り当てられたＭａｐタスク４２４やＲｅｄｕｃｅタスク４２５を、自装置に実行させる。また、タスクトラッカー４２１は、Ｍａｐタスク４２４やＲｅｄｕｃｅタスク４２５の進捗状況や処理の完了報告をジョブトラッカー４１１に通知する。

データノード４２２は、スレーブノード４０２内のＨＤＦＳ４２３を制御する。Ｍａｐタスク４２４は、Ｍａｐ処理を実行する。Ｍａｐ処理の処理結果は、Ｍａｐタスク４２４を実行したノード１０１の記憶領域に格納される。Ｒｅｄｕｃｅタスク４２５は、Ｒｅｄｕｃｅ処理を実行する。また、Ｒｅｄｕｃｅタスク４２５は、Ｒｅｄｕｃｅ処理を行う前段階として、シャッフル＆ソート処理を実行する。シャッフル＆ソート処理は、Ｍａｐ処理の結果を集約する処理を行う。具体的に、シャッフル＆ソート処理は、Ｍａｐ処理の結果をＫｅｙごとに並び替え、同一のＫｅｙとなったＶａｌｕｅを纏めて、Ｒｅｄｕｃｅ処理に出力する。

ＨＤＦＳクライアントアプリケーション４３１は、ＨＤＦＳを操作するアプリケーションである。ＨＤＦＳ ＡＰＩ４３２は、ＨＤＦＳにアクセスするＡＰＩである。ＨＤＦＳ ＡＰＩ４３２は、たとえば、ＨＤＦＳクライアントアプリケーション４３１からファイルのアクセス要求があった場合、データノード４２２に、ファイルを保持しているか否かを問い合わせる。

ＭａｐＲｅｄｕｃｅプログラム４４１は、Ｍａｐ処理を実行するプログラムと、Ｒｅｄｕｃｅ処理を実行するプログラムとである。ＪｏｂＣｏｎｆ４４２は、ＭａｐＲｅｄｕｃｅプログラム４４１の設定を記述したプログラムである。設定の例としては、Ｍａｐタスク４２４の生成数や、Ｒｅｄｕｃｅタスク４２５の生成数や、ＭａｐＲｅｄｕｃｅ処理の処理結果の出力先等である。

図５は、ＨＤＦＳの記憶内容の一例を示す説明図である。表５０１は、ＨＤＦＳの記憶内容の一例である。表５０１は、レコード５０１−１〜５０１−３を有している。表５０１は、ＫｅｙフィールドとＶａｌｕｅフィールドとを有している。たとえば、レコード５０１−１は、Ｋｅｙフィールドに、“Ｃｏｇａｎ Ｈｏｕｓｅ …”が格納されており、Ｖａｌｕｅフィールドに、“Ｔｈｅ Ｃｏｇａｎ Ｈｏｕｓｅ …”が格納されていることを示している。

図６は、ＨＤＦＳによるファイルの記憶方法の一例を示す説明図である。図６の（Ａ）では、メタデータテーブル４１４の記憶内容の一例を示している。図６の（Ｂ）では、メタデータテーブル４１４の記憶内容に従った、ＨＤＦＳ４１３、ＨＤＦＳ４２３の記憶内容の一例を示している。

図６の（Ａ）に示すメタデータテーブル４１４は、レコード６０１−１〜６０１−３を記憶する。メタデータテーブル４１４は、データＩＤｅｎｔｉｔｙ（ＩＤ）、ノードという２つのフィールドを含む。データＩＤフィールドには、データを一意に識別する情報が格納される。ノードフィールドには、データが格納されているノード１０１のＩＤが格納される。図６に示すノードフィールドは、ノード１０１のインデックスが格納されているとする。

たとえば、レコード６０１−１は、レコード５０１−１で示したデータが、ノード１０１＃Ａ、１０１＃Ｃ、１０１＃Ｇに格納されていることを示す。このように、ＨＤＦＳは、データを複製し、複製したデータをＨＤＦＳ４１３、ＨＤＦＳ４２３に格納する。複製したデータの格納先となるノードとしては、物理的に離れた位置にあるノードや、ネットワーク的に離れた位置にあるノードに置くことが好ましい。物理的に離れた位置にあるノードは、たとえば、ラックが異なるノードである。ネットワーク的に離れた位置にあるノードとは、たとえば、通信する際に通信を中継するスイッチの数が多いノードである。

（ノード１０１の機能構成）
次に、ノード１０１の機能構成について説明する。図７は、ノードの機能構成例を示すブロック図である。ノード１０１は、受付部７０１と、特定部７０２と、算出部７０３と、比較部７０４と、送信制御部７０５と、通信部７０６とを含む。制御部となる受付部７０１〜送信制御部７０５は、記憶装置に記憶されたプログラムをＣＰＵ３０１が実行することにより、受付部７０１〜送信制御部７０５の機能を実現する。記憶装置とは、具体的には、たとえば、図３に示したＲＯＭ３０２、ＲＡＭ３０３、ディスク３０５などである。または、通信インターフェース３０６を経由して他のＣＰＵが実行することにより、受付部７０１〜送信制御部７０５の機能を実現してもよい。また、通信部７０６は、通信インターフェース３０６でもよいし、通信インターフェース３０６の動作を制御するデバイスドライバを含んでもよい。デバイスドライバは、記憶装置に記憶されており、ＣＰＵ３０１が実行することにより、通信インターフェース３０６の動作を制御する。

また、ノード１０１は、複数のノードのうちのデータの送信先ノードと複数のノードの各々のノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を各々のノードに対応して記憶する経路テーブル７１１にアクセス可能である。送信先ノードは、常に固定であってもよいし、データに基づいて決定されてもよい。また、経路テーブル７１１は、複数のノードの各々のノード同士の通信の影響度を記憶していてもよい。経路テーブル７１１は、ＲＡＭ３０３、ディスク３０５といった記憶装置に格納されている。経路テーブル７１１は、各ノード１０１が有している。経路テーブル７１１の記憶内容の詳細は、図８にて後述する。

受付部７０１は、送信要求を受け付ける。たとえば、受付部７０１は、Ｍａｐ処理の処理結果をデータとしてＭａｐタスク４２４から受け付ける。より具体的な例として、自ノードが１０１＃Ａであり、ノード１０１＃ＡがＭａｐタスク４２４を実行したとする。このとき、ノード１０１＃Ａは、Ｍａｐタスク４２４によるＭａｐ処理の処理結果をノード１０１＃Ａの記憶領域に格納する。そして、ノード１０１＃Ａの受付部７０１は、ノード１０１＃Ａの記憶領域を定期的に参照することにより、ノード１０１＃Ａの記憶領域にＭａｐ処理の処理結果が書き込まれたことを検出する。なお、受け付けたデータは、ＲＡＭ３０３、ディスク３０５などの記憶領域に記憶される。

特定部７０２は、分散処理システム１００に含まれる複数のノード１０１から、自ノードが記憶するデータと同一の内容のデータを記憶する他ノードを特定する。たとえば、自ノードがノード１０１＃Ａであり、データとなるレコードがレコード５０１−１であれば、ノード１０１＃Ａが記憶するデータと同一の内容のデータを記憶するノード１０１＃Ｃ、１０１＃Ｇを特定する。具体的な特定方法として、たとえば、特定部７０２は、同一の内容のデータを記憶するノード１０１を、マスタノード４０１に問い合わせてもよい。

また、特定部７０２は、データに基づいて、複数のノード１０１から他ノードを特定してもよい。たとえば、特定部７０２は、データのハッシュを算出し、ハッシュを所定の値で割った余りに識別情報が対応するノード１０１を他ノードとして特定してもよい。また、特定部７０２は、コンシステントハッシングを実行する関数ｇ（）にデータを入力して、得た結果に対応するノード１０１を他ノードとして特定してもよい。なお、特定した他ノードの識別情報は、ＲＡＭ３０３、ディスク３０５などの記憶領域に記憶される。

算出部７０３は、自ノードと送信先ノードとの通信を中継するスイッチ１０２の数に基づいて、自ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出する。さらに、算出部７０３は、他ノードと送信先ノードとの通信を中継するスイッチ１０２の数に基づいて、他ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出する。

たとえば、自ノードがノード１０１＃Ａとなり、送信先ノードがノード１０１＃Ｃとなるとする。この場合、中継するスイッチ１０２は、スイッチ１０２＃２、１０２＃１、１０２＃３となるため、算出部７０３は、ノード１０１＃Ａとノード１０１＃Ｃの通信に対する影響度を１＋１＋１＝３というように算出する。また、算出部７０３は、スイッチ１０２＃１が上流スイッチであることを記憶しておき、上流スイッチを通常のスイッチ１０２数個分としてもよい。

また、算出部７０３は、自ノードと送信先ノードとの通信での、ノード１０１とスイッチ１０２のリンクの和を影響度として算出してもよい。ノード１０１とスイッチ１０２のリンクの数は、自ノードと送信先ノードとの通信を中継するスイッチ１０２の数より１大きい数値となる。たとえば、ノード１０１＃Ａとノード１０１＃Ｃのリンクの和は４となる。４つのリンクは、ノード１０１＃Ａとスイッチ１０２＃２のリンクと、スイッチ１０２＃２とスイッチ１０２＃１のリンクと、スイッチ１０２＃１とスイッチ１０２＃３のリンクと、スイッチ１０２＃３とノード１０１＃Ｃのリンクである。

また、算出部７０３は、上流スイッチが含まれるリンクに重みを付けて算出してもよい。たとえば、算出部７０３は、スイッチ１０２＃２とスイッチ１０２＃１のリンクと、スイッチ１０２＃１とスイッチ１０２＃３のリンクと、について、それぞれ２個分のリンクであるというようにして、影響度を算出してもよい。

また、算出部７０３は、自ノードと送信先ノードとの通信の帯域幅に基づいて、自ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出する。さらに、算出部７０３は、他ノードと送信先ノードとの通信の帯域幅に基づいて、他ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出してもよい。帯域幅とは、通信に用いる周波数の範囲である。帯域幅が広い程、通信速度が大きくなる。

たとえば、算出部７０３は、自ノードと送信先ノードとの通信の帯域幅のうちの最小値を影響度として算出してもよい。なお、帯域幅は値が大きい程性能がよいため、影響度が大きいと分散処理システム１００の性能が低下する度合いが大きくするために、たとえば、算出部７０３は、自ノードと送信先ノードとの通信の帯域幅のうちの最小値の逆数を、影響度として算出してもよい。また、算出部７０３は、所定のデータを帯域幅で除算した、データの到達時間を影響度として算出してもよい。

また、算出部７０３は、自ノードのプロセッサまたは自ノードのメモリの使用率に基づいて、自ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出する。さらに、算出部７０３は、他ノードのプロセッサまたは他ノードのメモリの使用率に基づいて、他ノードと送信先ノードとの通信が分散処理システム１００の性能に与える影響度合いを表す影響度を算出してもよい。

プロセッサは、たとえば、ＣＰＵやＤｉｇｉｔａｌ Ｓｉｇｎａｌ Ｐｒｏｃｅｓｓｏｒ（ＤＳＰ）である。プロセッサの使用率として、ノード１０１は、ＣＰＵの単位時間あたりの実行時間の比率を負荷量として算出する。また、別の算出方法として、ノード１０１は、ＣＰＵに割り当てられている処理数に基づいて算出してもよい。または、ノード１０１は、ＣＰＵに割り当てられている処理に付与されている処理量情報の合計を、ＣＰＵの負荷量として算出してもよい。なお、処理量情報は、事前に各処理を計測しておく。

メモリの使用率は、主記憶装置となるメモリの記憶容量のうち、ソフトウェアに割当済みの記憶容量の割合である。主記憶装置となるメモリは、ノード１０１のハードウェアでは、たとえば、ＲＡＭ３０３である。

たとえば、算出部７０３は、自ノードのＣＰＵ３０１の使用率を、自ノードと送信先ノードに対する影響度として算出する。また、算出部７０３は、自ノードのＲＡＭ３０３の使用率を自ノードと送信先ノードに対する影響度として算出してもよい。

また、算出部７０３は、自ノードと送信先ノードとの通信を中継するスイッチ１０２の数、自ノードと送信先ノードとの通信の帯域幅、自ノードのプロセッサまたは自ノードのメモリの使用率を組み合わせて、影響度を算出してもよい。たとえば、算出部７０３は、スイッチ１０２の数と自ノードのＣＰＵ３０１の使用率の和や積を、影響度として算出してもよい。なお、算出した影響度は、たとえば、経路テーブル７１１に記憶される。

比較部７０４は、経路テーブル７１１を参照して、自ノードと複数のノード１０１のうちのデータの送信先となる送信先ノードとの通信に対する影響度と、特定部７０２によって特定された他ノードと送信先ノードとの通信に対する影響度と、を比較する。

たとえば、データＸ’１を記憶するノード１０１＃Ａと送信先ノードとなるノード１０１＃Ｄの通信に対する影響度が３であり、データＸ’１と同一の内容となるデータＸ’２を記憶するノード１０１＃Ｃとノード１０１＃Ｄの通信に対する影響度が１とする。このとき、比較部７０４は、ノード１０１＃Ａとノード１０１＃Ｄの通信に対する影響度＝３と、ノード１０１＃Ｃとノード１０１＃Ｄの通信に対する影響度＝１と、を比較する。この場合、比較部７０４は、ノード１０１＃Ａとノード１０１＃Ｄの通信より、ノード１０１＃Ｃとノード１０１＃Ｄの通信の方が分散処理システム１００の性能が低下する度合いが小さいという比較結果を出力する。

また、経路テーブル７１１は、自ノードと送信先ノードとの通信に対する影響度と、他ノードと送信先ノードとの通信に対する影響度と、のうちのいずれか一方を記憶していない場合があってもよい。この場合、比較部７０４は、比較できないという比較結果を出力してもよい。また、経路テーブル７１１が自ノードと送信先ノードとの通信に対する影響度と、他ノードと送信先ノードとの通信に対する影響度と、のうちのいずれか一方を記憶していない場合、比較部７０４は、算出部７０３によって算出された影響度を用いて比較してもよい。

また、比較部７０４は、算出部７０３によって算出された自ノードと送信先ノードとの通信に対する影響度と、算出部７０３によって算出された他ノードと送信先ノードとの通信に対する影響度と、を比較してもよい。

また、比較部７０４は、次に示す条件を満たす場合、複数の他ノードの各々の他ノードと送信先ノードとの通信に対する影響度のうちの最小の影響度と、自ノードと送信先ノードとの通信に対する影響度と、を比較してもよい。条件とは、特定部７０２によって複数の他ノードが特定された場合である。たとえば、自ノードとなるノード１０１＃Ａが記憶するデータＸ’１と同一の内容となるデータＸ’２を記憶するノード１０１＃Ｃ、と、ノード１０１＃Ａが記憶するデータＸ’１と同一の内容となるデータＸ’３を記憶するノード１０１＃Ｇがあるとする。このとき、比較部７０４は、ノード１０１＃Ｃとノード１０１＃Ｄの通信に対する影響度とノード１０１＃Ｇとノード１０１＃Ｄの通信に対する影響度のうちの最小の影響度と、ノード１０１＃Ａとノード１０１＃Ｄの通信に対する影響度を比較する。なお、比較結果は、ＲＡＭ３０３、ディスク３０５などの記憶領域に記憶される。

送信制御部７０５は、比較部７０４による比較結果に基づいて、通信部７０６を制御して、送信先ノードにデータを送信する。また、送信制御部７０５は、自ノードと送信先ノードとの通信に対する影響度が他ノードと送信先ノードとの通信に対する影響度より小さい場合、通信部７０６を制御して、送信先ノードにデータを送信する。また、送信制御部７０５は、自ノードと送信先ノードとの通信に対する影響度が他ノードと送信先ノードとの通信に対する影響度より大きい場合、データを送信しない。

また、比較部７０４が比較できないという比較結果を出力していた場合、送信制御部７０５は、送信先ノードにデータを送信してもよい。このように、影響度の大小が判断できない場合、他ノードがデータを送信するか否か不明のため、自ノードがデータを送信しておくことにより、分散処理システム１００は、どのノードからもデータが送信先ノードに送信されないことを防ぐことができる。

また、送信制御部７０５は、比較結果と、各ノード１０１が共通して有する情報に基づいて、送信先ノードにデータを送信してもよい。たとえば、比較部７０４が、自ノードと送信先ノードとの通信に対する影響度と、他ノードと送信先ノードとの通信に対する影響度と、が同一であるという比較結果を出力したとする。このとき、送信制御部７０５は、自ノードを識別する番号が、他ノードの識別する番号より小さい場合、データを送信してもよい。ノード１０１を識別する番号とは、たとえば、Ｍｅｄｉａ Ａｃｃｅｓｓ Ｃｏｎｔｒｏｌ（ＭＡＣ）アドレスや、Ｉｎｔｅｒｎｅｔ Ｐｒｏｔｏｃｏｌ（ＩＰ）アドレスである。

通信部７０６は、複数のノード１０１と通信する。複数のノード１０１には、自ノードとの通信も含む。続けて、図８にて、影響度を記憶する経路テーブル７１１の説明を行う。

図８は、経路テーブルの記憶内容の一例を示す説明図である。経路テーブル７１１は、ノード１０１ごとに、該当のノード１０１が送信元ノードとなった場合に送信先ノードとの通信に対する影響度を記憶するテーブルである。たとえば、図８に示す経路テーブル７１１は、レコード８０１−Ａ〜８０１−Ｈを記憶する。経路テーブル７１１は、送信先ノードごとのフィールドを有する。また、経路テーブル７１１は、影響度が記憶ノードに依存して、送信先ノードに依存しない場合、１つのフィールドを有していてもよい。

たとえば、レコード８０１−Ａは、送信元ノードがノード１０１＃Ａとなる場合に、それぞれの送信先ノードに対する影響度について示している。具体的に、レコード８０１−Ａは、送信先ノードがノード１０１＃Ａである場合の影響度が０であり、送信先ノードがノード１０１＃Ｂである場合の影響度が２であり、送信先ノードがノード１０１＃Ｃである場合の影響度が６であることを示す。次に、図９を用いて、ＭａｐＲｅｄｕｃｅ処理の具体例について説明する。

図９は、ＭａｐＲｅｄｕｃｅ処理の具体例を示す説明図である。図９では、Ｍａｐ処理が、レコード５０１ごとに、Ｖａｌｕｅフィールドの単語の出現数を計数する処理であり、Ｒｅｄｕｃｅ処理が、単語の出現数を単語ごとに合計する処理であるとする。Ｍａｐ処理と、Ｒｅｄｕｃｅ処理を実行するノードを、図９では、ノード１０１＃Ａ、１０１＃Ｂ、１０１＃Ｃ、…、とする。

ノード１０１＃Ａがレコード５０１−１に対するＭａｐ処理を実行するように設定した理由として、ノード１０１＃Ａがレコード５０１−１を記憶しているため、レコード５０１−１を他のノード１０１に移行しなくともよいためである。レコード５０１−１を記憶するノード１０１＃Ｃ、１０１＃Ｇがレコード５０１−１に対するＭａｐ処理を実行してもよい。また、ノード１０１＃Ａ、１０１＃Ｃ、１０１＃Ｇの全てがレコード５０１−１に対するＭａｐ処理を実行してもよい。ノード１０１＃Ｂにレコード５０１−２に対するＭａｐ処理を実行するように設定した理由、ノード１０１＃Ｃにレコード５０１−３に対するＭａｐ処理を実行するように設定した理由も同様の理由である。

初めに、ノード１０１＃Ａ、１０１＃Ｂ、…は、Ｍａｐ処理を実行する。たとえば、ノード１０１＃Ａは、レコード５０１−１に対してＭａｐ処理を実行し、レコード５０１−１のＶａｌｕｅフィールドに出現した単語“Ｔｈｅ”、“Ｃｏｇａｎ”、…と、各単語の出現数をＫｅｙＶａｌｕｅの形式で出力する。具体的には、ノード１０１＃Ａは、Ｍａｐ処理を実行し、Ｍａｐ処理の結果として（Ｔｈｅ，２０１）、（Ｃｏｇａｎ，４２）、…、を出力する。

Ｍａｐ処理を実行後、ノード１０１＃Ａは、Ｍａｐ処理の結果を、シャッフル＆ソート処理を実行するノード１０１に送信する。具体的に、ノード１０１＃Ａは、（Ｔｈｅ，２０１）をノード１０１＃Ａに送信し、（Ｃｏｇａｎ，４２）をノード１０１＃Ｂに送信する。どのデータをどのノードに送信するかについては、たとえば、コンシステントハッシングの方法により、データに基づいて特定することができる。コンシステントハッシングとは、ノードの個数を増減させた時にもデータの保存先の変化を最小限に抑えるために使われるアルゴリズムである。

同様に、ノード１０１＃Ｂは、レコード５０１−２に対してＭａｐ処理を実行し、レコード５０１−２のＶａｌｕｅフィールドに出現した単語“Ｔｈｅ”、“Ａｎ”、…と、各単語の出現数をＫｅｙＶａｌｕｅの形式で出力する。具体的には、ノード１０１＃Ｂは、Ｍａｐ処理を実行し、（Ｔｈｅ，１０９）、（Ａｎ，１０）、…を出力する。

ノード１０１＃Ａ、１０１＃Ｂ、…によるＭａｐ処理の終了後、ノード１０１＃Ａ、１０１＃Ｂ、…、は、シャッフル＆ソート処理と、Ｒｅｄｕｃｅ処理を実行する。たとえば、ノード１０１＃Ａは、Ｍａｐ処理の結果となる（Ｔｈｅ，２０１）、（Ｔｈｅ，１０９）、…、に対してシャッフル＆ソート処理を実行して、（Ｔｈｅ，（２０１，１０９，…））を出力する。続けて、ノード１０１＃Ａは、シャッフル＆ソート処理の結果となる（Ｔｈｅ，（２０１，１０９，…））に対してＲｅｄｕｃｅ処理を実行し、（Ｔｈｅ，１０２１）を出力する。

図１０は、Ｍａｐ処理の詳細例を示す説明図である。図９にて、ノード１０１＃Ａ、１０１＃Ｃ、１０１＃Ｇの全てがレコード５０１−１に対するＭａｐ処理を実行してもよいことを説明した。図１０では、ノード１０１＃Ａ、１０１＃Ｃ、１０１＃Ｇの全てがレコード５０１−１に対するＭａｐ処理を実行した例を示す。また、図１０以降の説明において、レコード５０１−１を「データＸ」と呼称し、データＸを複製したレプリカを「データＸ１」、「データＸ２」、…と呼称する。たとえば、ノード１０１＃Ａは、データＸ１を記憶する。また、ノード１０１＃Ｃは、データＸ２を記憶し、ノード１０１＃Ｇは、データＸ３を記憶する。

データＸ１を記憶した状態にて、ノード１０１＃Ａは、Ｍａｐ処理を実行し、（Ｔｈｅ，２０１）、…、を出力する。以下、図１０以降の説明において、データＸのＭａｐ処理となる（Ｔｈｅ，２０１）を「データＸ’」と呼称し、データＸ’を複製したレプリカを「データＸ’１」、「データＸ’２」、…と呼称する。たとえば、ノード１０１＃Ａは、データＸ’１を記憶する。また、ノード１０１＃Ｃは、データＸ’２を記憶し、ノード１０１＃Ｇは、データＸ’３を記憶する。次に、図１１にて、データＸ’の送信先ノードについて説明する。

図１１は、Ｍａｐ処理結果の送信先ノードの一例を示す説明図である。図１１では、Ｍａｐ処理の結果をシャッフル＆ソート処理を実行するノード１０１に送信した結果を示している。具体的に、図１１（Ａ）がレコード５０１−１に対するＭａｐ処理の実行終了後を示しており、図１１（Ｂ）がレコード５０１−１に対するＭａｐ処理の結果の送信終了後を示している。

図１１（Ａ）では、ノード１０１＃Ａ、１０１＃Ｃ、１０１＃Ｇが、それぞれ、データＸ’１、データＸ’２、データＸ’３を記憶している。以下、図１１以降の説明において、データＸからデータＸ’を生成し、データＸ’を記憶するノードを「記憶ノードＳ」と呼称する。また、データＸ’１を記憶するノードを「記憶ノードＳ１」、データＸ’２を記憶するノードを「記憶ノードＳ２」、…と呼称する。具体的に、ノード１０１＃Ａが記憶ノードＳ１となり、ノード１０１＃Ｃが記憶ノードＳ２となり、ノード１０１＃Ｇが記憶ノードＳ３となる。記憶ノードのいずれかが、送信元ノードとなる。たとえば、図１１（Ａ）の状態から、記憶ノードＳ１〜Ｓ３のうちのいずれかが送信元ノードとなり、送信元ノードが、データＸ’に対してシャッフル＆ソート処理を実行する送信先ノードに送信する。

図１１（Ｂ）は、記憶ノードＳ１〜Ｓ３のうちのいずれかによるデータＸ’の送信の結果を示している。データＸ’の送信先ノードとなるノード１０１を、「送信先ノードＤ」と呼称し、データＸ’の送信先ノードの１番目を「送信先ノードＤ１」、データＸ’の送信先ノードの２番目を「送信先ノードＤ２」、…と呼称する。具体的に、ノード１０１＃Ｂが送信先ノードＤ１となり、ノード１０１＃Ｅが送信先ノードＤ２となり、ノード１０１＃Ｇが送信先ノードＤ３となる。データＸ’の送信先ノードの個数については、記憶ノードの個数と同数であってもよいし、異なってもよい。

続けて、記憶ノードＳ１〜Ｓ３のうちのどのノード１０１が、送信先ノードＤ１〜Ｄ３にデータＸ’を送信するかという送信方法について図１２〜図１４を用いて３つの送信方法を説明する。また、図１２〜図１４にて示す３つの送信方法は、ノード１０１間の通信量の増大を抑制するため、記憶ノードＳ１〜Ｓ３同士が通信せずとも、送信先ノードＤ１〜Ｄ３にデータＸ’を送信することができる方法である。

図１２は、データＸ’の第１の送信方法の例を示す説明図である。図１２に示す第１の送信方法は、各記憶ノードＳが、対応する送信先ノードＤに記憶するデータＸ’を送信する方法である。具体的には、記憶ノードＳ１が送信先ノードＤ１にデータＸ’１を送信し、記憶ノードＳ２が送信先ノードＤ２にデータＸ’２を送信し、記憶ノードＳ３が送信先ノードＤ３にデータＸ’３を送信する。図１２に示す第１の送信方法は、データＸ’１が１つ目の複製データであり、データＸ’２が２つ目の複製データであり、データＸ’３が３つ目の複製データであると区別できる場合に有効である。

図１３は、データＸ’の第２の送信方法の例を示す説明図である。図１３に示す第２の送信方法は、記憶ノードＳのいずれかが、送信先ノードＤ全てにデータＸ’を送信する方法である。具体的には、たとえば記憶ノードＳ１が、送信先ノードＤ１〜Ｄ３にデータＸ’１を送信する。なお、送信先ノードＤ２は、データＸ’１を受信し、データＸ’２として保存する。同様に、送信先ノードＤ３は、データＸ’１を受信し、データＸ’３として保存する。図１３に示す第２の送信方法は、データＸ’１〜Ｘ’３が区別でき、送信を行うノードが容易に決定できる場合に有効である。

図１４は、データＸ’の第３の送信方法の例を示す説明図である。図１４に示す第３の送信方法は、記憶ノードＳの全てが送信先ノードＤの全てにデータＸ’を送信し、送信先ノードＤは、重複したデータＸ’を除去して、データＸ’を保存する方法である。

具体的には、記憶ノードＳ１は、送信先ノードＤ１〜Ｄ３にデータＸ’１を送信する。同様に、記憶ノードＳ２は、送信先ノードＤ１〜Ｄ３にデータＸ’２を送信し、同様に、記憶ノードＳ３は、送信先ノードＤ１〜Ｄ３にデータＸ’３を送信する。続けて、送信先ノードＤ１〜Ｄ３は、同一の内容となるデータＸ’１〜Ｘ’３のうちのいずれか２つを除去し、残りの１つを保存する。図１４に示す第３の送信方法は、データＸ’１〜Ｘ’３が区別できない場合に有効である。

以上、図１２〜図１４にて示した送信方法は、記憶ノードＳ１〜Ｓ３同士が通信しないために、送信時に効率が悪くなる可能性がある。たとえば、第１の送信方法を選択していた場合、記憶ノードＳ１のネットワーク的に近い位置に送信先ノードＤ１〜Ｄ３があり、記憶ノードＳ２と送信先ノードＤ２とのネットワーク的に離れた位置にあるとする。この場合、記憶ノードＳ２が送信先データＤ２にデータＸ’２を送信すると、通信を中継するスイッチの数が多くなり、ネットワークが込み合う原因となる。また、ネットワーク的に遠い位置にある場合、スイッチ１０２＃１のような上流ノードを中継することになり、上流ノードがボトルネックとなる可能性がある。次に、図１５を用いて、記憶ノードＳ１〜Ｓ３同士が通信せずに、ネットワーク的に近い記憶ノードＳが送信先ノードＤにデータＸ’を送信する方法を行う例について説明する。

図１５は、データＸ’の送信判断の一例を示す説明図である。図１５では、記憶ノードＳ１〜Ｓ３が、送信先ノードＤ１〜Ｄ３に送信するか否かを判断する方法である。記憶ノードＳ１〜Ｓ３は、送信先ノードＤごとに、自ノードがデータＸ’を送信すべきか否かを、ノードｘからノードｙへの経路のコストを表す経路影響度関数ｆ（ｘ，ｙ）を用いて判断する。経路影響度関数ｆの具体例については、図１６〜図２０にて説明する。図１５の例で示す経路影響度関数ｆは、図１６で示す具体例であり、ノード１０１とスイッチ１０２のリンクのコストの総和を返す関数とする。

たとえば、記憶ノードＳ１となるノード１０１＃Ａは、送信先ノードＤ１に対して、ｆ（記憶ノードＳ１＝＃Ａ，送信先ノードＤ１＝＃Ｂ）、ｆ（記憶ノードＳ２＝＃Ｃ，＃Ｂ）、ｆ（記憶ノードＳ３＝＃Ｇ，＃Ｂ）の各影響度を算出する。算出した結果は以下のようになる。

ｆ（＃Ａ，＃Ｂ）＝２
ｆ（＃Ｃ，＃Ｂ）＝６
ｆ（＃Ｇ，＃Ｂ）＝６

次に、ノード１０１＃Ａは、算出された影響度群のうち、最小となった影響度の記憶ノードが自ノードか否かを判断する。この場合、最小となった影響度＝ｆ（＃Ａ，＃Ｂ）＝２であるため、ノード１０１＃Ａは、送信先ノードＤにデータＸ’を送信する送信元ノードが自ノードであると判断する。したがって、ノード１０１＃Ａは、ノード１０１＃ＢにデータＸ’１を送信する。続けて、ノード１０１＃Ａは、送信先ノードＤ２となるノード１０１＃Ｅに対して影響度を算出する。算出した結果は以下のようになる。

ｆ（＃Ａ，＃Ｅ）＝６
ｆ（＃Ｃ，＃Ｅ）＝６
ｆ（＃Ｇ，＃Ｅ）＝６

次に、ノード１０１＃Ａは、算出された影響度群のうち、最小となった影響度の記憶ノードが自ノードか否かを判断する。この場合、最小となった影響度＝ｆ（＃Ａ，＃Ｅ）＝６であるため、ノード１０１＃Ａは、送信先ノードＤにデータＸ’を送信する送信元ノードが自ノードであると判断する。したがって、ノード１０１＃Ａは、ノード１０１＃ＥにデータＸ’１を送信する。続けて、ノード１０１＃Ａは、送信元ノードＤ３となるノード１０１＃Ｇに対して影響度を算出する。算出した結果は以下のようになる。

ｆ（＃Ａ，＃Ｇ）＝６
ｆ（＃Ｃ，＃Ｇ）＝６
ｆ（＃Ｇ，＃Ｇ）＝０

次に、ノード１０１＃Ａは、算出された影響度群のうち、最小となった影響度の記憶ノードが自ノードか否かを判断する。この場合、最小となった影響度＝ｆ（＃Ｇ，＃Ｇ）＝０であるため、ノード１０１＃Ａは、送信先ノードＤにデータＸ’を送信する送信元ノードが自ノードでないと判断する。したがって、ノード１０１＃Ａは、ノード１０１＃ＧにデータＸ’１を送信しない。

同様に、記憶ノードＳ２となるノード１０１＃Ｃ、記憶ノードＳ３となるノード１０１＃Ｇでも、送信先ノードＤごとに、自ノードがデータＸ’を送信すべきか否かを判断して、送信すべきとなった場合、送信先ノードＤに送信する。具体的に、ノード１０１＃Ｃは、ノード１０１＃ＥにデータＸ’２を送信する。また、ノード１０１＃Ｇは、ノード１０１＃Ｅとノード１０１＃ＧにデータＸ’３を送信する。なお、ノード１０１＃Ｇは、自ノードにデータＸ’３を送信している。自ノードにデータＸ’を送信するとなった場合、ノード１０１は、送信先アドレスとして、自ノードのアドレスを設定してもよいし、ループバックアドレスを設定してもよい。または、自ノードに送信するとなった場合、ノード１０１は、実際に送信せずに、データＸ’を格納している記憶領域から、受信時にデータＸ’を格納する記憶領域に複製してもよい。以上の処理により、記憶ノードＳ１〜Ｓ３同士が通信せずに、送信先ノードＤに対してネットワーク的に近い記憶ノードＳがデータＸを送信することができる。

また、ノード１０１＃Ｅは、データＸ’１〜Ｘ’３を受信している。この場合、ノード１０１＃Ｅは、図１４にて示した第３の送信方法を用いて、データＸ’１〜Ｘ’３のうちのいずれか２つを除去し、残りの１つを保存してもよい。また、可能な限りデータＸ’が２箇所以上の記憶ノードＳから送信されないようにしてもよい。データＸ’が２箇所以上の記憶ノードＳから送信されないようにするには、最小となる影響度が複数ある場合に、他の基準による影響度を算出して、他の基準による影響度が小さい方を最小の影響度としてもよい。具体例として、図１５で用いた影響度は、図１６で後述する第１の例を用いている。最小となる影響度が複数ある場合、記憶ノードＳ１〜Ｓ３は、図１７で後述する第２の例を用いて、最小となる影響度を算出してもよい。次に、図１６〜図２０を用いて経路影響度関数ｆの具体例を説明する。

図１６は、経路影響度関数ｆの第１の具体例を示す説明図である。図１６で示す経路影響度関数ｆ（ｘ，ｙ）は、ノードｘからノードｙまでの経路上のコストの総和を返す関数である。具体的に、ノード１０１と下流スイッチのリンクのコストを１と定義し、下流スイッチと上流スイッチのリンクのコストを２と定義する。たとえば、図１６で示す経路影響度関数ｆは、ｆ（＃Ａ，＃Ｂ）＝１＋１＝２、ｆ（＃Ｃ，＃Ｂ）＝１＋２＋２＋１＝６、ｆ（＃Ａ，＃Ｃ）＝１＋２＋２＋１＝６、…となる。得られた影響度は、経路テーブル７１１の対応するレコードに格納される。具体的に、ｆ（＃Ａ，＃Ｂ）＝２は、レコード８０１−Ａのノード１０１＃Ｂフィールドに格納され、ｆ（＃Ｃ，＃Ｂ）＝６は、レコード８０１−Ｃのノード１０１＃Ｂフィールドに格納される。また、ｆ（＃Ａ，＃Ｃ）＝６は、レコード８０１−Ａのノード１０１＃Ｃフィールドに格納される。

記憶ノードから送信先ノードまでの経路の特定方法については、分散処理システム１００の管理者が特定してもよいし、記憶ノードが、記憶ノードから送信先ノードまでの経路を特定するコマンドを実行してもよい。

経路を特定するコマンドとしては、たとえば、スイッチ１０２がルータであれば、トレースルートコマンドである。たとえば、ノード１０１＃Ａ〜１０１＃Ｈは、予め、上流スイッチのＩＰアドレスを記憶しておく。次に、ノード１０１＃Ａがノード１０１＃Ｂまでのトレースルートコマンドを実行すると、ノード１０１＃Ａは、ノード１０１＃Ａからノード１０１＃Ｂまでのスイッチ１０２のＩＰアドレスの一覧を得ることができる。続けて、ノード１０１＃Ａは、ＩＰアドレスの一覧を用いて、ノード１０１＃Ｂまでの通信のコストの総和を算出する。具体的に、ノード１０１＃Ａは、ノード１０１から下流スイッチまでのコストと、下流スイッチ同士間のコストと、下流スイッチからノード１０１までのコスト、を１とし、上流スイッチが含まれた場合のコストを２として、コストの総和を算出する。算出した結果は、経路テーブル７１１の対応するレコードに格納される。

続けて、ノード１０１＃Ａは、ノード１０１＃Ａからノード１０１＃Ａまでの通信のコスト、ノード１０１＃Ａからノード１０１＃Ｃまでの通信のコスト、…、ノード１０１＃Ａからノード１０１＃Ｈまでの通信のコストを算出する。算出後、ノード１０１＃Ａは、ノード１０１＃Ａから各ノード１０１までの通信のコスト、をノード１０１＃Ｂ〜１０１＃Ｈに配布する。配布を受けたノード１０１＃Ｂ〜１０１＃Ｈは、経路テーブル７１１の自ノードに対応するレコードに配布内容を格納する。同様に、ノード１０１＃Ｂ〜１０１＃Ｈも、影響度を算出し、他ノードに配布する。これにより、ノード１０１＃Ａ〜１０１＃Ｈは、記憶ノードと送信先ノードがどのノード１０１であっても影響度を取得できる。

図１７は、経路影響度関数ｆの第２の具体例を示す説明図である。図１７で示す経路影響度関数ｆ（ｘ，ｙ）は、ノードｘからノードｙまでの経路について、スイッチを通過した数を返す関数である。このとき、高負荷になりがちなスイッチや、低性能のスイッチを、複数のスイッチとして数えてもよい。たとえば、スイッチ１０２＃１をスイッチ４つ分として数えるものとする。この場合、ｆ（＃Ａ，＃Ｂ）＝１、ｆ（＃Ｃ，＃Ｂ）＝１＋４＋１＝６、ｆ（＃Ａ，＃Ｃ）＝１＋４＋１＝６、…となる。

図１８は、経路影響度関数ｆの第３の具体例を示す説明図である。図１８で示す経路影響度関数ｆ（ｘ，ｙ）は、ノードｘからノードｙまでの経路について、データを送信した時にかかる時間を返す関数である。たとえば、ノードｘが、ノードｙに１６バイトのデータを送信し、送信にかかった時間を経路テーブル７１１のノードｘに対応するレコードに格納する。ノードｘは、実際にデータを送信してもよいし、ノードｘからノードｙまでの理論的な送信速度から、理論的な送信時間を算出してもよい。全てのノード１０１が、自ノードからの他ノードへのデータの送信時間を他ノードに配布し、全てのノード１０１が同一の情報を経路テーブル７１１に格納しておくことになる。配布する時期として、全てのノード１０１は、分散処理システム１００の運用開始時に一度配布してもよいし、定期的に配布してもよい。

たとえば、ノード１０１＃Ａからノード１０１＃Ｂへの１６バイトのデータの送信時間が１００［μ秒］であり、ノード１０１＃Ａからノード１０１＃Ｃへの１６バイトのデータの送信時間が１０２［μ秒］であるとする。このとき、図１８で示す経路影響度関数ｆは、ｆ（＃Ａ，＃Ｂ）＝１００［μ秒］となり、ｆ（＃Ａ，＃Ｃ）＝１０２［μ秒］となる。得られた影響度は、経路テーブル７１１の対応するレコードに格納される。具体的に、ｆ（＃Ａ，＃Ｂ）＝１００は、レコード８０１−Ａのノード１０１＃Ｂフィールドに格納され、ｆ（＃Ａ，＃Ｃ）＝１０２は、レコード８０１−Ａのノード１０１＃Ｃフィールドに格納される。また、ノード１０１＃Ａは、ノード１０１＃Ｃからｆ（＃Ｃ，＃Ｂ）＝１０２［μ秒］を受け付ける。ｆ（＃Ｃ，＃Ｂ）は、レコード８０１−Ｃのノード１０１＃Ｂフィールドに格納される。

図１９は、経路影響度関数ｆの第４の具体例を示す説明図である。図１９で示す経路影響度関数ｆ（ｘ，ｙ）は、ノードｘからノードｙまでの経路の帯域幅を返す関数である。具体的に、ｆ（ｘ，ｙ）は、経路上の帯域幅のうち、最小の帯域幅を返す関数である。

たとえば、ノード１０１＃Ａからノード１０１＃Ｂまでの経路のうち、ノード１０１と下流スイッチの帯域幅が１００［Ｍｂｐｓ］であり、下流スイッチと上流スイッチの帯域幅が１０［Ｍｂｐｓ］であるとする。この場合、図１９で示す経路影響度関数ｆは、ｆ（＃Ａ，＃Ｂ）＝Ｍｉｎ（１００，１００）＝１００［Ｍｂｐｓ］、ｆ（＃Ｃ，＃Ｂ）＝Ｍｉｎ（１００，１０，１０，１００）＝１０［Ｍｂｐｓ］、ｆ（＃Ａ，＃Ｃ）＝Ｍｉｎ（１００，１０，１０，１００）＝１０［Ｍｂｐｓ］、…となる。ただし、Ｍｉｎ（）は、引数内の最小値を返す関数である。得られた影響度は、経路テーブル７１１の対応するレコードに格納される。

また、分散処理システム１００は、帯域幅の定義を、分散処理システム１００を一定の条件にした状態において実測した値としてもよい。一定の条件とは、たとえば、全てのノード１０１に高負荷をかけた状態である。一定の条件において、ノード１０１＃Ａがノード１０１＃Ｂに送信できた単位時間当たりのデータ量が１１２［Ｍｂｉｔ］であれば、ノード１０１＃Ａは、ｆ（＃Ａ，＃Ｂ）＝１１２［Ｍｂｐｓ］と設定する。同様に、ノード１０１＃Ａは、ノード１０１＃Ｃ〜１０１＃Ｈにもデータを送信し、帯域幅を設定する。設定後、ノード１０１＃Ａは、設定した帯域幅をノード１０１＃Ｂ〜１０１＃Ｈに配布する。同様に、ノード１０１＃Ｂ〜１０１＃Ｈも、他ノードとの帯域幅を定義して、他ノードに配布する。

図２０は、経路影響度関数ｆの第５の具体例を示す説明図である。図２０で示す経路影響度関数ｆ（ｘ，ｙ）は、ノードｘのＣＰＵ使用率を返す関数である。たとえば、ある時点でのノード１０１＃ＡのＣＰＵ使用率が８０［％］であり、ノード１０１＃ＢのＣＰＵ使用率が５０［％］であり、ノード１０１＃ＣのＣＰＵ使用率が３０［％］であるとする。この場合、図２０で示す経路影響度関数ｆは、ｆ（＃Ａ，＃Ｂ）＝８０［％］、ｆ（＃Ｃ，＃Ｂ）＝３０［％］、ｆ（＃Ａ，＃Ｃ）＝８０［％］、…となる。各ノード１０１のＣＰＵ使用率は、全てのノード１０１に配布されている。配布する時期として、全てのノード１０１は、分散処理システム１００の運用開始時、事前に実験して測定した値を配布してもよいし、定期的に測定した値を配布してもよい。

図２０で示す経路影響度関数ｆ（ｘ，ｙ）の結果は、ノードｘに依存し、ノードｙに依存しない。したがって、経路テーブル７１１は、記憶ノードと送信先ノードの通信に対する影響度を記憶しなくてよく、記憶ノードに対する影響度があればよい。よって、経路テーブル７１１は、たとえば、図２０で示す記憶形態であってもよい。図２０で示す経路テーブル７１１は、ノード１０１ごとに、該当のノードのＣＰＵ使用率を記憶するテーブルである。具体的に、ノード１０１＃ＡのＣＰＵ使用率８０［％］がレコード８０１−Ａに格納され、ノード１０１＃ＢのＣＰＵ使用率５０［％］がレコード８０１−Ｂに格納され、ノード１０１＃ＣのＣＰＵ使用率３０［％］がレコード８０１−Ｃに格納される。次に、図２１および図２２にて、分散処理システム１００が実行するフローチャートを説明する。

図２１は、ＭａｐＲｅｄｕｃｅ処理手順の一例を示すフローチャートである。ＭａｐＲｅｄｕｃｅ処理は、複数のノード１０１で分散処理を行う処理である。マスタノード４０１は、データＸを有する記憶ノードに、Ｍａｐ処理の実行要求を通知する（ステップＳ２１０１）。データＸを有する記憶ノードがノード１０１＃Ａ〜１０１＃Ｈのうちいずれかというのは、マスタノード４０１がメタデータテーブル４１４を参照することで特定することができる。データＸを有するノード１０１が記憶ノードとなる。また、マスタノード４０１は、データＸを有する記憶ノード全てに実行要求を通知する。

実行要求を受け付けた記憶ノードは、データＸのＭａｐ処理を実行する（ステップＳ２１０２）。次に、記憶ノードは、送信判断処理を実行する（ステップＳ２１０３）。送信判断処理の詳細は、図２２にて後述する。次に、記憶ノードは、Ｍａｐ処理の処理結果となるデータＸ’を自ノードが送信することになった送信先ノードがあるか否かを判断する（ステップＳ２１０４）。データＸ’を自ノードが送信することになった送信先ノードがあるか否かについては、送信判断処理の出力結果を参照することにより判断することができる。

自ノードが送信することになった送信先ノードがあると判断した場合（ステップＳ２１０４：Ｙｅｓ）、記憶ノードは、データＸ’を送信先ノードに送信する（ステップＳ２１０５）。送信先ノードは、複数ある場合も存在する。送信後、記憶ノードは、ＭａｐＲｅｄｕｃｅ処理を終了する。自ノードが送信することになった送信先ノードがないと判断した場合（ステップＳ２１０４：Ｎｏ）、記憶ノードは、ＭａｐＲｅｄｕｃｅ処理を終了する。

データＸ’を受け付けた送信先ノードは、シャッフル＆ソート処理を実行する（ステップＳ２１０６）。続いて、送信先ノードは、Ｒｅｄｕｃｅ処理を実行する（ステップＳ２１０７）。ステップＳ２１０７の処理終了後、送信先ノードは、ＭａｐＲｅｄｕｃｅ処理を終了する。ＭａｐＲｅｄｕｃｅ処理を実行することにより、分散処理システム１００は、ジョブをノード１０１に分散して処理することができる。

図２２は、送信判断処理手順の一例を示すフローチャートである。送信判断処理は、記憶ノードＳｘが、送信先ノードにデータＸ’を送信するか否かを判断する処理である。また、送信判断処理は、図２１のステップＳ２１０１の処理にて、Ｍａｐ処理の実行要求を受け付けた記憶ノード全てが行う。

記憶ノードＳｘは、データＸのＭａｐ処理の処理結果となるデータＸ’を取得する（ステップＳ２２０１）。次に、記憶ノードＳｘは、データＸに対してコンシステントハッシングを実行するｇ（Ｘ）を実行し、データＸ’を記憶する記憶ノードＳ１，Ｓ２，…Ｓｎを特定する（ステップＳ２２０２）。続けて、記憶ノードＳｘは、データＸ’に対してコンシステントハッシングを実行するｇ（Ｘ’）を実行し、データＸ’の送信先となる送信先ノードＤ１，Ｄ２，…，Ｄｍを特定する（ステップＳ２２０３）。なお、ｎ、ｍは自然数である。

次に、記憶ノードＳｘは、未選択の送信先ノードＤｊを選択する（ステップＳ２２０４）。ｊは、１からｍのうちのいずれかの整数である。続けて、記憶ノードＳｘは、経路影響度関数ｆ（Ｓ１，Ｄｊ），ｆ（Ｓ２，Ｄｊ），…，ｆ（Ｓｎ，Ｄｊ）を実行する（ステップＳ２２０５）。次に、記憶ノードＳｘは、経路影響度関数の結果が最小となったｆ（Ｓｉ，Ｄｊ）について、記憶ノードＳｉが記憶ノードＳｘか否かを判断する（ステップＳ２２０６）。

ステップＳ２２０５、ステップＳ２２０６について、記憶ノードＳｘは、初めにｆ（Ｓｘ，Ｄｊ）を実行し、次に、ｆ（Ｓ１，Ｄｊ）を実行し、ｆ（Ｓｘ，Ｄｊ）がｆ（Ｓ１，Ｄｊ）より大きいか比較してもよい。ｆ（Ｓｘ，Ｄｊ）がｆ（Ｓ１，Ｄｊ）より大きい場合、記憶ノードＳｘがデータを送信先ノードＤｊに送信する可能性がなくなるため、ステップＳ２２０８：Ｎｏのルートを通り、次の送信先ノードを選択してもよい。これにより、ｆ（Ｓ１，Ｄｊ）〜ｆ（Ｓｎ，Ｄｊ）全てを実行しなくともよい場合が発生し、記憶ノードＳｘは処理時間を短縮できる。

記憶ノードＳｉが記憶ノードＳｘである場合（ステップＳ２２０６：Ｙｅｓ）、記憶ノードＳｘは、データＸ’を自ノードが送信先ノードＤｊに送信することを記憶する（ステップＳ２２０７）。ステップＳ２２０７の実行終了後、または、記憶ノードＳｉが記憶ノードＳｘでない場合（ステップＳ２２０６：Ｎｏ）、記憶ノードＳｘは、全ての送信先ノードを選択したか否かを判断する（ステップＳ２２０８）。

まだ未選択の送信先ノードがある場合（ステップＳ２２０８：Ｎｏ）、記憶ノードＳｘは、ステップＳ２２０４の処理に移行する。全ての送信先ノードを選択した場合（ステップＳ２２０８：Ｙｅｓ）、記憶ノードＳｘは、データＸ’を自ノードが送信することとなった送信先ノードＤの識別情報を出力する（ステップＳ２２０９）。ステップＳ２２０９の処理終了後、記憶ノードＳｘは、送信判断処理を終了する。送信判断処理を実行することにより、ノード１０１は、ノード１０１間で通信しなくとも、送信元ノードが自ノードであるか否かを判断できる。

以上説明したように、本実施の形態にかかるノード１０１によれば、同じデータを持つ各ノード１０１が同一基準で送信先ノードとの通信にかかる負荷が他ノードより低いか判断し、低い場合に自ノードが送信元ノードとなる。これにより、分散処理システム１００は、ノード１０１間通信で送信元ノードを決めなくても分散処理システム１００にかかる負荷が低い経路でデータ送信できる。

また、ノード１０１によれば、マスタノード４０１が集中して送信元ノードを決めずによくなり、ノード１０１にかかる負荷を分散することができる。また、元データのレプリカを保持するサーバのうち経路コストが最も低いものからデータの再配置先となるノードに通信することにより、通信が高コストな経路を通過することを抑制することができる。また、ノード１０１によれば、特定の経路に通信が集中してボトルネックになるのを防ぐことができる。また、ノード１０１によれば、スループットを向上させ、データ転送にかかる時間を削減でき、高速化、低コスト化、低負荷化を実現することができる。

また、ノード１０１によれば、自ノードと送信先ノードの通信に対する影響度が他ノードと送信先ノードの通信に対する影響度より小さい場合、データを送信してもよい。これにより、分散処理システム１００は、１度の比較で自ノードがデータを送信すべきか否かを判断できるため、自ノードがデータを送信すべきかの判断を高速に行える。

また、ノード１０１によれば、複数の他ノードの各々の他ノードと送信先ノードの通信に対する影響度のうちの最小値と、自ノードと送信先ノードとの通信に対する影響度を比較してもよい。これにより、分散処理システム１００は、ノード１０１間通信で送信元を決めなくても分散処理システム１００にかかる負荷が最も低いノードがデータを送信できる。

また、ノード１０１によれば、データに基づいて、複数のノード１０１から他ノードを特定してもよい。これにより、ノード１０１は、マスタノード４０１等に問い合わせなくても他ノードを特定できるため、他ノードを特定することにかかる通信を削減することができる。

また、ノード１０１によれば、自ノードと送信先ノードとの通信を中継するスイッチの数に基づいて、自ノードと送信先ノードの通信に対する影響度を算出してもよい。これにより、分散処理システム１００は、中継するスイッチの数が少ない、分散処理システム１００にかかる負荷が低い経路でデータ送信できる。

また、ノード１０１によれば、自ノードと送信先ノードとの通信の帯域幅に基づいて、自ノードと送信先ノードの通信に対する影響度を算出してもよい。これにより、分散処理システム１００は、帯域幅が広く、輻輳が発生しにくい通信経路にてデータ送信ができる。

また、ノード１０１によれば、自ノードのプロセッサまたはメモリの使用率に基づいて、自ノードと送信先ノードの通信に対する影響度を算出してもよい。これにより、分散処理システム１００は、処理能力に余裕があるノードにてデータ送信が行えるため、ノードの高負荷によるデータ送信処理の遅延を防ぐことができる。

また、本実施の形態にかかる分散処理システム１００はｈａｄｏｏｐを採用しているが、ｈａｄｏｏｐに限らず、冗長性のあるデータが複数のノードにあり、複数のノードから送信先ノードに送信する時に本実施の形態にかかる送信制御方法を適用することができる。

なお、本実施の形態で説明した送信制御方法は、予め用意されたプログラムをパーソナル・コンピュータやワークステーション等のコンピュータで実行することにより実現することができる。本送信制御プログラムは、ハードディスク、フレキシブルディスク、ＣＤ−ＲＯＭ、Ｍａｇｎｅｔｏ Ｏｐｔｉｃａｌ（ＭＯ）、Ｄｉｇｉｔａｌ Ｖｅｒｓａｔｉｌｅ Ｄｉｓｃ（ＤＶＤ）ディスク、Ｕｎｉｖｅｒｓａｌ Ｓｅｒｉａｌ Ｂｕｓ（ＵＳＢ）メモリ等のコンピュータで読み取り可能な可搬型記録媒体に記録され、コンピュータによって記録媒体から読み出されることによって実行される。また本送信制御プログラムは、インターネット等のネットワークを介して配布してもよい。

上述した実施の形態に関し、さらに以下の付記を開示する。

（付記１）システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、
前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、特定した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較し、
比較結果に基づいて、前記複数のノードと通信する通信部を制御して、前記送信先ノードに前記データを送信する、
処理を前記第１ノードに実行させることを特徴とする送信制御プログラム。

（付記２）前記送信する処理は、
前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度が前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度より小さい場合、前記通信部を制御して、前記送信先ノードに前記データを送信することを特徴とする付記１に記載の送信制御プログラム。

（付記３）前記比較する処理は、
複数の第２ノードが特定された場合、前記複数の第２ノードの各々の第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度のうちの最小の影響度と、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする付記２に記載の送信制御プログラム。

（付記４）前記特定する処理は、
前記データに基づいて、前記複数のノードから前記第２ノードを特定することを特徴とする付記１〜３のいずれか一つに記載の送信制御プログラム。

（付記５）前記第１ノードに、
前記第１ノードと前記送信先ノードとの通信を中継するスイッチ装置の数に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
前記第２ノードと前記送信先ノードとの通信を中継するスイッチ装置の数に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
前記比較する処理は、
算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする付記１〜４のいずれか一つに記載の送信制御プログラム。

（付記６）前記第１ノードに、
前記第１ノードと前記送信先ノードとの通信の帯域幅に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
前記第２ノードと前記送信先ノードとの通信の帯域幅に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
前記比較する処理は、
算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする付記１〜５のいずれか一つに記載の送信制御プログラム。

（付記７）前記第１ノードに、
前記第１ノードのプロセッサまたは前記第１ノードのメモリの使用率に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
前記第２ノードのプロセッサまたは前記第２ノードのメモリの使用率に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
前記比較する処理は、
算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする付記１〜６のいずれか一つに記載の送信制御プログラム。

（付記８）システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、
前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、特定した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較し、
比較結果に基づいて、前記複数のノードと通信する通信部を制御して、前記送信先ノードに前記データを送信する、
処理を前記第１ノードに実行させる送信制御プログラムを記録した前記第１ノードに読み取り可能な記録媒体。

（付記９）システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定する特定部と、
前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、前記特定部によって特定された前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較する比較部と、
前記比較部による比較結果に基づいて、前記送信先ノードに前記データを送信する通信部と、
を有することを特徴とする通信ノード。

（付記１０）システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、
前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、特定した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較し、
比較結果に基づいて、前記複数のノードと通信する通信部を制御して、前記送信先ノードに前記データを送信する、
処理を前記第１ノードが実行することを特徴とする送信制御方法。

１００ 分散処理システム
１０１ ノード
１０２ スイッチ
７０１ 受付部
７０２ 特定部
７０３ 算出部
７０４ 比較部
７０５ 送信制御部
７０６ 通信部
７１１ 経路テーブル

Claims (9)

1. システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、
   前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、特定した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較し、
   比較結果に基づいて、前記複数のノードと通信する通信部を制御して、前記送信先ノードに前記データを送信する、
   処理を前記第１ノードに実行させることを特徴とする送信制御プログラム。
2. 前記送信する処理は、
   前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度が前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度より小さい場合、前記通信部を制御して、前記送信先ノードに前記データを送信することを特徴とする請求項１に記載の送信制御プログラム。
3. 前記比較する処理は、
   複数の第２ノードが特定された場合、前記複数の第２ノードの各々の第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度のうちの最小の影響度と、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする請求項２に記載の送信制御プログラム。
4. 前記特定する処理は、
   前記データに基づいて、前記複数のノードから前記第２ノードを特定することを特徴とする請求項１〜３のいずれか一つに記載の送信制御プログラム。
5. 前記第１ノードに、
   前記第１ノードと前記送信先ノードとの通信を中継するスイッチ装置の数に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
   前記第２ノードと前記送信先ノードとの通信を中継するスイッチ装置の数に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
   前記比較する処理は、
   算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする請求項１〜４のいずれか一つに記載の送信制御プログラム。
6. 前記第１ノードに、
   前記第１ノードと前記送信先ノードとの通信の帯域幅に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
   前記第２ノードと前記送信先ノードとの通信の帯域幅に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
   前記比較する処理は、
   算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする請求項１〜５のいずれか一つに記載の送信制御プログラム。
7. 前記第１ノードに、
   前記第１ノードのプロセッサまたは前記第１ノードのメモリの使用率に基づいて、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出し、
   前記第２ノードのプロセッサまたは前記第２ノードのメモリの使用率に基づいて、前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度を算出する処理を実行させ、
   前記比較する処理は、
   算出した前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、算出した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較することを特徴とする請求項１〜６のいずれか一つに記載の送信制御プログラム。
8. システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定する特定部と、
   前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、前記特定部によって特定された前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較する比較部と、
   前記比較部による比較結果に基づいて、前記送信先ノードに前記データを送信する通信部と、
   を有することを特徴とする通信ノード。
9. システムに含まれる複数のノードから、第１ノードが記憶するデータと同一の内容のデータを記憶する第２ノードを特定し、
   前記複数のノードのうちの前記データの送信先となる送信先ノードと前記複数のノードの各々のノードとの通信が前記システムの性能に与える影響度合いを表す影響度を前記各々のノードに対応して記憶する記憶部を参照して、前記第１ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、特定した前記第２ノードと前記送信先ノードとの通信が前記システムの性能に与える影響度合いを表す影響度と、を比較し、
   比較結果に基づいて、前記複数のノードと通信する通信部を制御して、前記送信先ノードに前記データを送信する、
   処理を前記第１ノードが実行することを特徴とする送信制御方法。

Images