JP7435735B2 - 分散処理システム、分散処理システムの制御方法、及び、分散処理システムの制御装置 - Google Patents

Description

後述の実施の形態は、例えば、分散処理システム、分散処理システム制御方法、及び、分散処理システム制御装置に関する。

以下の開示は例示であり、本願や本発明の範囲を限定するものではない。

一般に、２以上のコンピュータから構成される分散処理システムが利用されている。分散処理システムにおいては、例えば、実行を要求されたジョブを複数のノードにデプロイして並列実行させる、分散並列処理を実行し得る。分散処理システムで用いられる並列分散ストレージは、1つのファイルを一定のサイズ（例えば128 MB）に分割したブロックと呼ばれるデータを格納し得る。並列分散ストレージは、２以上のノード、もしくはコンピュータから構成されてもよい。並列分散ストレージは、例えば、ブロックを3個に複製し、複数のノードに分散配置し得る。

あるノードにおいて、実行を要求されたジョブの処理対象のブロックが、そのあるノード上に格納されている場合、このノードのストレージからブロックを読み込み得る。また、そのあるノード上に処理対象のブロックが存在しない場合、処理対象ブロックを持つ他のノードにブロックの転送を要求し得る。

このように、ジョブを実行するノードに処理対象ブロックが存在しない時に、ブロックのネットワーク転送が発生し得る。ブロックのネットワーク転送が頻発する場合、ネットワーク負荷の増大のため、分散処理システム全体の性能が低下することがあり得る。

例えば、大量ユーザが分散処理システムに同時アクセスし、大量のジョブが実行されることがあり得る。このような状況では、多くのジョブが処理対象ブロックの格納されているノード以外で実行されることがあり得る。こうした場合、処理対象ブロックのネットワーク転送が頻発し得る。このため、分散処理システムを構成するクラスタ全体でデータ転送量が多くなることがあり得る。

後述の特許文献のうち、例えば、特許文献4は、「分散システム」に関連し得る。しかし、特許文献4の全体のアーキテクチャは、上記のようなネットワーク負荷の増大を解消し得るものではない。

特許文献4では、移動元と移動先のブロックを入れ替えるものであり得る（図4、段落0034から0036など）。しかし、単にブロックを入れ替えても、あるブロックに対して同時接続で大量のアクセスが発生する場合、そのブロックを格納する特定のノードにアクセスが集中し得る。特許文献4では、大量のアクセス要求により、ボトルネックや応答の遅延を発生する可能性がある。

特許文献4にはApache Hadoopのレプリケーションの例も記載されている。とはいえ、特許文献4は、図4のステップ47にもあるように、ブロックの交換を行うにすぎない。特許文献4は、この交換の際に、Apache Hadoopのレプリケーションポリシーを参照し得る。即ち、特許文献4は、ブロックの交換を行うことを開示するにすぎない。特許文献4においては、あるブロックに対して同時接続で大量のアクセスが発生する場合、そのブロックを持っている特定のノードにアクセスが集中する可能性がありえる。実際、特許文献4は、どのようなタイミングでレプリケーションを行うのかについて、開示も示唆もしていない。

上述の開示は例示であり、本願や本発明の範囲を限定するものではない。

特開2017-191387号公報 特開2017-016404号公報 特表2015-532997号公報 特開2014-186364号公報 特開2004-126716号公報

以下は例示であり、本願や本発明の範囲を限定するものではない。

一般に、分散処理システムにおいて、ブロックのネットワーク転送をできるだけ抑え、ネットワーク負荷を抑制することが好ましい。これにより、分散処理システムにおいて、システム全体の性能を向上させることができ得る。

上記の点は、本願や本発明の範囲を限定するものではない。

以下の記載は例示であり、本願や本発明の範囲を限定するものではない。

後述の実施の形態においては、例えば、分散処理システムが開示され得る。この分散処理システムは、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照し、複製対象のブロックを選択し、ブロックの複製の作成の要求を送信する、マスタノードと、マスタノードからブロックの複製の作成の要求を受信してブロックを送信する、第1のワーカーノードと、ブロックを受信してブロックの複製を格納する、第2のワーカーノードと、を具備してもよい。

後述の実施の形態においては、例えば、分散処理システムの制御方法が開示され得る。この分散処理システムの制御方法は、マスタノードとワーカーノードとを含む分散処理システムの制御方法であって、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照して複製対象のブロックを選択することと、ブロックの複製を格納することと、を具備してもよい。

後述の実施の形態においては、例えば、分散処理システムにおける制御装置が開示され得る。この制御装置は、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照し、複製対象のブロックを選択し、ブロックの複製の作成の要求をワーカーノードへ送信する、ものであってもよい。

上記の点は、本願や本発明の範囲を限定するものではない。

後述の実施の形態によれば、例えば、分散処理システムにおいて、ブロックのネットワーク転送を抑え、ネットワーク負荷を抑制し得る。ただし、これは、本願や本発明の範囲を限定するものではない。

分散処理システムの第１の実施形態を示す概略図である。 処理が実行されるタイミングを示す、タイミングチャートである。 全体ブロック転送情報の例を示す図である。 全体ブロック読込情報の例を示す図である。 ブロック転送情報を取得する処理を説明するための図である。 ブロック転送情報を取得する処理を説明するための図である。 ブロック読込情報を取得する処理を説明するための図面である。 ブロック読込情報を取得する処理を説明するための図面である。 ブロック転送情報とブロック読込情報を収集する処理を示す図である。 ブロック転送情報とブロック読込情報を収集する処理を示す図である。 ブロック配分判定部の処理を表す概略図である。 図11の処理を詳細に説明するフローチャートである。 図12のステップE1の処理の例を表す図である。 ブロック位置情報の例を示す図である。 分散処理システムの例を示す図である。 ジョブの並列分散処理を説明する図である。 ジョブの並列分散処理を説明する図である。 クライアントが分散処理システムにファイルの書き込みを要求する場合の処理を示す、概略図である。 分散処理システムでのジョブの実行に関する概略図である。 コンピュータの構成の例を示す図である。 分散処理システムの制御装置を表す図面である。 分散処理システムの制御装置の処理の流れを表すフローチャートである。

以下、図面を参照し、いくつかの実施形態について説明する。以下の開示は例示である。本願と本発明は、以下の開示のみに限定して解釈されるべきではない。

「分散処理システムの例」
分散処理システムの例としては、例えば、Apache Hadoopがある。分散処理システムは、並列分散処理の能力と、並列分散ストレージ領域と、を併せ持つシステムであり得る。図15は、分散処理システムの例を示す図である。図15に示す分散処理システムもしくはクラスタサーバシステムは、マスタノードと少なくとも１つのワーカーノードから構成され得る。

マスタノードはクラスタサーバシステム全体のリソース管理を行う制御装置もしくはコンピュータである。マスタノードは、クラスタリソース管理部と、クラスタデータ管理部を含む。マスタノードはブロック位置情報を格納するマスタノードストレージを有する。

ワーカーノードは、マスタノードにネットワークを介して接続されたコンピュータである。ワーカーノードは、ジョブの実行やデータの格納を行う。ワーカーノードはジョブ実行管理部と、データ管理部と、データ通信部を含む。ワーカーノードはワーカーノードストレージを有する。この分散処理システム／クラスタサーバシステムでは、複数（1からn、nは１以上の自然数）のワーカーノードに、データを分散して格納する。この分散処理システム／クラスタサーバシステムでは、ジョブの並列分散処理を行う。

クライアントは、ネットワークを介してこの分散処理システム／クラスタサーバシステムに接続されたコンピュータである。例えば、図15のクラスタサーバシステムは、クライアントからのリクエストに応じて、ジョブを実行する。

図16、17は、ジョブの並列分散処理の動作を説明する図である。クライアントは、マスタノードに、ジョブ実行のリクエストを送信する（図16、「1.ジョブ実行をリクエスト」）。マスタノードは、クライアントからのリクエストを受信する。マスタノードは、受信したリクエストに対応するジョブを実行するために、ジョブ管理プロセスを起動するために十分なリソースを持ったワーカーノードを選択する。ジョブは、並列に実行され得る複数のジョブを含み得る。マスタノードは、選択したワーカーノードに割り当てるジョブを決定する。マスタノードは、ジョブ管理プロセスの起動を、選択したワーカーノードに要求する（図16、「2.ジョブ管理プロセスの起動を割り当て」）。

要求を受信したワーカーノードのジョブ実行管理部は、ジョブ管理プロセスを起動する（図16、「3.ジョブ管理プロセス起動」）。起動されたジョブ管理プロセスは、マスタノードから割り当てられたジョブを実行するためのリソースを、マスタノードから取得する（図17、「4.ジョブ実行リソースを要求」、「5.ジョブ実行可能リソースを取得」）。換言すれば、図16、17では、ジョブを実行するワーカーノードが選択され、選択されたワーカーノードにジョブが割り当てられ、ジョブが並列分散実行される。

ここで、分散処理システムにおける並列分散ストレージについて説明する。並列分散ストレージにおいては、1つのファイルが所定のサイズ（例えば128MB）のデータに分割される。ここで、分割されたデータのことをブロックという。並列分散ストレージにおいては、ブロックが複製され（例えば3個の複製）、ワーカーノードに格納される。

図18は、クライアントが分散処理システムにファイルの書き込みを要求する場合の処理を示す、概略図である。クライアントはマスタノードにファイルのクラスタサーバシステムへの書き込みを要求する。このとき、クライアントは、書き込まれるファイルのサイズをマスタノードに通知する。マスタノードのクラスタデータ管理部は、クライアントから要求を受信する（図18、「1.書き込み先の要求」）。クラスタデータ管理部は、書き込み対象のファイルを128MBのブロックに分割する場合、いくつのブロックに分割されるのかを計算する。ファイルのサイズが384MBである場合、このファイルは、3個のブロックに分割されることが分かる。クラスタデータ管理部は、3個のワーカーノードをランダムに選択する。例えば、クラスタデータ管理部は、ワーカーノード1、2、及び3を選択する。

クラスタデータ管理部は、クライアントに、書き込み先となるワーカーノードの情報を送信する。クライアントは、クラスタデータ管理部から、書き込み先となるワーカーノードの情報を受信する（図18、「2.書き込み先を通知」）。

クライアントは、書き込み先のワーカーノード（1、2、及び3）のデータ管理部に、分割したファイル（ブロック）の書き込みを要求する（図18、「3.分割したブロックの書き込み要求」）。書き込み要求を受けたワーカーノードは、自身のワーカーノードストレージにブロックを書き込む。ワーカーノードは、さらに、他のワーカーノードに複製したブロックの書き込みを要求する。例えば、ワーカーノード1は、ワーカーノード2に、複製の書き込みを要求する。複製の数は所定の数であってよい。複製の書き込みは、所定の書き込み先であってもよい。このように、クラスタサーバシステムは、所定の数のブロックの複製を作成し、複数のワーカーノードに分散格納する。

図19は、分散処理システム／クラスタサーバシステムでのジョブの実行に関する概略図である。分散処理システムにおいて実行中のジョブは、分散処理システムに格納されたブロックを参照する。ジョブによって参照されるブロックが、ジョブが実行されているワーカーノード上に格納されている場合、このワーカーノードのワーカーノードストレージからブロックを読み込む。一方、ジョブによって参照されるブロックが、ジョブが実行されているワーカーノード上に格納されていない場合、ジョブによって参照されるブロックを格納するワーカーノードからブロックを転送する処理が行われる。図19はこの処理を示している。ジョブが実行されるワーカーノード2にジョブによって参照されるブロックBがない場合、例えば、ワーカーノード2はワーカーノード3にブロックBの送信を要求する。ワーカーノード3はブロックBを送信し、ワーカーノード2はジョブを実行する。

「第1の実施形態」
図1は、分散処理システムの第１の実施形態を示す概略図である。図1に示されるように、この実施形態は、クライアント1と、クライアント１が接続するクラスタサーバシステム2（分散処理システム）とを含む。以下の説明では、上記の分散処理システムの例と重複する部分を省略している。

図１のクラスタサーバシステム2は、複数のサーバマシンから構成され得る。各サーバマシンは、図20に示すように、CPU201、メモリ202、ストレージ203（HDD（hard disk drive）やSSD（solid state disk））、送受信機204などを備えたコンピュータであってもよい。各サーバマシンには、オペレーティングシステム(OS)がインストールされている。また、各サーバマシンは物理マシン、仮想マシンのいずれであってもよい。各サーバマシンは、同じ筐体内に配置された複数のコンピュータであってもよい。各サーバマシンは、それらの全部もしくは一部が、異なる筐体に配置されてもよい。各サーバマシンは、ネットワークを介して相互に接続され、それらの全部もしくは一部が遠隔地に配置されてもよい。

クラスタサーバシステム2のサーバマシンは、マスタノード21とワーカーノード群22を含む。マスタノード21は、クラスタサーバシステム２の全体のリソース管理を行う。ワーカーノード群22の各々のワーカーノード１からn（nは自然数）は、ジョブを実行し、および／または、データ（ブロック）を格納する。マスタノード21は複数のマスタノードから構成されていても良い。ワーカーノード群22は、複数のワーカーノード221～22nから構成されている。「n」はワーカーノードの数がｎであることを表している。

マスタノード21は、クラスタリソース管理部211と、クラスタデータ管理部212と、クラスタ全体のブロック情報収集部213と、ブロック配分判定部214と、マスタノードストレージ215と、を有する。これらの構成要素はコンピュータプログラムとして実装され、マスタノード21のCPU（central processing unit）で実行されてもよい。

クラスタリソース管理部211は、ジョブ実行時にワーカーノード１からnへのジョブの割り当てを行うために、クラスタサーバシステム2全体のサーバマシンのリソースの管理を行う。クラスタデータ管理部212は、各ブロックの格納場所と、各ワーカーノードのワーカーノードストレージの空き容量を含む状態を管理する。

クラスタ全体のブロック情報収集部213は、ワーカーノード群22から後述するブロック転送情報22151とブロック読込情報22152とを収集して保存する。ブロック配分判定部214は、複製数を増減する必要のあるブロックを判定し、対応するワーカーノードに複製指示を行う。

マスタノードストレージ215は、例えば、HDDあるいはSSDなどのストレージデバイスである。マスタノードストレージ215は、ブロック位置情報2151と、全体ブロック転送情報2152と、全体ブロック読込情報2153と、を有する。

ブロック位置情報2151は、クラスタサーバシステム2に格納されている各々のファイルについて、ファイルを構成しているブロックがどのワーカーノードに格納されているのかを示す、情報である。全体ブロック転送情報2152は、クラスタサーバシステム2全体でワーカーノード間のブロックの転送が発生した回数を記録した情報である。全体ブロック読込情報2153は、クラスタサーバシステム2全体におけるブロック毎の読み込み回数を記録した情報である。

次に、ワーカーノード221から22nについて説明する。例えば、ワーカーノード221は、ジョブ実行管理部2211と、データ管理部2212と、データ通信部2213と、ブロック情報収集部2214と、ワーカーノードストレージ2215と、を有する。ジョブ実行管理部2211は、ジョブの割り当てや実行を行う。データ管理部2212は、ブロックの読み込みや書き込みを行う。データ通信部2213は、ブロックを他ノードに転送する。ブロック情報収集部2214については後述する。

ワーカーノード221のワーカーノードストレージ2215は、ブロックを格納する。更に、ワーカーノードストレージ2215は、ブロック転送情報22151とブロック読込情報22152を格納する。ブロック転送情報22151は、ワーカーノード1から他のワーカーノードにブロックを転送した回数を記録した情報である。ブロック読込情報22152は、各々のブロックの読み込み回数を記録した情報である。

ブロック情報収集部2214は、ブロック転送情報22151とブロック読込情報22152を収集する。ブロック情報収集部2214は、ブロック転送情報22151とブロック読込情報22152を、マスタノード21のクラスタ全体のブロック情報収集部213に、送信する。この送信は、例えば、所定の時間間隔で行われてもよい。あるいは、この送信は、クラスタサーバシステム（クラスタサーバ管理システム）2の管理者の指示を受けて行われてもよい。

ワーカーノード222～22nは、ワーカーノード221と同様の構成を有する。また、クラスタサーバシステム2は、ワーカーノード221～22nが持つワーカーノードストレージ2215は、クライアントの視点から、論理的に1つのファイルシステムとして扱うことができるよう、構成されてもよい。これにより、分散ストレージが実現され得る。各ワーカーノードストレージは、1つもしくは複数のHDDやSSDを有してもよい。各ワーカーノードストレージは、ネットワークストレージあるいはストレージエリアネットワークにアクセスするよう構成されてもよい。

以下、本実施形態におけるクラスタサーバシステム2の処理の流れが説明される。ここでは、ワーカーノード群22の全部もしくは一部のワーカーノードにデータ（ブロック）がすでに格納されているものとする。

図2は、後述の処理が実行されるタイミングを示す、タイミングチャートである。マスタノード21のブロック配分判定部214は、ブロックの複製数増減処理（ブロック配分判定）を、ユーザの指定したタイミングで実行してもよい。このタイミングは、例えば、24時間ごとの午前0時であってもよい。あるいは、マスタノード21のブロック配分判定部214は、ブロックの複製数増減処理（ブロック配分判定）を、ユーザの指示が入力されたときに実行してもよい。あるいは、マスタノード21のブロック配分判定部214は、ブロックの複製数増減処理（ブロック配分判定）を、少なくとも１つのワーカーノードのワーカーノードストレージ2215の使用率が所定の閾値に達したことが検出されたときに実行してもよい。この所定の閾値は、例えば、90%であってもよい。あるいは、ユーザがこの所定の閾値を指定してもよい。

また、マスタノード21のクラスタ全体のブロック情報収集部213は、ブロックの複製数増減処理（ブロック配分判定）が実行されるときに、全体ブロック転送情報2152と全体ブロック読込情報2153とに保存されている値をゼロにリセットしてもよい。このリセット処理は、ブロックの複製数増減処理（ブロック配分判定）が実行されるときに、毎回、もしくは所定の回数ごとに実行されてもよい。あるいは、このリセット処理は、ユーザからの指示を入力したときに実行されてもよい。

ブロックの複製数増減処理（ブロック配分判定）が実行された後、次のブロックの複製数増減処理（ブロック配分判定）が実行までの間に、各ワーカーノードでブロックの転送回数がカウントされ、全体ブロック転送情報2152が更新される。ブロックの複製数増減処理（ブロック配分判定）が実行された後、次のブロックの複製数増減処理（ブロック配分判定）が実行までの間に、各ワーカーノードでブロックの読み込み回数がカウントされ、全体ブロック読込情報2153が更新される。

図3は、全体ブロック転送情報2152の例を示す図である。全体ブロック転送情報2152はクラスタサーバシステム2全体でワーカーノード間のブロックの転送が発生した回数を記録するための情報である。図3のA、B、C、…は、ブロックの識別情報であってもよい。例えば、図3は、ワーカーノード223（図3の「#3」）からブロックBが他のワーカーノードへ12回送信されたことを表している。例えば、図3は、ワーカーノード225（図3の「#5」）からブロックDが他のワーカーノードへ2回送信されたことを表している。

図4は全体ブロック読込情報2153の例を示す図である。全体ブロック読込情報2153はクラスタサーバシステム2全体における各ブロックが読み込まれた回数を記録するための情報である。図4は、例えば、ブロックBがクラスタサーバシステム2全体で20回読み込まれたことを表している。

次に、図5と6は、ワーカーノード221～22nにおいて、ワーカーノードストレージ2215に格納されるブロック転送情報22151を取得する処理を説明するための図である。
ブロック転送情報22151は、他のワーカーノードにブロックを転送した回数を記録する情報である。ブロック転送情報22151は、ブロックに関する識別情報などのブロック情報と、転送回数とを、関連付けたテーブルのような情報であってもよい。例えば、ワーカーノード222でジョブが実行されていると仮定する。ワーカーノード222にこのジョブの処理対象のブロックBが格納されていない場合、ブロックBを持っているワーカーノード223のデータ通信部が、ワーカーノード222のデータ通信部へ、ブロックBを送信する。この時、ワーカーノード223のデータ通信部は、ワーカーノード223のブロック情報収集部へ、ブロックBの送信があったことを通知する。ワーカーノード223のブロック情報収集部は、ワーカーノード223のデータ転送部からブロック送信があったことを認識もしくは確認する（図6のステップA1）。ワーカーノード223のブロック情報収集部は、送信されたブロックBの転送回数のカウントをアップするため、ワーカーストレージノードのブロック転送情報にアクセスして更新／格納する（図6のステップA2）。上記の処理は他のワーカーノードについても同様である。

次に、図7と8は、ワーカーノード221～22nにおいて、ワーカーノードストレージ2215に格納されているブロック読込情報22152を取得する処理を説明するための図である。ブロック読込情報はブロック毎の読み込み回数を記録する情報である。ブロック読込情報はブロックに関する識別子などのブロック情報とその読み込み回数を関連付けたテーブルのような情報であってもよい。例えば、ワーカーノード223にあるデータ管理部がワーカーストレージノードからブロックBを読み込む時、データ管理部は、ブロック情報収集部にブロックBを読み込むことを通知する。ブロック情報収集部は読み込まれたブロックBのブロック情報を認識／確認する（図8のステップB1）。ブロック情報収集部は読み込まれたブロックBの読込回数のカウントをアップするため、ワーカーノード223のワーカーストレージノードのブロック読込情報にアクセスして更新／格納する（図8のステップB2）。上記の処理は、他のワーカーノードも同様に行う。

図9と図10は、マスタノード21が、ワーカーノード221～22nのブロック転送情報とブロック読込情報を収集する処理を示す図である。ワーカーノード221～22nにおいて、ブロック情報収集部は、ワーカーノードストレージが格納しているブロック転送情報およびブロック読込情報を、マスタノード21のクラスタ全体のブロック情報収集部213へ送信する（図10のステップC1）。この送信処理は、あらかじめ指定された日時に、例えば、毎日午前0時に実行してもよい。あるいは、この送信処理は、クラスタサーバシステム2が、ユーザもしくは管理者の指示を入力したときに実行してもよい。クラスタ全体のブロック情報収集部213は、各ワーカーノード221～22nから受信したブロック転送情報を、マスタノードストレージ215の全体ブロック転送情報2152に保存／格納する（図10のステップC2）。クラスタ全体のブロック情報収集部213は、各ワーカーノード221～22nから受信したブロック読込情報を、マスタノードストレージ215の全体ブロック読込情報2153に保存／格納する（図10のステップC2）。

ワーカーノード221乃至22nが、マスタノード21へブロック転送情報とブロック読込情報を送信するタイミングは、ユーザもしくは管理者が設定可能であってもよい。また、これらが送信された後で、ワーカーノード221乃至22nのワーカーノードストレージに格納されたブロック転送情報とブロック読込情報の値をゼロにリセットしてもよい。あるいは、ワーカーノード221乃至22nは、ブロック転送情報とブロック読込情報を所定の回数送信したときに、ワーカーノードストレージに格納されたブロック転送情報とブロック読込情報をゼロにリセットしてもよい。

以下、図11、12、13、及び14を参照し、マスタノード21のブロック配分判定部214の処理の説明をする。図11は、ブロック配分判定部214の処理を表す概略図である。図１１の構成要素の符号番号は、図１の構成要素の符号番号と同一である。図11では、図面を見やすくするため、一部の符号番号が省略されている。

図11において、マスタノード21のマスタノードストレージ215は、全体ブロック読込情報2153と全体ブロック転送情報2152とを保存／格納する。ブロック配分判定部214は、全体ブロック読込情報2153と全体ブロック転送情報2152を参照して、複製対象および削除対象のブロックを判定する（図11のステップS1101）。ここでは、ブロック配分判定部214は、ブロックBの複製を生成することを決定する。ここでは、ブロック配分判定部214は、ブロックAの削除することを決定する。ブロック配分判定部214は、ワーカーノードのデータ管理部に複製指示および／または削除指示を送信／出力する。図11では、ブロック配分判定部214が、ワーカーノード223のデータ管理部に複製指示を送信／出力する（図11のステップS1102）。図11では、ブロック配分判定部214が、ワーカーノード22nのデータ管理部に削除指示を送信／出力する（図11のステップS1103）。

図11において、例えば、ブロック配分判定部214は、ブロックBを複製対象と判定する（図11のステップS1101）。ワーカーノード223は、ブロックBが格納されているワーカーノードである。ワーカーノード222は、ブロックBが格納されていないワーカーノードである。マスタノード21からの指示に従い、ワーカーノード223のデータ管理部は、ワーカーノードストレージからブロックBを読み出す。ワーカーノード223のデータ管理部は、ワーカーノード222のデータ管理部へ読み出したブロックBを送信し、ブロックBの複製数を1つ格納する（増やす）ことを要求する。ワーカーノード222のデータ管理部は、ワーカーノード223のデータ管理部からブロックBを受信し、ブロックBの複製をワーカーノードストレージに格納する。

一方、ブロックAは、例えば、過去において利用頻度が高いと判定されたために4つの複製が格納されていたものである。マスタノード21のブロック配分判定部214は、例えば、ブロックAの利用頻度が下がったと判定する。この場合に、マスタノード21のブロック配分判定部214は、ブロックAの複製数を1個削除するべきであると判断する。図11において、マスタノード21のブロック配分判定部214から削除指示を受信した場合に（図11のステップS1103）、ワーカーノード22nのデータ管理部は、ワーカーノードストレージからブロックAを削除する（図11のステップS1105）。

図12は、図11の処理を詳細に説明するフローチャートである。図12は、ブロック配分判定部214の処理の流れを示している。まず、ブロックの複製数を増加させる処理を説明する（図12のステップE1からE6）。

マスタノード21のブロック配分判定部214は、マスタノードストレージ215に保存／格納されている全体ブロック転送情報2152を読み込む。ブロック配分判定部214は、全体ブロック転送情報2152の集計を行う。この集計処理において、ブロック配分判定部214は、全体ブロック転送情報2152中のブロックごとワーカーノードごとの転送回数の間の最大値を計算してもよい。あるいは、ブロック配分判定部214は、ブロックごとの平均値を計算してもよい。ブロック配分判定部214は、集計した値が閾値（以下、複製閾値という）以上となったブロックを複製対象として選出する（図12のステップE1）。マスタノード21のユーザもしくは管理者が、この複製閾値を予め設定してもよい。複製閾値を使用する代わりに、例えば、ブロック配分判定部214は、転送回数の上位の所定個数のブロックを複製対象としてもよい。

図13は、図12のステップE1の処理の例を表す図である。この例では、ブロック配分判定部214は、全体ブロック転送情報2152を集計して集計結果（ブロックごと、ワーカーノードごとの最大値）を得る。図13では、例えば、ブロックAの集計結果が「6」である。これは、ブロックAの転送回数の最大値が6回であることを示している。図13では、例えば、ブロックCの集計結果が「4」である。これは、ブロックCの転送回数の最大値が4回であることを示している。全体ブロック転送情報2152は、ブロック番号も格納してもよい。ブロック番号は、クラスタサーバシステム2において一意の番号であってもよい。例えば、ブロックAのブロック番号を「345」であるとする。全体ブロック転送情報2152は、「345」を「A」に関連付けて格納してもよい。ブロック配分判定部214は、このブロック番号をもとに、図14のブロック位置情報2151に格納されたブロックに関する情報と対応を取ってもよい。

図13の例では、複製閾値を10としている。ブロック配分判定部214は、複製閾値である転送回数が10以上のブロックBとEを複製対象ブロックとして選出する。

以下、図12を参照する。ブロック配分判定部214は、選出した複製対象ブロックを1つずつ選択する（図12のステップE2）。ブロック配分判定部214は、選択したブロックのワーカーノードにおける所在を示すブロック位置情報2151を参照する。ブロック配分判定部214は、選択したブロック（複製対象ブロック）の現在の複製数を算出する（図12のステップE3）。

ここで、図14はブロック位置情報2151の例を示す図である。図14において、「/user/user1/employee.txt」はファイルパスを含むファイル名である。このファイルパスは、クラスタサーバシステム2においてユーザ（クライアント）から見たパスであってもよい。図14は、このファイルが３個のブロックから構成されていることを示している。それぞれのブロック名をブロックA、ブロックB、及びブロックCとする。ブロックAは、ブロックの識別情報であるブロック番号345に対応するものであってもよい。この例では、理解のため、ブロック位置情報がブロック名を含んでいる。ブロックを識別するために、ブロック名の代わりにブロック番号が用いられてもよい。ブロック番号は、クラスタサーバシステム2において一意の番号であってもよい。あるいは、ブロックの識別のため、ファイル名を含むファイルのパスと、ブロック番号とを組み合わせて用いてもよい。

図14は、ブロックAおよびその複製が、ワーカーノード221、222、223、224に格納されていることを、示している。図14では、ワーカーノード221を、簡単のため、「1」として表している。他のワーカーノードも同様である。同じように、図14は、ブロックBおよびその複製が、ワーカーノード221、223、224に格納されているといった情報を有していることを、示している。ブロック配分判定部214は、図14のブロック位置情報を参照し、例えば、ブロックAが合計4個格納されていることもわかる。即ち、ブロック配分判定部214は、ブロックAの現在の複製数を4と算出する。他のブロックについても現在の複製数が算出され得る（図12のステップE3）。

以下、図12を参照する。例えば、ブロック配分判定部214は、ブロック位置情報2151から算出した選択したブロック（複製対象ブロック）の現在の複製数が、複製数の既定値より大きく、全体のノード数以下の範囲外であれば、複製対象ブロックの複製は作成しないこととしてもよい（図12のステップE4）。例えば、全てのワーカーノードに選択したブロック（複製対象ブロック）がすでに格納されている場合、このブロックがさらに複製されると無駄となり得る。この判定によって、このような無駄な複製を回避し得る。ひいては、クラスタサーバシステム2全体で、ワーカーノードストレージの使用率を向上し得る。複製数の既定値は、ユーザもしくは管理者が予め設定してもよい。複製数の既定値は、例えば、ワーカーノード数の半分であってもよい。複製数の既定値は、例えば、0（ゼロ）であってもよい。この場合、ワーカーノードの数と等しくなるまで常に、複製の作成が可能となる。

また、図12のステップE4において、例えば、ブロック配分判定部214は、ブロック位置情報2151から算出した選択したブロック（複製対象ブロック）の現在の複製数が、複製数の既定値より大きい場合（複製数の既定値＜現在の複製数）に、複製対象ブロックの複製を作成しないこととしてもよい。

現在の複製数が図12のステップE4の条件をみたす場合（図12のステップE4、「Yes」）、ブロック配分判定部214は、複製対象ブロックを格納していないワーカーノードを全て選出する。ブロックの複製は、１ノードに１個までしか作成されないこととなる。この処理において、ブロック配分判定部214は、ブロック位置情報を参照してもよい。ブロック配分判定部214は、選出されたワーカーノードに、ワーカーノードストレージに選択したブロック（複製対象ブロック）を格納するための十分な空き領域があるか否かを問い合わせる。ブロック配分判定部214は、この問い合わせの結果に基づき、選択したブロック（複製対象ブロック）を格納するための十分な空き領域があるワーカーノードを、選択したブロック（複製対象ブロック）の複製を格納するための候補のワーカーノードとする。

ブロック配分判定部214は、候補のワーカーノードの中からランダムに1つのワーカーノードを選択する。この処理において、例えば、ブロック配分判定部214は、ワーカーノードストレージに最も大きな空き領域があるワーカーノードを選んでもよい。ブロック配分判定部214は、候補のワーカーノードのリストを作成し、所定の乱数を用いて1つのワーカーノードを選択してもよい。あるいは、乱数の代わりに現在時刻をマイクロ秒単位で取得し、末尾の所定の桁数の数字を用いてもよい。選択されたワーカーノードを、格納先ワーカーノードとする。

ブロック配分判定部214は、ブロック位置情報を参照して、選択したブロック（複製対象ブロック）がすでに格納されているワーカーノードの1つを送信元ワーカーノードとする。選択したブロック（複製対象ブロック）がすでに格納されているワーカーノードが複数ある場合、ブロック配分判定部214は、これらの中の１つを任意に選択してもよい。例えば、ブロック配分判定部214は、図14の例では、先頭に位置するワーカーノードを選択してもよい。例えば、ブロック配分判定部214は、各ワーカーノードにpingなどの信号を送信し、最も早い時間内に返信が受信されたノードを選択してもよい。

ブロック配分判定部214は、選択したブロック（複製対象ブロック）を格納先ワーカーノードに送信するよう、送信元ワーカーノードに対して要求する。ブロック配分判定部214は、選択したブロック（複製対象ブロック）を格納するよう格納先ワーカーノードに対して要求する。送信元ワーカーノードは選択したブロック（複製対象ブロック）を格納先ワーカーノードに送信する。格納先ワーカーノードは、送信元ワーカーノードから選択したブロック（複製対象ブロック）を受信し、選択したブロック（複製対象ブロック）を格納する（図12のステップE5）。ブロック配分判定部214は、図12のステップE2で選出された全てのブロックに対して、上記の処理を繰り返す（図12のステップE6）。

次に、図12を参照してブロックの複製数を減少させる処理を説明する（図12のステップE7からステップE10）。

ブロック配分判定部214は、複製数を減少する対象のブロックを選択する。ブロック配分判定部214は、ブロック位置情報2151を参照し、複製数が既定値より多いブロックを抽出する。図14は、ブロック位置情報2151の例を示す。この抽出のために、ブロック配分判定部214は、ブロック位置情報2151を参照し、各ブロックについてブロック及びその複製が格納されているワーカーノード（ノード）の数をカウントする。ブロックの複製数を増加させる処理（図12のステップE1からE6）で説明されたように、ブロックの複製は１ノードに１個までしか作成されない。このため、ノード数をカウントすれば、ブロックとその複製の数がカウントされる。

ブロック配分判定部214は、各ブロックについてブロック及びその複製の数を、既定値と比較する。ブロック配分判定部214は、既定値より多いと判断した場合に、該当するブロックを抽出する。この既定値は、ユーザによって予め指定された数であってもよい。この既定値は、ブロックの複製数を減少させる処理の開始時に、ブロック配分判定部214は、クラスタサーバシステム2全体のワーカーストレージの使用率が例えば90%といった所定の閾値よりも大きい場合に、この既定値から1を除算してもよい。

抽出されたブロックは複数個になることもある。ブロック配分判定部214は、図12のステップE1で選出した複製対象ブロックを、予め削除の対象外としてもよい（図12のステップE7）。ブロック配分判定部214は、クラスタ全体のブロック読込情報2153を参照し、抽出したブロックの読み込み回数を取得する（図12のステップE8）。ブロック配分判定部214は、抽出したブロックの各々について、読み込み回数が閾値（以下、READ閾値という）以下であるか否かを判断する（図12のステップE9）。

読み込み回数がREAD閾値以下であるブロックについて、ブロック配分判定部214は、ブロック位置情報を参照し、そのようなブロックが格納されているワーカーノードをランダムに選出する。この処理において、例えば、ブロック配分判定部214は、候補のワーカーノードのリストを作成し、所定の乱数を用いて1つのワーカーノードを選択してもよい。あるいは、乱数の代わりに現在時刻をマイクロ秒単位で取得し、末尾の所定の桁数の数字を用いてもよい。あるいは、ブロック配分判定部214は、ワーカーノードストレージに最も小さな空き領域があるワーカーノードを選んでもよい。

ブロック配分判定部214は、選択されたワーカーノードに削除の対象のブロックを削除するよう、要求する。選択されたワーカーノードは、ブロック配分判定部214からの要求を受信し、該当するブロックを削除する（図12のステップE10）。抽出されたブロックが複数個である場合、ブロック配分判定部214は、この処理を抽出されたブロック１つずつに関して実行する。

READ閾値はユーザが指定するものであってもよい。READ閾値を用いた上述のような処理により、クラスタサーバシステム2は、読み込み回数に基づいてブロックが利用される頻度を判断することができる。クラスタサーバシステム2は、あまり使用されていないと判断されたブロックを削除し得る。これにより、システム全体のストレージの有効な利用が図られ得る。一方、クラスタサーバシステム2は、READ閾値以上のブロックを利用される頻度が高いブロックであると判断し得る。クラスタサーバシステム2は、このようなブロックを削除せずに複製数を維持し得る。これにより、クラスタサーバシステム2は、アクセスの頻度の高いファイルを高速に読み出すことができ得る。

ブロックの複製数を増加させる処理と、ブロックの複製数を減少させる処理とは、順序が逆でもよい。ブロックの複製数を増加させる処理と、ブロックの複製数を減少させる処理とは、別個独立に実行されてもよい。

図20は、コンピュータの構成の例を示す。上述のノード、ワーカーノード、マスタノードなどは、図20と同一、もしくは類似の構成を有してもよい。図20の構成は、CPU（central processing unit、中央処理装置）、メモリ、ストレージ、送受信機を含み得る。ストレージはHDD(hard disk drive、ハードディスク)やSSD(solid state drive)であり得る。ストレージは、ストレージエリアネットワークを含み得る。複数のコンピュータが、ストレージを共有してもよい。ストレージは、ネットワーク上のストレージや、ストレージエリアネットワークを含み得る。図20のコンピュータは光学式ドライブや光磁気記録装置を更に有してもよい。送受信機は、10BASE-T、1000BASE-T、無線通信装置、光通信装置などであってもよい。図20は、コンピュータは、一般用途のコンピュータであってもよい。図20は、コンピュータは、OS（operating system）を含んでもよい。図20は、コンピュータは、複数のコンピュータから構成されてもよい。図20は、コンピュータは、1台のコンピュータに複数の実行環境を含むものでもよい。上記の実施形態は、ソフトウェアによって実装されてもよい。

図21は、本実施の形態における分散処理システムの制御装置（マスタノード31）を示す図である。マスタノード31は、複製対象のブロックを選択する選択手段311と、ワーカーノードへブロックの複製の作成要求を送信する要求手段312と、を含む。

図22は、図21の分散処理システムの制御装置（マスタノード31）の処理の流れを表すフローチャートである。選択手段311は、複製対象のブロックを選択する（図22のステップE11）。要求手段312は、複製要求をワーカーノードへ送信する（図22のステップE12）。

「他の側面」
上記の実施形態のほか、例えば、以下のような側面があってもよい。

第1の側面は、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照し、複製対象のブロックを選択し、ブロックの複製の作成の要求を送信する、マスタノードと、マスタノードからブロックの複製の作成の要求を受信して前記ブロックを送信する、第1のワーカーノードと、ブロックを受信して前記ブロックの複製を格納する、第2のワーカーノードと、を含む、分散処理システムであり得る。

第2の側面は、上記の分散処理システムであって、マスタノードが、ブロック位置情報から得た各ブロックの格納先のワーカーノードの数を参照し、ブロックの複製の作成の要求を送信するか否かを判断する、分散処理システムであり得る。

第3の側面は、上記の分散処理システムであって、マスタノードが、複製対象のブロックの複製の数が所定の数より多く、すべてのワーカーノードの数より少ない場合に、ブロックの複製の作成の要求を送信する、分散処理システムであり得る。

第4の側面は、上記の分散処理システムであって、マスタノードが、ブロックの複製の作成のための空きのリソースを有するワーカーノードを第2のワーカーノードとして選択する、分散処理システムでありえる。

第5の側面は、上記の分散処理システムであって、マスタノードが、複製対象のブロックを格納していないワーカーノードの中からランダムに第2のワーカーノードを選択する、分散処理システムでありえる。

第6の側面は、上記の分散処理システムであって、マスタノードが、各ブロックの格納先のワーカーノード示すブロック位置情報を参照して削除するブロックを判断し、削除するブロックを格納する第3のワーカーノードを選択して、第3のワーカーノードに削除の要求を送信し、第3のワーカーノードが、削除の要求を受信し、削除するブロックを削除する、分散処理システムでありえる。

第7の側面は、上記の分散処理システムであって、マスタノードが、ブロック位置情報から得た各ブロックの格納先のワーカーノードの数を参照し、格納先のワーカーノードの数が所定の数より多い場合に、第3のワーカーノードに削除の要求を送信する、分散処理システムであり得る。

第8の側面は、上記の分散処理システムであって、マスタノードが、削除するブロックを格納するワーカーノードの中からランダムに第3のワーカーノードを選択する、分散処理システムでありえる。

第9の側面は、マスタノードとワーカーノードとを含む分散処理システムの制御方法であって、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照して複製対象のブロックを選択するステップと、ブロックの複製を格納するステップと、を含む制御方法であり得る。

第10の側面は、分散処理システムにおける制御装置であって、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照し、複製対象のブロックを選択し、ブロックの複製の作成の要求をワーカーノードへ送信する、分散処理システムにおける制御装置であり得る。

上記の実施形態により、例えば、ブロックのネットワーク転送を抑え、ネットワーク負荷によるシステム全体の性能を向上させることが可能となり得る。上記の実施形態では、頻繁に処理対象となるブロックについて、複製数を増やし、各ワーカーノードに分散して格納し得る。このため、そのようなブロックを持つワーカーノード数をシステム全体として増やすことができる。これにより、他のワーカーノードからのブロックの転送の頻度を抑えることが可能となり得る。また、ジョブが実行されるワーカーノード上に処理対象ブロックが存在する確率を上げることができる。このため、ジョブの実行をより短い時間で完了することができ得る。

また、上記の実施形態によれば、複製数がすでに増加されたブロックについて、利用頻度が下がった場合、その複製ブロックを自動的に削除し得る。このため、上記の実施形態によれば、クラスタ全体でストレージリソースの逼迫を防ぐことが可能となり得る。

以上の開示は例である。従って、本願と本発明は、上記の実施形態などの開示のみに限定して解釈されるべきではない。

この出願は、２０２０年３月１７日に出願された日本国特願２０２０－０４６５４２号を基礎とする優先権を主張し、その開示の全てをここに取り込む。

本発明は、分散処理システム、分散処理システムの制御方法、及び、分散処理システムの制御装置に適用してもよい。

２１１ クラスタリソース管理部
２１２ クラスタデータ管理部
２１３ ブロック情報収集部
２１４ ブロック配分判定部
２１５ マスタノードストレージ
２２１ ワーカーノード
２２２ ワーカーノード
２２３ ワーカーノード
２２４ ワーカーノード
２２５ ワーカーノード
２１５１ ブロック位置情報
２１５２ 全体ブロック転送情報
２１５３ 全体ブロック読込情報
２２１１ ジョブ実行管理部
２２１２ データ管理部
２２１３ データ通信部
２２１４ ブロック情報収集部
２２１５ ワーカーノードストレージ
２２１５１ ブロック転送情報
２２１５２ ブロック読込情報

Claims (10)

1. 全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照し、複製対象のブロックを選択し、ブロックの複製の作成の要求を送信する、マスタノードと、
   前記マスタノードから前記ブロックの複製の作成の要求を受信して前記ブロックを送信する、第１のワーカーノードと、
   前記ブロックを受信して前記ブロックの複製を格納する、第２のワーカーノードと、
   を具備する前記分散処理システム。
2. 前記マスタノードが、ブロック位置情報から得た各ブロックの格納先のワーカーノードの数を参照し、前記ブロックの複製の作成の要求を送信するか否かを判断する、
   請求項１に記載の分散処理システム。
3. 前記マスタノードは、前記複製対象のブロックの複製の数が所定の数より多く、すべてのワーカーノードの数より少ない場合に、前記ブロックの複製の作成の要求を送信する、請求項１または２に記載の分散処理システム。
4. 前記マスタノードが、前記ブロックの複製の作成のための空きのリソースを有するワーカーノードを前記第２のワーカーノードとして選択する、請求項１乃至３のいずれか一項に記載の分散処理システム。
5. 前記マスタノードが、前記複製対象のブロックを格納していないワーカーノードの中からランダムに前記第２のワーカーノードを選択する、
   請求項１乃至４のいずれか一項に記載の分散処理システム。
6. 前記マスタノードが、各ブロックの格納先のワーカーノード示すブロック位置情報を参照して削除するブロックを判断し、前記削除するブロックを格納する第３のワーカーノードを選択して、前記第３のワーカーノードに削除の要求を送信し、
   前記第３のワーカーノードが、前記削除の要求を受信し、前記削除するブロックを削除する、
   請求項１乃至５のいずれか一項に記載の分散処理システム。
7. 前記マスタノードが、前記ブロック位置情報から得た各ブロックの格納先のワーカーノードの数を参照し、前記格納先のワーカーノードの数が所定の数より多い場合に、前記第３のワーカーノードに削除の要求を送信する、
   請求項６に記載の分散処理システム。
8. 前記マスタノードが、前記削除するブロックを格納するワーカーノードの中からランダムに前記第３のワーカーノードを選択する、請求項６または７に記載の分散処理システム。
9. マスタノードとワーカーノードとを含む分散処理システムにおける処理方法であって、前記処理方法は、
   前記マスタノードが、全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照して複製対象のブロックを選択し、ブロックの複製の作成の要求を送信することと、
   第１のワーカーノードが、前記マスタノードから前記ブロックの複製の作成の要求を受信して前記ブロックを送信することと、
   第２のワーカーノードが、前記ブロックを受信して前記ブロックの複製を格納することと、
   を具備する前記処理方法。
10. 全体ブロック転送情報に格納された分散処理システムにおけるブロックの転送回数を参照して複製対象のブロックを選択する選択手段と、
    第１のワーカーノードへブロックの複製の作成要求を送信する第１の要求手段と、
    第２のワーカーノードへ前記ブロックを受信して前記ブロックの複製を格納するための格納要求を送信する第２の要求手段と、
    を具備する分散処理システムの制御装置。

Images