(57)【要約】【課題】ＭａｐＲｅｄｕｃｅフレームワークを使用するシステムにおいて、各ワーカサーバに対して割り当てるデータサイズを適切に定めてシステム全体の処理性能を向上させる。【解決手段】本発明に係るデータ割当制御プログラムは、各ワーカサーバに対してデータを割り当てる前にあらかじめ各ワーカサーバのベンチマーク結果を取得し、その結果に基づき各ワーカサーバに対して割り当てるデータサイズを調整する。【選択図】図７
2015084152.tif 000002
【特許請求の範囲】
【請求項１】
ＭａｐＲｅｄｕｃｅフレームワークを使用してデータを分散処理する複数のワーカサーバに対して前記データを分割して割り当てる処理をコンピュータに実行させるデータ割当制御プログラムであって、
前記複数のワーカサーバのうち少なくともいずれかの計算リソースは、他の前記ワーカサーバが備える計算リソースとは処理性能が異なっており、
前記データ割当制御プログラムは、前記コンピュータに、
前記ワーカサーバの処理性能を測定するベンチマークのために用いるベンチマークデータを前記ワーカサーバに対して配布するベンチマークデータ配布ステップ、
前記ベンチマークデータを用いて前記ベンチマークを実施するよう要求するリクエストを前記ワーカサーバに対して発行しそのベンチマークの結果を受信するベンチマーク実行指示ステップ、
各前記ワーカサーバに対して前記データを分割して割り当てる際の前記データの配分割合を求める分割割合算出ステップ、
前記分割割合算出ステップの結果に基づき前記データを分割して各前記ワーカサーバに対して割り当てるデータ割当ステップ、
を実行させ、
前記分割割合算出ステップにおいては、前記コンピュータに、
前記ベンチマークの結果に基づき、前記割り当てられた前記データに対する各前記ワーカサーバの処理性能の合計が最大になる前記配分割合を求めさせる
ことを特徴とするデータ割当制御プログラム。
【請求項２】
前記ベンチマーク配布ステップにおいては、前記コンピュータに、
前記データ割当ステップにおいて前記ワーカサーバが割り当てを受ける前記データをサンプリングすることにより前記ベンチマークデータを作成させる  
ことを特徴とする請求項１記載のデータ割当制御プログラム。
【請求項３】
前記ベンチマーク実行指示ステップにおいては、前記コンピュータに、
各前記ワーカサーバが前記ベンチマークデータを用いて前記ベンチマークを実施した結果として、前記ベンチマークを実施する過程で処理した前記ベンチマークデータのサイズとその処理時間との間の対応関係を記述したデータを、前記ワーカサーバ毎に受信させる
ことを特徴とする請求項１記載のデータ割当制御プログラム。
【請求項４】
請求項１記載のデータ割当制御プログラムを実行するマスタサーバ、
ＭａｐＲｅｄｕｃｅフレームワークのＭａｐ関数を実行するＭａｐワーカサーバ、
ＭａｐＲｅｄｕｃｅフレームワークのＲｅｄｕｃｅ関数を実行するＲｅｄｕｃｅワーカサーバ、
を有し、
前記マスタサーバは、前記Ｍａｐワーカサーバと前記Ｒｅｄｕｃｅワーカサーバを前記複数のワーカサーバとして取り扱う
ことを特徴とするＭａｐＲｅｄｕｃｅシステム。
【請求項５】
前記Ｍａｐワーカサーバおよび前記Ｒｅｄｕｃｅワーカサーバは、
１以上の計算リソースを備えており、
前記マスタサーバから前記ベンチマークデータを受け取ると、前記ベンチマークを実施する過程において、
前記ベンチマークデータを分割して前記計算リソース毎にその分割したベンチマークデータを割り当てることができる場合は、前記ベンチマークデータを前記計算リソース毎に分割して割り当てる際の配分割合を、前記ベンチマークデータに対する処理時間が最小となるように求め、そのときの前記ベンチマークデータに対する処理時間を前記ベンチマークの結果として出力し、
前記ベンチマークデータを分割して前記計算リソース毎にその分割したベンチマークデータを割り当てることができない場合は、前記計算リソース毎に前記ベンチマーク全体を処理させて処理時間が最小である前記計算リソースを求め、その計算リソースの識別子とそのときの前記ベンチマークデータに対する処理時間を前記ベンチマークの結果として出力する
ことを特徴とする請求項４記載のＭａｐＲｅｄｕｃｅシステム。
【請求項６】
前記Ｍａｐワーカサーバおよび前記Ｒｅｄｕｃｅワーカサーバは、
前記マスタサーバから前記データを受け取ると、前記データを分割して前記計算リソース毎にその分割したデータを割り当てることができるか否かを、前記ベンチマークの結果に基づき判断し、
前記データを分割して前記計算リソース毎にその分割したデータを割り当てることができる場合は、前記データを前記計算リソース毎に分割して割り当てる際の配分割合を前記ベンチマークの結果に基づき決定し、その配分割合にしたがって前記計算リソースに対して前記データを割り当てて処理し、
前記データを分割して前記計算リソース毎にその分割したデータを割り当てることができない場合は、前記ベンチマークにおいて処理時間が最小であった前記計算リソースに対して前記データを割り当てて処理する
ことを特徴とする請求項５記載のＭａｐＲｅｄｕｃｅシステム。
【請求項７】
前記Ｍａｐワーカサーバおよび前記Ｒｅｄｕｃｅワーカサーバは、
前記マスタサーバから前記データを受け取ったとき、自身が備えている前記計算リソースのなかに計算を実施することができないものが存在する場合は、その旨を前記マスタサーバに対して通知するとともに、計算を実施することができる前記計算リソースを用いて前記データを処理し、
前記マスタサーバは、
計算を実施することができない前記計算リソースが存在する旨の通知を前記Ｍａｐワーカサーバまたは前記Ｒｅｄｕｃｅワーカサーバから受け取ると、前記ベンチマーク実行指示ステップおよび前記分割割合算出ステップを再実行することにより、計算を実施することができない前記計算リソースを除いた前記データの配分割合を改めて求める
ことを特徴とする請求項４記載のＭａｐＲｅｄｕｃｅシステム。
【請求項８】
ＭａｐＲｅｄｕｃｅフレームワークを使用してデータを分散処理する複数のワーカサーバに対して前記データを分割して割り当てるデータ割当制御装置であって、
前記複数のワーカサーバのうち少なくともいずれかの計算リソースは、他の前記ワーカサーバが備える計算リソースとは処理性能が異なっており、
前記データ割当制御装置は、
前記ワーカサーバの処理性能を測定するベンチマークのために用いるベンチマークデータを前記ワーカサーバに対して配布するベンチマークデータ配布部、
前記ベンチマークデータを用いて前記ベンチマークを実施するよう要求するリクエストを前記ワーカサーバに対して発行しそのベンチマークの結果を受信するベンチマーク実行指示部、
各前記ワーカサーバに対して前記データを分割して割り当てる際の前記データの配分割合を求める分割割合算出部、
前記分割割合算出部の処理結果に基づき前記データを分割して各前記ワーカサーバに対して割り当てるデータ割当部、
を備え、
前記分割割合算出部は、
前記ベンチマークの結果に基づき、前記割り当てられた前記データに対する各前記ワーカサーバの処理性能の合計が最大になる前記配分割合を求める
ことを特徴とするデータ割当制御装置。
【請求項９】
ＭａｐＲｅｄｕｃｅフレームワークを使用してデータを分散処理する複数のワーカサーバに対して前記データを分割して割り当てるデータ割当制御方法であって、
前記複数のワーカサーバのうち少なくともいずれかの計算リソースは、他の前記ワーカサーバが備える計算リソースとは処理性能が異なっており、
前記データ割当制御方法は、
前記ワーカサーバの処理性能を測定するベンチマークのために用いるベンチマークデータを前記ワーカサーバに対して配布するベンチマークデータ配布ステップ、
前記ベンチマークデータを用いて前記ベンチマークを実施するよう要求するリクエストを前記ワーカサーバに対して発行しそのベンチマークの結果を受信するベンチマーク実行指示ステップ、
各前記ワーカサーバに対して前記データを分割して割り当てる際の前記データの配分割合を求める分割割合算出ステップ、
前記分割割合算出ステップの結果に基づき前記データを分割して各前記ワーカサーバに対して割り当てるデータ割当ステップ、
を有し、
前記分割割合算出ステップにおいては、
前記ベンチマークの結果に基づき、前記割り当てられた前記データに対する各前記ワーカサーバの処理性能の合計が最大になる前記配分割合を求める
ことを特徴とするデータ割当制御方法。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ＭａｐＲｅｄｕｃｅフレームワークを使用してデータを分散処理する複数のワーカサーバに対してデータを最適に分割して割り当てる技術に関する。
【背景技術】
【０００２】
増大する大量のデジタルデータを処理するために、多数のノードでデータを分散させて高速に処理する技術が求められている。分散処理を実装する手法として、下記非特許文献１に示すＧｏｏｇｌｅ社が開発したＭａｐＲｅｄｕｃｅフレームワークがよく利用されている。
【０００３】
ＭａｐＲｅｄｕｃｅは、データ処理をＭａｐタスクとＲｅｄｕｃｅタスクの２段階に分けて処理するのが特徴である。Ｍａｐタスクは大量データから必要な情報を抽出し、Ｒｅｄｕｃｅタスクは抽出した情報を集約する。Ｍａｐタスクを実施するノードをＭａｐワーカノードなどと呼び、Ｒｅｄｕｃｅタスクを実施するノードをＲｅｄｕｃｅワーカノードなどと呼ぶ。ＭａｐＲｅｄｕｃｅフレームワークにより、開発者は開発作業をＭａｐタスクとＲｅｄｕｃｅタスクのみに対して集中することができ、容易に分散環境を構築・利用することができる。
【０００４】
従来のＭａｐＲｅｄｕｃｅ技術は、同等の処理性能を有するノードを用いて分散処理を実施することを前提としている。下記非特許文献１には、各サーバをＭａｐＲｅｄｕｃｅのワーカノードとしてＭａｐＲｅｄｕｃｅシステムを構築する方法が記述されている。下記非特許文献２には、マルチコアＣＰＵ（Ｃｅｎｔｒａｌ  Ｐｒｏｃｅｓｓｉｎｇ  Ｕｎｉｔ）の各コアをワーカノードとしてＭａｐＲｅｄｕｃｅシステムを構築する方法が記述されている。下記非特許文献３には、メニーコアを有するＧＰＧＰＵ（Ｇｅｎｅｒａｌ  Ｐｕｒｐｏｓｅ  Ｃｏｍｐｕｔｉｎｇ  ｏｎ  Ｇｒａｐｈｉｃｓ  Ｐｒｏｃｅｓｓｉｎｇ  Ｕｎｉｔｓ）を備えたコンピュータにおいて、各ＧＰＧＰＵをワーカノードとしてＭａｐＲｅｄｕｃｅシステムを構築する方法が記述されている。
【先行技術文献】
【非特許文献】
【０００５】
【非特許文献１】Jeffrey Dean and Sanjay Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters", Proceedings of the 6th conference on Symposium on Operating Systems Design & Implementation, 2004
【非特許文献２】Colby Ranger, Ramanan Raghuraman, Arun Penmetsa, Gary Bradski, and Christos Kozyrakis, "Evaluating MapReduce for Multi-core and Multiprocessor Systems", In Proceedings of the 13th Intl. Symposium on High-Performance Computer Architecture, 2007
【非特許文献３】Bingsheng He, Wenbin Fang, Qiong Luo, Naga K. Govindaraju and Tuyong Wang, "Mars: a MapReduce framework on graphics processors", In Proceedings of the 17th international conference on Parallel architectures and compilation techniques, 2008
【発明の概要】
【発明が解決しようとする課題】
【０００６】
大量データをより高速に処理するためには、サーバ、マルチコアＣＰＵ、ＧＰＧＰＵなどの計算リソースを組み合わせて計算させることが必要である。しかしながら、複数種類の計算リソースを組み合わせてＭａｐＲｅｄｕｃｅフレームワークによる計算を効率的に実施するためには、以下の課題があると考えられる。
【０００７】
従来のＭａｐＲｅｄｕｃｅシステムにおいては、ＭａｐタスクやＲｅｄｕｃｅタスクを処理する各ノードは、計算リソースの処理能力が同等であることを想定していると考えられる。しかし実際のコンピュータ環境においては、各ノードを構成するＣＰＵのコア数および各ＣＰＵの性能、ＧＰＧＰＵの個数および各ＧＰＧＰＵの性能がそれぞれ異なる場合がある。このとき、従来のＭａｐＲｅｄｕｃｅフレームワークのように、ＭａｐタスクあるいはＲｅｄｕｃｅタスクを実行する各ワーカサーバに対して同じサイズのデータを割り当てると、これら計算リソースの処理能力が異なる各ワーカサーバの性能を効率的に発揮することが困難になってしまう。
【０００８】
本発明は、上記のような課題に鑑みてなされたものであり、ＭａｐＲｅｄｕｃｅフレームワークを使用するシステムにおいて、各ワーカサーバに対して割り当てるデータサイズを適切に定めてシステム全体の処理性能を向上させることを目的とする。
【課題を解決するための手段】
【０００９】
本発明に係るデータ割当制御プログラムは、各ワーカサーバに対してデータを割り当てる前にあらかじめ各ワーカサーバのベンチマーク結果を取得し、その結果に基づき各ワーカサーバに対して割り当てるデータサイズを調整する。
【発明の効果】
【００１０】
本発明に係るデータ割当制御プログラムによれば、処理性能が異なる計算リソースを備えたワーカサーバによって構成されるＭａｐＲｅｄｕｃｅシステムにおいて、各ワーカサーバに対して分割データを効率的に割り当て、ＭａｐＲｅｄｕｃｅシステムの処理性能を向上させることができる。
【図面の簡単な説明】
【００１１】
【図１】従来のＭａｐＲｅｄｕｃｅシステムの構成例を示す図である。
【図２】入力ＤＢ１０３の構成とデータ例を示す図である。
【図３】中間ＤＢ１０５の構成とデータ例を示す図である。
【図４】出力ＤＢ１０７の構成とデータ例を示す図である。
【図５】Ｍａｐ関数１０４１とＲｅｄｕｃｅ関数１０６１の例を示す図である。
【図６】従来のＭａｐＲｅｄｕｃｅシステムの処理シーケンス図である。
【図７】本発明に係るＭａｐＲｅｄｕｃｅシステム１００のシステム構成例を示す図である。
【図８】ワーカ管理ＤＢ５００の構成とデータ例を示す図である。
【図９】マスタサーバ１０２の機能ブロック図である。
【図１０】ＭａｐＲｅｄｕｃｅシステム１００の動作を説明するシーケンス図である。
【図１１】ステップＳ１００３においてＭａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施するベンチマークテストのフローチャートである。
【図１２】ステップＳ１００６とＳ１００８においてＭａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施するＭａｐ処理およびＲｅｄｕｃｅ処理のフローチャートである。
【発明を実施するための形態】
【００１２】
以下では本発明の理解を容易にするため、まず従来のＭａｐＲｅｄｕｃｅシステムの構成について説明し、その後に本発明に係るＭａｐＲｅｄｕｃｅシステムの構成について説明する。
【００１３】
＜従来のＭａｐＲｅｄｕｃｅシステムの構成＞
図１は、従来のＭａｐＲｅｄｕｃｅシステムの構成例を示す図である。図１に示すＭａｐＲｅｄｕｃｅシステムは、クライアントコンピュータ１０１、マスタサーバ１０２、入力データベース（ＤＢ）１０３、１以上のＭａｐワーカサーバ１０４、中間ＤＢ１０５、１以上のＲｅｄｕｃｅワーカサーバ１０６、出力ＤＢ１０７を有する。
【００１４】
クライアントコンピュータ１０１は、ＭａｐＲｅｄｕｃｅシステムに対して計算開始を指示し、タスクの進捗状況を画面表示するための端末である。マスタサーバ１０２は、各ワーカサーバに対してＭａｐ処理およびＲｅｄｕｃｅ処理を実行するよう指示する。入力ＤＢ１０３は、Ｍａｐ処理に対する入力データを格納するデータベースである。Ｍａｐワーカサーバ１０４は、入力ＤＢ１０３から取得した入力データをＭａｐ処理するためのサーバであり、複数のＭａｐワーカサーバ１０４によってＭａｐ処理を分散実行する場合もある。中間ＤＢ１０５は、Ｍａｐ処理の結果を格納するデータベースである。Ｒｅｄｕｃｅワーカサーバ１０６は、中間ＤＢ１０５から取得したデータをＲｅｄｕｃｅ処理するためのサーバであり、複数のＲｅｄｕｃｅワーカサーバ１０６によってＲｅｄｕｃｅ処理を分散実行する場合もある。出力ＤＢ１０７は、Ｒｅｄｕｃｅ処理の結果を格納するデータベースである。
【００１５】
図２は、入力ＤＢ１０３の構成とデータ例を示す図である。ここではＭａｐＲｅｄｕｃｅシステムが文書データ内の単語の出現頻度をカウントする例について説明する。入力ＤＢ１０３は、ＭａｐＲｅｄｕｃｅシステムに対する入力データである各文書データを保持するデータベースであり、ＩＤフィールド１０３１、文書フィールド１０３２、サイズフィールド１０３３を有する。
【００１６】
ＩＤフィールド１０３１は、各文書データの識別番号を保持する。文書フィールド１０３２は、各文書データまたはそのファイルパスやファイス名を保持する。サイズフィールド１０３３は、各文書データのデータサイズを保持する。
【００１７】
図３は、中間ＤＢ１０５の構成とデータ例を示す図である。中間ＤＢ１０５は、Ｍａｐワーカサーバ１０４が入力ＤＢ１０３上の各入力データからｋｅｙとｖａｌｕｅを抽出した結果を格納するデータベースである。ここでは各文書データ内の単語をｋｅｙとし、各文書データ内におけるその出現頻度をｖａｌｕｅとして取り扱うこととした。
【００１８】
中間ＤＢ１０５は、単語フィールド１０５１、カウントフィールド１０５２を有する。単語フィールド１０５１は、Ｍａｐ処理によって各文書データから抽出された単語（本例においてはｋｅｙに相当）を保持する。カウントフィールド１０５２は、各文書データにおける各単語の出現頻度をカウントした結果（本例においてはｖａｌｕｅに相当）を保持する。
【００１９】
図４は、出力ＤＢ１０７の構成とデータ例を示す図である。出力ＤＢ１０７は、Ｒｅｄｕｃｅサーバ１０６が中間ＤＢ１０５上のデータを集約した結果を格納するデータベースであり、単語フィールド１０７１、出現頻度フィールド１０７２を有する。単語フィールド１０７１は、単語フィールド１０５１に対応する。出現頻度フィールド１０７２は、カウントフィールド１０５２を集約した結果を保持する。
【００２０】
図５は、Ｍａｐ関数１０４１とＲｅｄｕｃｅ関数１０６１の例を示す図である。以下、各関数について説明する。
【００２１】
Ｍａｐ関数１０４１は、入力ＤＢ１０３のＩＤフィールド１０３１に対応する文書ＩＤ（ｄｏｃｉｄ）と当該文書内の文字列（ｄｏｃ）を引数とする。Ｍａｐ関数１０４１は、当該文書内に含まれる単語（ｔｅｒｍ）を発見すると、単語（ｔｅｒｍ）とカウント（ｃｏｕｎｔ）１を中間ＤＢ１０５の各フィールドに格納する（Ｅｍｉｔ）。
【００２２】
Ｒｅｄｕｃｅ関数１０６１は、中間ＤＢ１０５の単語フィールド１０５１とカウントフィールド１０５２にそれぞれ対応する単語（ｔｅｒｍ）とカウント（ｃｏｕｎｔｓ）を引数とする。Ｒｅｄｕｃｅ関数１０６１は、カウントを集計してその出現頻度を求め、出力ＤＢ１０７に格納する（Ｅｍｉｔ）。
【００２３】
図６は、従来のＭａｐＲｅｄｕｃｅシステムの処理シーケンス図である。以下、図５の各ステップについて説明する。
【００２４】
（図６：ステップＳ２００）
ユーザは、クライアントコンピュータ１０１上において、ＭａｐＲｅｄｕｃｅ処理を開始するよう指示する。クライアントコンピュータ１０１は、その指示にしたがって、マスタサーバ１０２に対してＭａｐＲｅｄｕｃｅ処理を開始するよう指示するリクエストを発行する。
【００２５】
（図６：ステップＳ２０１）
マスタサーバ１０２は、クライアントコンピュータ１０１からＭａｐＲｅｄｕｃｅ処理を開始するよう指示するリクエストを受信すると、Ｍａｐワーカサーバ１０４およびＲｅｄｕｃｅワーカサーバ１０６に対して、それぞれＭａｐ処理とＲｅｄｕｃｅ処理を開始するよう指示する。
【００２６】
（図６：ステップＳ２０２）
Ｍａｐワーカサーバ１０４は、入力ＤＢ１０３からＩＤフィールド１０３１と文字列（文書フィールド１０３２の文書データ内に含まれる全文字列）を取得し、Ｍａｐ関数１０４１を実行してその結果を中間ＤＢ１０５に格納する。
【００２７】
（図６：ステップＳ２０３）
Ｒｅｄｕｃｅワーカサーバ１０６は、中間ＤＢ１０５から単語フィールド１０５１とカウントフィールド１０５２を取得し、Ｒｅｄｕｃｅ関数１０６１を実行してその結果を出力ＤＢ１０７に格納する。
【００２８】
＜本発明に係るＭａｐＲｅｄｕｃｅシステムの構成＞
図７は、本発明に係るＭａｐＲｅｄｕｃｅシステム１００のシステム構成例を示す図である。ＭａｐＲｅｄｕｃｅシステム１００の構成は、以下の点において従来のＭａｐＲｅｄｕｃｅシステムとは異なる。
【００２９】
（相違点１）ワーカサーバの計算リソース
ＭａｐＲｅｄｕｃｅシステム１００におけるＭａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６は、計算リソースの処理能力が必ずしも全てのワーカサーバについて同一ではない。各ワーカサーバは、マルチコアＣＰＵやＧＰＧＰＵを備えている。各マルチコアＣＰＵのコア数や性能、各ワーカサーバが備えるＧＰＧＰＵの個数、などによって定まる各ワーカサーバの計算リソースの処理能力は、少なくともいずれかのワーカサーバにおいてはその他のワーカサーバとは異なる。
【００３０】
（相違点２）ワーカ管理ＤＢ５００
マスタサーバ１０２は、各ワーカサーバの処理性能を管理するワーカ管理ＤＢ５００を備える。ワーカ管理ＤＢ５００は、各ワーカサーバが処理するデータのサイズとその処理時間との間の対応関係を記述するデータを保持するデータベースである。具体例については後述の図８で説明する。
【００３１】
図８は、ワーカ管理ＤＢ５００の構成とデータ例を示す図である。ワーカ管理ＤＢ５００は、ワーカＩＤフィールド５０１と処理時間フィールド５０２を有する。ワーカＩＤフィールド５０１は、各ワーカサーバを一意に識別するＩＤを保持する。処理時間フィールド５０２は、各ワーカサーバが処理するデータのサイズとそのサイズのデータを処理するために要する時間との間の対応関係を記述するデータを保持する。このデータは例えばデータサイズと処理時間との間の対応関係を表す関数として記述することができるが、これに限られるものではない。
【００３２】
図９は、マスタサーバ１０２の機能ブロック図である。マスタサーバ１０２は、ベンチマークデータ配布部１０２１、ベンチマーク実行指示部１０２２、分割割合算出部１０２３、データ割当部１０２４を備える。各機能部の動作については後述する。これら機能部は、その機能を実装した回路デバイスなどのハードウェアを用いて構成することもできるし、同様の機能を実装したソフトウェアをＣＰＵなどの演算装置が実行することによって構成することもできる。
【００３３】
図１０は、ＭａｐＲｅｄｕｃｅシステム１００の動作を説明するシーケンス図である。以下、図１０の各ステップについて説明する。
【００３４】
（図１０：ステップＳ１００１）
ユーザは、クライアントコンピュータ１０１上において、各ワーカサーバのベンチマークテストを実施するよう指示する。クライアントコンピュータ１０１は、その指示にしたがって、マスタサーバ１０２に対してベンチマークテストを実施するよう指示するリクエストを送信する。
【００３５】
（図１０：ステップＳ１００２）
マスタサーバ１０２は、クライアントコンピュータ１０１からベンチマークテストを実施するよう指示するリクエストを受信する。ベンチマークデータ配布部１０２１は、ベンチマークテストにおいて用いるベンチマークデータを各ワーカサーバに対して送信する。ベンチマーク実行指示部１０２２は、そのベンチマークデータを用いてベンチマークテストを実施するよう各ワーカサーバに対して指示する。このとき、Ｍａｐ関数１０４１とＲｅｄｕｃｅ関数１０６１を実装したベンチマークプログラムを併せて配布してそのベンチマークプログラムを用いてベンチマークテストを実施するよう指示してもよいし、ＭａｐＲｅｄｕｃｅフレームワーク上で実装された既存の処理プログラムを用いて同テストを実施してもよい。
【００３６】
（図１０：ステップＳ１００３）
Ｍａｐワーカサーバ１０４およびＲｅｄｕｃｅワーカサーバ１０６は、ベンチマークデータを受信すると、自身が備えるマルチコアＣＰＵやＧＰＧＰＵなどの計算リソースに対してベンチマークデータを分割して割り当て、処理が完了するまでに要する時間を測定する。Ｍａｐワーカサーバ１０４およびＲｅｄｕｃｅワーカサーバ１０６は、各計算リソースに対して割り当てる分割データの配分割合を変更しながら同様の測定を繰り返し、各計算リソースに対する分割データの最適な配分割合を求める。Ｍａｐワーカサーバ１０４およびＲｅｄｕｃｅワーカサーバ１０６は、各計算リソースに対する最適な配分割合を求めると、その配分割合を前提として図８で説明したデータサイズと処理時間の対応関係を表す関数を作成し、ワーカ管理ＤＢ５００の処理時間フィールド５０２に格納するとともに完了応答をベンチマーク実行指示部１０２２に対して送信する。本ステップの詳細については、後述の図１１で説明する。
【００３７】
（図１０：ステップＳ１００４）
本ステップは、ステップＳ２００と同様である。
【００３８】
（図１０：ステップＳ１００５）
分割割合算出部１０２３は、クライアントコンピュータ１０１からＭａｐＲｅｄｕｃｅ処理を開始するよう指示するリクエストを受信すると、入力ＤＢ１０３が保持している入力データを分割して各Ｍａｐワーカサーバ１０４に対して配分する際の分割割合を、ワーカ管理ＤＢ５００を参照して決定する。このとき、図８で説明したデータサイズと処理時間との対応関係に基づき、各Ｍａｐワーカサーバ１０４全体の処理性能が最大となるように、各Ｍａｐワーカサーバ１０４に対して割り当てる分割データを決定する。データ割当部１０２４は、各Ｍａｐワーカサーバ１０４が処理すべき分割データを各Ｍａｐワーカサーバ１０４に対して通知する。
【００３９】
（図１０：ステップＳ１００５：補足その１）
分割割合算出部１０２３は、各Ｍａｐワーカサーバ１０４が処理するデータの合計処理時間が最小になるように、各Ｍａｐワーカサーバ１０４に対して割り当てる分割データを決定する。例えば下記式１で表される処理時間Ｔが最小になるように、各分割データを決定する。Ｆｉ（ｘ）はワーカ管理ＤＢ５００が格納しているｉ番目のＭａｐワーカサーバ１０４がデータサイズｘを処理するのに要する時間を表す関数である。ｄｊは入力ＤＢ１０３が格納しているｊ番目のデータのサイズである。ｎｉはｉ番目のＭａｐワーカサーバ１０４が処理するデータの個数である。Ｎはワーカサーバの台数である。
【００４０】
【数１】
2015084152.tif 000003
【００４１】
（図１０：ステップＳ１００５：補足その２）
上記式１は、データ番号ｊが小さい順に、各ワーカサーバｉに対して分割データを順次割り当てることを前提としたものである。各ワーカサーバｉに対して割り当てるデータｊの順序は必ずしもデータ番号ｊの昇順に限られるものではなく、例えばワーカサーバｉとデータ番号ｊの全ての組み合わせを考慮して処理時間Ｔが最小になる組み合わせを求めてもよい。ただし全ての組み合わせに対して処理時間を求める処理は時間がかかるため、上記式１などのような簡易化した計算式を用いても、実用上十分であると考えられる。
【００４２】
（図１０：ステップＳ１００６）
データ割当部１０２４は、Ｍａｐワーカサーバ１０４に対して、Ｍａｐ関数１０４１を実行するよう指示する。Ｍａｐワーカサーバ１０４は、分割割合算出部１０２３によって指定されたデータを入力ＤＢ１０３から取得し、そのデータに対してＭａｐ関数１０４１を実行する。Ｍａｐワーカサーバ１０４は、その実行結果を中間ＤＢ１０５に格納するとともに、データ割当部１０２４に対して実行完了を通知する。本ステップの詳細は後述の図１２で説明する。
【００４３】
（図１０：ステップＳ１００６：補足その１）
ベンチマークデータ配布部１０２１は、ステップＳ１００３において、ＭａｐＲｅｄｕｃｅシステム１００が実際に処理するデータをサンプリングすることにより、ベンチマークデータを作成してもよい。この場合、本ステップにおけるＭａｐワーカサーバ１０４の処理時間は、ステップＳ１００３において得られた処理時間に準じたものとなる。
【００４４】
（図１０：ステップＳ１００６：補足その２）
以上の説明においては、記載の便宜上、分割割合算出部１０２３が各ワーカサーバに対する分割データの配分割合を求めて各ワーカサーバに対しその配分割合を通知することとしたが、データ割当部１０２４がＭａｐ関数１０４１を実行するようＭａｐワーカサーバ１０４に対して指示する際にその配分割合を指示してもよい。Ｒｅｄｕｃｅ処理についても同様である。
【００４５】
（図１０：ステップＳ１００７）
分割割合算出部１０２３は、全てのＭａｐワーカサーバ１０４がステップＳ１００６を完了したことを確認した後、ステップＳ１００５と同様に、各Ｒｅｄｕｃｅワーカサーバ１０６に対して割り当てる分割データを決定し、各Ｒｅｄｕｃｅワーカサーバ１０６に対して通知する。
【００４６】
（図１０：ステップＳ１００８）
データ割当部１０２４は、Ｒｅｄｕｃｅワーカサーバ１０６に対して、Ｒｅｄｕｃｅ関数１０６１を実行するよう指示する。Ｒｅｄｕｃｅワーカサーバ１０６は、分割割合算出部１０２３が指定したデータを中間ＤＢ１０５から取得してＲｅｄｕｃｅ処理を実施し、その結果を出力ＤＢ１０７に格納する。本ステップの詳細はステップＳ１００６の詳細と併せて後述の図１２で説明する。
【００４７】
図１１は、ステップＳ１００３においてＭａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施するベンチマークテストのフローチャートである。Ｍａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施する処理は、Ｍａｐ処理とＲｅｄｕｃｅ処理の違いを除いて同様であるため、以下では図１１においてＭａｐワーカサーバ１０４が実施する各ステップについて説明する。
【００４８】
（図１１：ステップＳ１１０１）
Ｍａｐワーカサーバ１０４は、ベンチマークデータ配布部１０２１からベンチマークデータを受信する。ステップＳ１００２で説明したように、併せてベンチマークプログラムを受信してもよい。
【００４９】
（図１１：ステップＳ１１０２）
Ｍａｐワーカサーバ１０４は、ベンチマークデータを処理するプログラム（ベンチマークプログラムを併せて受信した場合はそのプログラム）のデータフローグラフを作成し、これに基づき、Ｍａｐ関数１０４１の引数であるｄｏｃ（文字配列）とｃｏｕｎｔｓ（数字配列）に対するアクセスパターンを解析する。
【００５０】
（図１１：ステップＳ１１０３）
Ｍａｐワーカサーバ１０４は、ステップＳ１１０２の解析結果に基づき、Ｍａｐ関数１０４１の引数である配列を分割して各計算リソースに対して並列的に割り当てることができるか否か（すなわち並列計算が可能か否か）を判定する。分割可能である場合はステップＳ１１０５に進み、分割不可能である場合はステップＳ１１０４に進む。
【００５１】
（図１１：ステップＳ１１０４）
Ｍａｐワーカサーバ１０４は、各計算リソースに対して個別にベンチマークデータを割り当ててＭａｐ関数１０４１を実施し、各計算リソースがベンチマークデータを処理するために要する時間を求める。Ｍａｐワーカサーバ１０４は、各計算リソースの処理時間を内部的に記憶しておく。
【００５２】
（図１１：ステップＳ１１０５）
Ｍａｐワーカサーバ１０４は、Ｍａｐ関数１０４１の引数である配列を分割して各計算リソースに対して割り当て、Ｍａｐ関数１０４１を実施し、処理時間を測定する。Ｍａｐワーカサーバ１０４は、各計算リソースに対して割り当てるデータサイズの配分割合を偏向しながら同様の測定を繰り返し、処理時間が最も短い最適な配分割合を求める。Ｍａｐワーカサーバ１０４は、その最適配分割合を内部的に記憶しておく。
【００５３】
（図１１：ステップＳ１１０６）
Ｍａｐワーカサーバ１０４は、ステップＳ１１０４またはＳ１１０５において得られたデータサイズと処理時間との間の対応関係を、ワーカ管理ＤＢ５００に格納する。ステップＳ１１０４については、各計算リソースの処理時間をワーカ管理ＤＢ５００に併せて格納してもよい。ステップＳ１１０５については、各計算リソースに対するデータサイズの最適な配分割合をワーカ管理ＤＢ５００に併せて格納してもよい。
【００５４】
（図１１：ステップＳ１１０６：補足）
ワーカ管理ＤＢ５００は処理の便宜上設けたものであり、省略することもできる。すなわちベンチマーク実行指示部１０２２は、ワーカ管理ＤＢ５００を介してベンチマーク結果を受け取ることもできるし、各ワーカサーバからベンチマーク結果を直接受け取ることもできる。
【００５５】
図１２は、ステップＳ１００６とＳ１００８においてＭａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施するＭａｐ処理およびＲｅｄｕｃｅ処理のフローチャートである。Ｍａｐワーカサーバ１０４とＲｅｄｕｃｅワーカサーバ１０６がそれぞれ実施する処理は、Ｍａｐ処理とＲｅｄｕｃｅ処理の違いを除いて同様であるため、以下では図１２においてＭａｐワーカサーバ１０４が実施する各ステップについて説明する。
【００５６】
（図１２：ステップＳ１２０１）
Ｍａｐワーカサーバ１０４は、分割割合算出部１０２３が指定したデータを入力ＤＢ１０３から取得する。
【００５７】
（図１２：ステップＳ１２０２〜Ｓ１２０３）
Ｍａｐワーカサーバ１０４は、ハードウェア障害などの原因により利用できない状態になっている計算リソースが存在するか否かを確認する（Ｓ１２０２）。存在する場合はステップＳ１２０７へ進み、存在しない場合はステップＳ１２０４へ進む（Ｓ１２０３）。
【００５８】
（図１２：ステップＳ１２０４）
Ｍａｐワーカサーバ１０４は、入力ＤＢ１０３から取得した入力データを分割して各計算リソースに対して割り当てることができるか否かを判断する。分割可能である場合はステップＳ１２０６へ進み、分割不可能である場合はステップＳ１２０５へ進む。図１２において実施するベンチマークテストにおいて用いるベンチマークデータが、入力ＤＢ１０３内のデータをサンプリングすることによって生成されている場合は、ベンチマークデータと本ステップにおける入力データは同様の傾向を有していると考えられるので、ステップＳ１１０３の結果に準じてデータ分割可否を判断してもよい。あるいは、計算負荷の観点から許容できる場合は、本ステップにおいて改めてデータ分割可否を判断してもよい。
【００５９】
（図１２：ステップＳ１２０５）
Ｍａｐワーカサーバ１０４は、空いている（計算実施中でない）計算リソースのうち最も処理性能の高い計算リソースを、ステップＳ１１０４の結果に基づき特定し、その計算リソースに対して入力データを割り当ててＭａｐ処理を実施する。
【００６０】
（図１２：ステップＳ１２０６）
Ｍａｐワーカサーバ１０４は、ステップＳ１１０５の結果得られた配分割合に基づき、Ｍａｐ関数１０４１の引数（ｄｏｃｉｄ：文字配列、ｃｏｕｎｔｓ： 数字配列）を分割して各計算リソースに割り当て、Ｍａｐ関数１０４１を実行させてその結果を集計する。
【００６１】
（図１２：ステップＳ１２０７）
Ｍａｐワーカサーバ１０４は、利用できない計算リソースを除いたうえで図１１と同様のベンチマークを実施し、ワーカ管理ＤＢ５００を更新する。Ｍａｐワーカサーバ１０４はさらに、マスタサーバ１０２に対して利用できない計算リソースが存在している旨の障害通知を送信する。マスタサーバ１０２はこの障害通知を受信すると、ステップＳ１００５を再実施し、各ワーカサーバに対して割り当てる分割データをＭａｐワーカサーバ１０４に対して改めて通知する。Ｒｅｄｕｃｅサーバ１０６が本ステップを実施する場合は、同様にステップＳ１００７を再実施する。以後本フローチャートを実施する際には、更新後のベンチマーク結果および各ワーカサーバに対するデータ配分割合を用いる。
【００６２】
（図１２：ステップＳ１２０８）
Ｍａｐワーカサーバ１０４は、既にステップＳ１２０１において受信済のデータについては、利用可能な計算リソースを用いて計算を続行する。
【００６３】
（図１２：ステップＳ１２０９）
Ｍａｐワーカサーバ１０４は、計算結果を中間ＤＢ１０５に格納する。
【００６４】
以上、本発明の実施形態について説明した。本発明は上記した実施形態の形態に限定されるものではなく、様々な変形例が含まれる。上記実施形態は本発明を分かりやすく説明するために詳細に説明したものであり、必ずしも説明した全ての構成を備えるものに限定されるものではない。
【００６５】
上記各構成、機能、処理部、処理手段等は、それらの一部や全部を、例えば集積回路で設計する等によりハードウェアで実現してもよい。また、上記の各構成、機能等は、プロセッサがそれぞれの機能を実現するプログラムを解釈し、実行することによりソフトウェアで実現してもよい。各機能を実現するプログラム、テーブル、ファイル等の情報は、メモリ、ハードディスク、ＳＳＤ（Ｓｏｌｉｄ  Ｓｔａｔｅ  Ｄｒｉｖｅ）等の記録装置、ＩＣカード、ＳＤカード、ＤＶＤ等の記録媒体に格納することができる。
【００６６】
＜本発明のまとめ＞
以上のように、本発明に係るＭａｐＲｅｄｕｃｅシステムは、あらかじめベンチマークテストを実施することにより、各ワーカサーバに対して割り当てる分割データの最適な配分割合をあらかじめ求めておき、その配分割合にしたがって各ワーカサーバに対してデータを配分する。これにより、各ワーカサーバの計算リソースの処理性能が異なる場合であっても、ＭａｐＲｅｄｕｃｅシステムの処理性能を効果的に引き出すことができる。
【００６７】
また、本発明に係るＭａｐＲｅｄｕｃｅシステムにおいて、各ワーカサーバは、ベンチマークテストの過程において、自身が備える各計算リソースに対して割り当てる分割データの配分割合を最適化する。これにより、各ワーカサーバに対して配分する分割データの割合を最適化した上で、各ワーカサーバ内部においても各計算リソースの処理性能を効果的に発揮させることができる。
【符号の説明】
【００６８】
１００：ＭａｐＲｅｄｕｃｅシステム、１０１：クライアントコンピュータ、１０２：マスタサーバ、１０３：入力ＤＢ、１０４：Ｍａｐワーカサーバ、１０５：中間ＤＢ、１０６：Ｒｅｄｕｃｅワーカサーバ、１０７：出力ＤＢ。
【図１】
2015084152.tif 000004
【図２】
2015084152.tif 000005
【図３】
2015084152.tif 000006
【図４】
2015084152.tif 000007
【図５】
2015084152.tif 000008
【図６】
2015084152.tif 000009
【図７】
2015084152.tif 000010
【図８】
2015084152.tif 000011
【図９】
2015084152.tif 000012
【図１０】
2015084152.tif 000013
【図１１】
2015084152.tif 000014
【図１２】
2015084152.tif 000015