(57)【要約】      （修正有）【課題】MapReduceにおいてジョブ統合を行った場合でもジョブの同時実行を可能とし、サーバリソースの稼働率の低下を抑止する。【解決手段】MapReduceジョブを構成する複数のMapperまたはReducerが同一の入力データを処理実行するMapReduceジョブ実行システムで、MapReduceジョブで用いられる入出力データとMapReduceジョブ同士の関係との定義ファイルに基づき、第一のMapReduceジョブの出力データを入力データとして処理する第二のMapReduceジョブの存在を判定し、存在すると判定された場合、同一の入力データに基づいて出力データを生成する第一のReducerと、同一の入力データに基づいて異なる出力データを生成する第二のReducerとを再定義し、第一のReducerが第二のReducerよりも先に完了するように、Reduceタスクの実行順序を制御する。【選択図】図１７
2015095096.tif 000002
【特許請求の範囲】
【請求項１】
MapReduceジョブを構成する複数のMapperもしくはReducerが同一の入力データを処理して複数のMapReduceジョブを実行するMapReduceジョブ実行システムであって、
前記複数のMapReduceジョブで用いられる入出力データとMapReduceジョブ同士の関係とが記述された定義ファイルに基づいて、前記複数のMapReduceジョブのうち、第一のMapReduceジョブの出力データを入力データとして処理する第二のMapReduceジョブが存在するか否かを判定する前処理部と、
前記第二のMapReduceジョブが存在すると判定された場合に、前記同一の入力データに基づいて前記出力データを生成する第一のReducerと、前記同一の入力データに基づいて前記出力データと異なるデータを生成する第二のReducerとを異なるReduceタスクとして再定義し、前記第一のReducerが前記第二のReducerよりも先に完了するように、前記Reduceタスクの実行順序を制御して実行するReduce処理部と、
を備えることを特徴とするMapReduceジョブ実行システム。
【請求項２】
前記MapReduceジョブ実行システムは、
前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記出力データを出力する第一のMap処理部と、前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記異なるデータを出力する第二のMap処理部と、を有する第一のMapタスク処理部を備え、
前記Reduce処理部は、前記第一のReducerが第一のMap処理部が出力した前記出力データを読み込み、前記第二のReducerが第二のMap処理部が出力した前記異なるデータを読み込んで、前記第一のReducerと前記第二のReducerとを異なるタスクとして実行する、
ことを特徴とする請求項１に記載のMapReduceジョブ実行システム。
【請求項３】
前記MapReduceジョブ実行システムは、
前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記第一のReducerと前記第二のReducerとに共通の出力データを出力する第三のMap処理部、を有する第二のMapタスク処理部を備え、
前記Reduce処理部は、前記第一のReducerと前記第二のReducerとが前記共通の出力データを読み込んで、前記第一のReducerと前記第二のReducerの一部とを異なるタスクとして実行する、
ことを特徴とする請求項１に記載のMapReduceジョブ実行システム。
【請求項４】
前記Reduce処理部は、前記第一のReducerが実行するReduceタスクに対応する第一のパーティション番号を、前記第二のReducerが実行するReduceタスクに対応する第二のパーティション番号よりも小さい値に設定することにより、前記Reduceタスクの実行順序を制御する、
ことを特徴とする請求項１〜３のいずれか１項に記載のMapReduceジョブ実行システム。
【請求項５】
前記Reduce処理部は、前記第二のReducerが実行するReduceタスクのパーティション番号のうちのあらかじめ定められた最後の部分のパーティション番号のReduceタスクであるキャッシュタスクと、前記第二のReducerが実行するReduceタスクのうちの前記キャッシュタスクと同じ数のReduceタスクであって、前記Reduceタスクのパーティション番号より後の番号のReduceタスクである拡張タスクとを定め、前記拡張タスクを、前記第二のReducerの一部の異なるタスクとして実行する、
ことを特徴とする請求項３に記載のMapReduceジョブ実行システム。
【請求項６】
前記MapReduceジョブ実行システムは、
前記MapReduceジョブごとに、MapタスクおよびReduceタスクの実行状況を監視し、その監視結果を表示部に表示するスケジュール部をさらに備える、
ことを特徴とする請求項１〜５のいずれか１項に記載のMapReduceジョブ実行システム。
【請求項７】
MapReduceジョブを構成する複数のMapperもしくはReducerが同一の入力データを処理して複数のMapReduceジョブを実行するMapReduceジョブ実行方法であって、
前記複数のMapReduceジョブで用いられる入出力データとMapReduceジョブ同士の関係とが記述された定義ファイルに基づいて、前記複数のMapReduceジョブのうち、第一のMapReduceジョブの出力データを入力データとして処理する第二のMapReduceジョブが存在するか否かを判定する判定ステップと、
前記第二のMapReduceジョブが存在すると判定された場合に、前記同一の入力データに基づいて前記出力データを生成する第一のReducerと、前記同一の入力データに基づいて前記出力データと異なるデータを生成する第二のReducerとを異なるReduceタスクとして再定義する再定義ステップと、
前記第一のReducerが前記第二のReducerよりも先に完了するように、前記Reduceタスクの実行順序を制御して実行する実行ステップと、
を備えることを特徴とするMapReduceジョブ実行方法。
【請求項８】
前記MapReduceジョブ実行方法において、
前記判定ステップの後、前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記出力データを出力する第一のMap処理ステップと、前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記異なるデータを出力する第二のMap処理ステップと、を含み、
前記実行ステップでは、前記第一のReducerが第一のMap処理部が出力した前記出力データを読み込み、前記第二のReducerが第二のMap処理部が出力した前記異なるデータを読み込んで、前記第一のReducerと前記第二のReducerとを異なるタスクとして実行する、
ことを特徴とする請求項７に記載のMapReduceジョブ実行方法。
【請求項９】
前記MapReduceジョブ実行方法において、
前記判定ステップの後、前記定義ファイルに基づいて、前記同一の入力データを読み込んでMapタスクを実行して前記第一のReducerと前記第二のReducerとに共通の出力データを出力する第三のMap処理ステップを含み、
前記実行ステップでは、前記第一のReducerと前記第二のReducerとが前記共通の出力データを読み込んで、前記第一のReducerと前記第二のReducerの一部とを異なるタスクとして実行する、
ことを特徴とする請求項７に記載のMapReduceジョブ実行方法。
【請求項１０】
前記実行ステップでは、前記第一のReducerが実行するReduceタスクに対応する第一のパーティション番号を、前記第二のReducerが実行するReduceタスクに対応する第二のパーティション番号よりも小さい値に設定することにより、前記Reduceタスクの実行順序を制御する、
ことを特徴とする請求項７〜９のいずれか１項に記載のMapReduceジョブ実行方法。
【請求項１１】
前記実行ステップでは、前記第二のReducerが実行するReduceタスクのパーティション番号のうちのあらかじめ定められた最後の部分のパーティション番号のReduceタスクであるキャッシュタスクと、前記第二のReducerが実行するReduceタスクのうちの前記キャッシュタスクと同じ数のReduceタスクであって、前記Reduceタスクのパーティション番号より後の番号のReduceタスクである拡張タスクとを定め、前記拡張タスクを、前記第二のReducerの一部の異なるタスクとして実行する、
ことを特徴とする請求項９に記載のMapReduceジョブ実行方法。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、MapReduceジョブ実行システム、MapReduceジョブ実行方法に関する。
【背景技術】
【０００２】
近年、大規模データを処理するための並列分散型フレームワークとしてMapReduce（非特許文献１）が提案されている。MapReduceは、MapとReduceという2つの並列処理から構成される処理モデルである。Mapは、互いに依存関係のない入力データを並列処理し、処理結果をkey-value形式のデータとして出力する。出力結果は、key毎にグルーピングされ、Reduceへの入力になる。Reduceは、このグループ毎に並列処理し結果を出力する。原理的には、MapやReduceの入力データの単位（レコード）毎に並列化可能であるが、細かく分割しすぎると処理のオーバーヘッドが大きくなるため、通常、複数のレコードを束ねたタスク毎に並列処理する。
【０００３】
MapReduceの高速化手法として複数のMapやReduceが同一データを入力とするとき、これら複数のMapやReduceを1つのMapReduceジョブに纏めて実行する手法（非特許文献２）が提案されている。本明細では、この手法をジョブ統合と呼ぶ。また、一般にジョブの高速化手法として互いに依存関係がないジョブを同時に実行することで、サーバリソースの稼働率を向上する方法が知られており、この手法はMapReduceジョブに対しても適用できる。
【先行技術文献】
【非特許文献】
【０００４】
【非特許文献１】MapReduce: Simplified Data Processing on Large Clusters, OSDI’04
【非特許文献２】MRShare: Sharing Across Multiple Queries in MapReduce [Tomasz Nykiel, Michalis Potamias, Chaitanya Mishra, George Kollios, Nick Koudas, VLDB2010
【発明の概要】
【発明が解決しようとする課題】
【０００５】
非特許文献２の手法は、I/O量を削減することができるが、ジョブ統合によって同時実行可能なジョブ数が少なくなるため、依存関係がないジョブを同時実行する手法が使えなくなることがある、という課題がある。
【０００６】
例えば、図９には、３つのMapReduceジョブから構成されるジョブネット（複数のジョブの依存関係を非循環有向グラフで示したもの）を示している。点線は、ジョブ統合を行った後のジョブ構成を示す。Job1は、元々２つのMapReduceジョブであった部分を１つのジョブに統合したものである。ジョブ統合を行わない場合、Mapper1とReducer1から構成されるジョブ（ジョブA）と、Mapper2とReducer2から構成されるジョブ(ジョブB)とを同時実行できる。つまり、最初にジョブAを実行し、ジョブAの終盤で多重度が低下してきたらジョブBを開始し、ジョブBの終盤で多重度が低下してきたらジョブ2を開始する、という具合に、複数のジョブを重ね合わせながらジョブネットを実行することができる。
【０００７】
しかし、ジョブAとジョブBをジョブ1に統合すると、ジョブ1が終了するまでジョブ2を開始することができないため、ジョブ1の終盤で多重度が低下しても同時実行可能なジョブがなくなり、サーバリソースの稼働率が低下するという課題が発生する。
【０００８】
本発明は、上記に鑑みてなされたものであって、MapReduceにおいてジョブ統合を行った場合であってもジョブの同時実行を可能とし、サーバリソースの稼働率の低下を抑止することが可能なMapReduceジョブ実行システム、MapReduceジョブ実行方法を提供することを目的とする。
【課題を解決するための手段】
【０００９】
上述した課題を解決し、目的を達成するために、本発明にかかるMapReduceジョブ実行システムは、MapReduceジョブを構成する複数のMapperもしくはReducerが同一の入力データを処理して複数のMapReduceジョブを実行するMapReduceジョブ実行システムであって、前記複数のMapReduceジョブで用いられる入出力データとMapReduceジョブ同士の関係とが記述された定義ファイルに基づいて、前記複数のMapReduceジョブのうち、第一のMapReduceジョブの出力データを入力データとして処理する第二のMapReduceジョブが存在するか否かを判定する前処理部と、前記第二のMapReduceジョブが存在すると判定された場合に、前記同一の入力データに基づいて前記出力データを生成する第一のReducerと、前記同一の入力データに基づいて前記出力データと異なるデータを生成する第二のReducerとを異なるReduceタスクとして再定義し、前記第一のReducerが前記第二のReducerよりも先に完了するように、前記Reduceタスクの実行順序を制御して実行するReduce処理部と、を備えることを特徴とするMapReduceジョブ実行システムとして構成される。
【００１０】
また、本発明は、上記MapReduceジョブ実行システムで行われるMapReduceジョブ実行方法としても把握される。
【発明の効果】
【００１１】
本発明によれば、MapReduceにおいてジョブ統合を行った場合であってもジョブの同時実行を可能とし、サーバリソースの稼働率の低下を抑止することが可能となる。
【図面の簡単な説明】
【００１２】
【図１】システム構成の一例を示す図である。
【図２】マスタノードのソフトウェア構成を示す図である。
【図３】計算機のハードウェア構成を示す図である。
【図４】ジョブ管理テーブルの一例を示す図である。
【図５】タスク管理テーブルの一例を示す図である。
【図６】ジョブネット定義テーブルの一例を示す図である。
【図７】ジョブネット状態テーブルの一例を示す図である。
【図８】Childプロセスのソフトウェア構成を示す図である。
【図９】ジョブネットの一例を示す図である。
【図１０】ジョブネットの別の一例を示す図である。
【図１１】図９に示したジョブネットに対応するジョブ定義ファイルを示す図である。
【図１２】図１０に示したジョブネットに対応するジョブ定義ファイルを示す図である。
【図１３】JobConfiguratorの処理概要を示す図である。
【図１４】図１１に示したジョブ定義ファイルの更新内容を示す図である。
【図１５】図１２に示したジョブ定義ファイルの更新内容を示す図である。
【図１６】Mapタスクの処理概要を示す図である。
【図１７】Reduceタスクの処理概要を示す図である。
【図１８】ジョブ監視システムの画面例を示す図である。
【発明を実施するための形態】
【００１３】
以下に添付図面を参照して、本発明にかかるMapReduceジョブ実行方法、MapReduceジョブ実行装置の実施の形態を詳細に説明する。
【００１４】
図１は、本発明をMapReduceのOSS(Open Source Software)の１つであるHadoopに適用した場合のシステム構成図である。本システムは、１台のマスタノード（１０）、４台のスレーブノード（２０）、1台のユーザ端末（３０）から構成される（ノードとはサーバと同義である）。マスタノード（１０）およびスレーブノード（２０）は、並列分散処理を実現するサーバ群であり、ユーザ端末（３０）は、ユーザがジョブを投入するための端末である。尚、本実施例では、スレーブノードを４台としているが、本発明を適用する上で、スレーブノードの台数は1台以上の任意の台数であってもよい。
【００１５】
マスタノード（１０）では、ジョブネットの設定情報（例えば、ジョブ定義ファイルに定義されているMapタグ、Reduceタグ等に記載されている情報）に基づきジョブのパラメタを設定するJobConfigurator（１０１）、ジョブネット全体の実行を制御するJobScheduler（１０２）、MapReduceジョブの実行を制御するJobTracker（１０３）、分散ファイルシステムにおけるメタデータを管理するNameNode（１０４）が稼働する。スレーブノード（２０）では、Mapタスク・Reduceタスクの実行を管理するTaskTacker(２０１)、分散ファイルシステムにおけるデータアクセス処理を実行するDataNode（２０２）、タスクの実体であるChild（２０３）が稼働する。
【００１６】
Child（２０３）は、MapReduceジョブ実行中のみ動作するプロセスであり、TaskTracker（２０１）から起動される。JobTracker（１０３）、NameNode(１０４)、TaskTracker（２０１）、DataNode(２０２)、Child（２０３）は、相互に連携してMapReduceジョブの実行機能を提供する。JobConfigurator（１０１）は、MapReduceジョブ実行前に前処理として各ジョブのパラメタを設定する。JobScheduler（１０２）は、JobTracker（１０３）に対してMapReduceジョブの投入を指示する。
【００１７】
図２に、マスタノード（１０）内のソフトウェア構成の詳細を示す。尚、図２では、本発明とは関係の低いNameNode（１０４）は省略している。JobConfigurator（１０１）は、ジョブネットの構成を定義するジョブネット定義テーブル（Ｔ３）と、各ジョブの詳細を記述したジョブ定義ファイル群（Ｆ１、Ｆ２）とを保持する。JobScheduler（１０２）は、ジョブネットの実行状態を管理するジョブネット状態テーブル（Ｔ４）を保持する。
【００１８】
JobTracker（１０３）は、MapReduceジョブの進捗を管理するジョブ管理テーブル（Ｔ１）と、タスクの実行状態を管理するタスク管理テーブル（Ｔ２）とを保持する。JobTracker（１０３）は、実行中のMapReduceジョブを構成するタスク群をスレーブノード（２０）に割り当てるが、TaskScheduler（１０５）はこの割り当て処理を行うモジュールである。
【００１９】
本システムを構成するノード及びユーザ端末のハードウェア構成を、図３に示す。いずれも一般的な計算機やコンピュータである。各ノード及びユーザ端末の内部では、ＣＰＵ（３０１）、ＬＡＮＩ／Ｆ（３０２）、メモリ（３０３）、入出力Ｉ／Ｆ（３０４）、ＨＤＤＩ／Ｆ（３０５）が相互に接続されている。ＬＡＮＩ／Ｆ（３０２）は、ネットワークに接続され、ノード（１０、２０）同士やユーザ端末（３０）とノードの間で通信するときに利用する。ＨＤＤＩ／Ｆ（３０５）には、ＨＤＤ（３０９）が接続されている。各種プログラムやデータは、ＨＤＤ（３０９）に保存されており、ＨＤＤＩ／Ｆ（３０５）を経由して読み出され、メモリ（３０３）上に展開される。
【００２０】
また、展開されたプログラムは、ＣＰＵ（３０１）によって実行される。入出力Ｉ／Ｆ（３０４）には、ディスプレイ（３０６）、キーボード（３０７）、マウス（３０８）が接続されており、ユーザに対するインタフェースを提供する。尚、マスタノード（１０）、スレーブノード（２０）には、入出力Ｉ／Ｆ（３０４）に接続される入出力デバイスは必ずしも必要ではない。
【００２１】
図４は、JobTracker（１０３）が管理するジョブ管理テーブル（Ｔ１）の詳細である。ジョブ管理テーブル（Ｔ１）は、MapReduceジョブ毎に、ジョブを識別するためのＩＤである「JobID」、Mapタスクの総数である「Map数」、Map処理の進捗状況を表す「Map進捗」、Reduceタスクの総数である「Reduce数」、Reduce処理の進捗状況を表す「Reduce進捗」を対応付けて保持する。JobTracker（１０３）は、MapやReduceの進捗を、「実行済みタスク数／実行中タスク数／未実行タスク数」という形式で記載する。図４では、例えば、JOB1のMap数およびReduce数は、それぞれ800および40であり、Mapの実行済みタスク数が20、実行中タスク数が32、未実行タスク数が748であることを示している。また、40あるReduceのタスクの全てが実行済みであることを示している。
【００２２】
図５は、JobTracker（１０３）が管理するタスク管理テーブル（Ｔ２）である。タスク管理テーブル（Ｔ２）は、タスクを識別するためのIDである「TaskID」、タスクの実行状態を管理する「状態フラグ」、タスクが拡張タスクであるか否かを示す「拡張フラグ」を対応付けて保持する。「TaskID」は、「ジョブID_タスク種別_シーケンス番号」という形式で記載する。タスク種別によって、Mapタスク（mと記載）とReduceタスク（rと記載）とを区別する。例えば、ジョブIDがjob1、シーケンス番号が0001のMapタスクの場合、タスクIDは、「job1_m_0001」となる。「状態フラグ」には、未実行、実行済み、実行中の３種類がある。
【００２３】
「拡張フラグ」には、そのタスクが拡張タスクであれば「1」が記載され、拡張タスクでなければ「0」と記載される。拡張タスクについての詳細は後述するが、拡張タスクは特定のスレーブノードのみで実行可能となる。そして、タスクをスレーブノードに割り当てていくとき、TaskTracker（２０１）は、自身が実行可能な拡張タスクのタスクＩＤやシーケンス番号の一覧を保持しており、この情報に基づき、TaskScheduler（１０５）は、拡張タスクを実行可能なスレーブノードに割り当てる。尚、通常のタスクは、どのスレーブノードに割り当てても実行可能である。
【００２４】
JobTracker（１０３）は、MapReduceジョブが開始されると、上述したジョブネットの設定情報（ジョブ定義ファイル）のMapタグ、ReduceタグをカウントしてMapタスク数、Reduceタスク数を取得し、ジョブ管理テーブル（Ｔ１）に該当するジョブのエントリを追加する。この段階では、ジョブ管理テーブル（Ｔ１）には、Map数、Reduce数が記録されるとともに、Map、Reduceが進捗なしの状態として記録される。また、JobTracker（１０３）は、タスク管理テーブル（Ｔ２）にジョブを構成する全タスクに関するエントリを追加する。また、JobTracker（１０３）は、ジョブ開始時に拡張タスクのシーケンス番号の一覧を受け取り、これに基づきタスク管理テーブル（Ｔ２）の「拡張フラグ」を設定する。
【００２５】
ジョブが開始されると、TaskScheduler（１０５）は、各タスクをスレーブノードに割り当てていく。割り当てたタスクは、TaskScheduler（１０５）によって、タスク管理テーブル（Ｔ２）の状態フラグが「実行中」に更新され、スレーブノード（２０）上で実行が完了したら、状態フラグは「実行済」に更新される。ジョブを構成するすべてのタスクの実行が完了したら、ジョブが完了したことになるため、JobTracker（１０３）は、ジョブ管理テーブル（Ｔ１）及びタスク管理テーブル（Ｔ２）から、そのジョブに関するすべてのエントリを削除する。尚、MapReduceはすべてのMapタスクが完了しないと、Reduceタスクを開始することができない処理モデルであるため、TaskScheduler（１０５）は、すべてのMapタスクが完了するまでReduceタスクの割り当てを行わない。
【００２６】
図６に、ジョブネット定義テーブル（Ｔ３）の詳細を示す。本テーブルは、ジョブを識別するためのＩＤである「JobID」、そのジョブの出力データを入力データとして利用するジョブのジョブＩＤである「NextJob」、そのジョブの入力データを出力するジョブのジョブＩＤである「PrevJob」、入力ファイルパスを示す「入力データ」から構成される。PrevJobが記載されていないジョブ（job1）が、最初に実行可能なジョブである。
【００２７】
ジョブネット定義テーブル（Ｔ３）は、各MapReduceジョブの依存関係を定義するものであり、JobTracker（１０３）は、本テーブルに基づきMapReduceジョブの実行順序を決定する。JobConfigurator（１０１）は、ジョブ定義ファイル群（Ｆ１、Ｆ２）に記載された情報に基づき、ジョブネット定義テーブル（Ｔ３）に設定すべき情報を生成する。図６では、例えば、/data/hoge/を入力データとするJob1が最初に実行され、続いて、/data/foo1/を入力データとするJob2が実行されることを示している。
【００２８】
図７に、ジョブネット状態テーブル（Ｔ４）の詳細を示す。JobScheduler（１０２）は、ジョブネット定義テーブル（Ｔ３）に記載されたジョブネット全体のうち、実行中及び未実行のジョブに関するジョブの情報のみ本テーブルに記録され、本テーブルを利用して次に投入するジョブを決定する。本テーブルは、上述した「JobID」、「NextJob」、「PrevJob」、そのジョブの実行状態を表す「状態フラグ」、そのジョブの入力データパスである「入力データ」から構成される。「NextJob」、「PrevJob」は、ジョブネット定義テーブル（Ｔ３）と同じ意味である。
【００２９】
「状態フラグ」には、「未実行」または「実行中」のいずれかが記載される。実行中及び未実行のジョブの有無の判定は、例えば、ジョブ管理テーブル（Ｔ１）を参照することにより行われる。実行が完了したジョブに関するエントリ（ジョブＩＤが「JobID」に一致するエントリ）は、本テーブルから削除されるため、「状態フラグ」が「実行済」となるエントリが記載されることはない。また、実行が完了したジョブのジョブＩＤは「PrevJob」からも削除される。図７に示す例では、上述したJob1が実行中であり、続いて実行されるJob2が未実行であることを示している。
【００３０】
図８に、Child（２０３）内部のソフトウェア構成を示す。Childは、Map処理を行うMapTask（４１）とReduce処理を行うReduceTask（４４）から構成される。1つのタスクは、1つのChildプロセスで実行されるため、Map処理を実行するChild（２０３）はMapTask（４１）の部分のみが動作し、Reduce処理を実行するChild（２０３）はReduceTask（４４）の部分のみが動作することになる。Mapper（４３）は、Map処理を行うアプリケーションロジックであり、処理の種類毎にMapper（４３）が構成される。
【００３１】
Reducer（４７）は、Reduce処理を行うアプリケーションロジックであり、処理の種類ごとにReducer（４７）が構成される。本実施例ではジョブ統合を可能とするために、1つのChild（２０３）内に複数種類のMapper（４３）とReducer（４７）を実行できるようにしている。MapTask（４３）、ReduceTask（４４）共に、アプリケーションに依存しない共通的な処理を行う共通処理部（４２、４５）を保持する。共通処理としては、指定されたkeyに基づいてレコードをソートしたり、ソート済みのレコードをChild間で受け渡したりする処理があるが、詳細は後述する。
【００３２】
図９に、本システムが実行するジョブネットの一例を示す。このジョブネットは、２つのMapReduceジョブを統合したジョブであるJob1と、Job1の結果を入力データとするJob2から構成される。Job1において、ジョブ統合を行わずにMapper1とMapper2とを個別のMapReduceジョブ内で実行すると、Data1を２回読み込むことになる。ジョブ統合により、Mapper1とMapper2とで共通の入力データ（Data1）の読み込み処理を1回にすることができる。これは、Mapにおけるジョブ統合である。
【００３３】
図中に「ShuffleSetting」と記載されている部分は、Mapが出力したレコードをグルーピングするための設定情報である。具体的には、「key」は、グルーピングのためのkeyとして利用するフィールドを示す。例えば、Data2に対応する記述に「key=0,1」とあるが、これは出力レコードの0番目と1番目のフィールドをグルーピングのためのkeyとして利用するという意味である。「part」は、Mapが出力したレコードを分離する個数（パーティション数）であり、レコードを受け取るReducerのタスク数に等しい。
【００３４】
例えば、Data2に対応する記述に「part=20」とあるが、これによりReduce1のタスク数が20であることが分かる。同一keyは同一Reduceタスクに送信する必要があることから、この分離はkeyに基づき行われる。「fid」は、上記説明したグルーピングとパーティショニングとデータ送信を行う単位（フロー）の識別子である。同一フローとして処理されるレコード群について、keyとなるフィールドのデータ型とパーティション数は一致しなければならない。また、同一Reducerに複数種類のデータを渡す場合、それらデータは同一フローとする必要がある。図９では、Data2とData3は「fid」が異なるためフローは異なる。
【００３５】
図１１に、図９に示したジョブネットに対応するジョブ定義ファイル（Ｆ１）を示す。なお、以下では、ジョブ定義ファイル（Ｆ１）（後述するジョブ定義ファイル（Ｆ２）も同様。）をテキスト形式で記述されているが、実際にはオブジェクトとして定義されている。JobConfigurator（１０１）は、このジョブ定義ファイル（Ｆ１）からジョブの情報（ジョブネット定義テーブル（Ｔ３）に設定すべき情報）を抽出したり、JobTracker（１０３）は、Mapタグ、Reduceタグを取得する。図９に示したジョブネットに記載されていない内容として、ジョブ定義ファイル（Ｆ１）の２行目〜８行目の「datasection」タグ内に記述された、ジョブの入出力データのパス名がある。
【００３６】
図１１に示すジョブ定義ファイル（Ｆ１）の例では、図９に示したJob1で識別されるジョブのdatasectionタグには、パス「/data/hoge/」がdata1として定義され、パス「/data/foo1/」がdata4として定義され、パス「/data/foo2/」がdata5として定義されていることを示している。そして、mapsectionタグには、datasectionタグで定義されたdata1がmap1の入力データ（in）であり、data2がその出力データ（out）であることを示している。また、map2は、map1と同じdata1を入力データとし、data3がその出力データとして定義されていることを示している。
【００３７】
さらに、shufflesectionタグには、図９に示した２つのフロー、パーティション数、グルーピングキー、シャッフル&ソートされるデータのデータＩＤ（data2、data3）が定義されていることを示している。さらに、reducesectionタグでは、shufflesectionタグで定義されたdata2がreduce1の入力データ（in）であり、data4がその出力データ（out）であることを示している。同様に、shufflesectionタグで定義されたdata3がreduce2の入力データ（in）であり、data5がその出力データ（out）であることを示している。Job2についても、job1と同様に、図９に示したジョブネットに対応して定義されている。このように、ジョブ定義ファイルには、ジョブで実行されるmapやreduceで用いられる入出力データやそのジョブの構成（Map数、Reduce数、MapとReduceとの依存関係等）が定義されている。
【００３８】
図１０に、別のジョブネットの例を示す。このジョブネットは、複数のReducer（Reducer1とReducer2）が同一データ（Data2）を共通のデータとして入力にとる。ジョブ統合を行わない場合、Reducer毎に入力データを送信するため、Data2を２回送信することになる。一方、Data2をMapper1から1回だけReduceタスクを実行するChildに送信し、Child内で受信したレコードReducer1とReducer2に順に渡すことで、Data2の送信回数を1回にすることができる。これは、Reduceにおけるジョブ統合である。各Reducerに対してデータを異なるフローとして送信すると、そのデータは２回送信されてしまう。従って、Reduceにおけるジョブ統合では、複数のReduceに対して共通のデータを同一フローとして送信する必要がある。
【００３９】
図１２に、図１０のジョブネットに対応するジョブ定義ファイル（Ｆ２）を示す。記号の意味はＦ１と同様であるため、ここではその説明を省略するが、datasection、mapsection、shufflesection、reducesectionの各タグにおいて、図１１の場合と同様に、入出力データや、パス等が定義されている。本実施例では、図９、図１０に示したジョブネットを例として、統合ジョブ（Job1）が終了する前に、統合ジョブの後続ジョブ（Job2）を開始可能とするMapReduceジョブの実行方法について説明する。ジョブ定義ファイルにジョブネット全体の処理手順が記述されており、ユーザは、ジョブネットの実行時にジョブ定義ファイルを指定する。ユーザがジョブネットの実行を指示すると、JobConfigurator（１０１）は指示されたジョブ定義ファイルを解析し、ジョブの依存関係を抽出する。
【００４０】
例えば、図１１に示したジョブ定義ファイル（Ｆ１）の場合、ジョブＩＤが「job1」のジョブのMap処理の入力が「data1」で表されるデータであり、このデータのパス名を「datasection」タグ内から参照することで、ジョブの入力ファイルが「/data/hoge/」であることが分かる。同様に、Reduce処理の出力が「data4」及び「data5」で表されるデータであり、このデータのパス名を「datasection」タグ内から参照することで、ジョブの出力ファイルが「/data/foo1/」及び「/data/foo2/」であることが分かる。同様に、ジョブＩＤが「job2」のジョブの入力ファイルが「/data/foo1/」であり、出力ファイルが「/data/goo1/」であることが分かる。
【００４１】
ジョブＩＤ「job1」のジョブの入力ファイルを生成するジョブがないことから、「job1」は最初に実行されるジョブであり、ジョブＩＤ「job2」のジョブの入力ファイルを生成するジョブは「job1」であることから、「job2」は「job1」の後に実行されることが分かる。JobConfigurator（１０１）は、ジョブネット定義テーブル（Ｔ３）に、上の解析で判明したジョブの依存関係と各ジョブの入力データのパス名を登録する。今の場合、このテーブルの内容は、図６に示すものになる。
【００４２】
続いて、JobConfigurator（１０１）は、ジョブ定義ファイルの各ジョブの定義を解析し、パーティショニングに関するジョブのパラメタ設定を行う。この処理の詳細を、図１３に示す。
【００４３】
まず、JobConfigurator（１０１）は、ジョブ定義ファイル内の「job」タグをチェックしジョブ一覧を抽出し（Ｓ１０１）、以下の解析処理をすべてのジョブに対して実行する（Ｓ１０２）。JobConfigurator（１０１）は、解析対象のジョブの「reducesection」タグをチェックし、1つのジョブ内に複数のReducerが含まれているかどうかを確認する（Ｓ１０３）。もし、Reducerが複数含まれていた場合（Ｓ１０３；Ｙｅｓ）、JobConfigurator（１０１）は、そのジョブの出力データを入力とする後続ジョブが存在するかどうかを確認する。後続ジョブの有無は、上で説明したようにジョブの依存関係をチェックすることで判定できる（Ｓ１０４）。
【００４４】
Ｓ１０３でReducerが1つであった場合（Ｓ１０３；Ｎｏ）、JobConfigurator（１０１）は、統合ジョブとその後続ジョブの重ね合わせ実行は不可能であるため、通常のパーティション設定を行う。また、Ｓ１０４で後続ジョブがないことが分かった場合も（Ｓ１０４；Ｎｏ）、通常のパーティション設定を行う。ジョブ定義ファイルには、既にフロー毎にパーティション数が記載されているため、JobConfigurator（１０１）が行うパーティション設定では、対象とするジョブ全体でのパーティション数と各フローのパーティション番号の開始番号とを決定することになる。各フローは、「開始番号」〜「開始番号＋そのフローのパーティション数−１」の範囲のパーティション番号を利用する。
【００４５】
通常のパーティション設定では、ジョブ定義ファイルに記載されたフローに対して順にパーティション番号の開始番号を割り当てる。図９、図１１に示すジョブネットの場合、ジョブＩＤが「job2」のジョブに対して通常のパーティション設定を行うことになるが、「job2」はフローが1つ（fid=0のフロー）であるため、そのフローについてジョブ全体のパーティション番号の開始番号をそのフローのパーティション番号の開始番号（＝０）とする。また、ジョブ全体のパーティション数は上記フローのパーティション数である「5」に等しくなる。JobConfigurator（１０１）は、ジョブ全体のパーティション数を「shufflesecion」タグの「partall」属性に、各フローのパーティション番号の開始番号を「flow」タグの「partmin」属性に追記する。
【００４６】
修正して再定義した後のジョブ定義ファイル（Ｆ１）の内容を、図１４に示す。尚、フローが複数存在する場合は、JobConfigurator（１０１）は、異なるフローがパーティション番号の範囲を重複しないように割り当てていく。図１４に示す例では、Job2の「shufflesecion」タグに「partall」属性として5が設定され、フローＩＤが０（fid=0）のフローでは、パーティション数が5であり、その開始番号が0に設定されていることを示している。また、そのフローでは、key=1でグルーピングされた入力データdata6を処理することを示している。Job1についてもJob2と同様に、図９に示したジョブネットに対応して設定されている。
【００４７】
Ｓ１０３でReducerが複数存在することが確認され（Ｓ１０３；Ｙｅｓ）、さらにＳ１０４で後続ジョブが存在することが確認された場合（Ｓ１０４；Ｙｅｓ）、JobConfigurator（１０１）は、後続ジョブの入力ファイルを生成するReducerが入力データを受け取るフローと、それ以外のフローが存在することを確認する（Ｓ１０６）。
【００４８】
例えば、図９、図１１に示すジョブネット内のジョブＩＤが「job1」のジョブは、Ｓ１０３とＳ１０４の条件を満たすためこのチェックを行うことになるが、後続ジョブ（job2）の入力ファイル（/data/foor1）のデータＩＤは「data4」であり、これを出力するReducerは「reduce1」である。「reduce1」の入力データを送信するフローはフローＩＤが「0」のフローであり、このジョブには別のフロー（フローＩＤが「1」のフロー）が存在する。従って、このジョブは上記条件を満たす（Ｓ１０６；Ｙｅｓ）。
【００４９】
Ｓ１０６の条件を満たす場合（Ｓ１０６；Ｙｅｓ）、JobScheduler（１０２）は、後続ジョブの入力ファイルを生成するReducerが入力データを受け取るフローから入力データを受け取るすべてのReducerが、他のReducerよりも先に終了するようにReduce処理のスケジューリングを行う。これによって解析対象ジョブが終了する前に、後続ジョブを開始できるようになる。具体的には、後続ジョブの入力ファイルを生成するReducerが入力データを受け取るフローのパーティション番号を、他のフローのパーティション番号よりも小さい値になるようなパーティション設定を行う。
【００５０】
Hadoopでは、ジョブが開始されると、JobTracker（１０３）によりジョブを構成されるすべてのReduceタスクの情報がタスク管理テーブル（Ｔ２）に登録されるが、パーティション番号がReduceタスクのシーケンス番号となる。そして、Reduceタスクは、シーケンス番号が小さいものから順にTaskTracker（２０１）に割り当てられていく。通常、ジョブのReduce数はシステム全体が同時実行できるタスク数よりも十分大きく、シーケンス番号が小さいReduceから順に実行されていく。従って、パーティション番号を小さい値に設定したフローに関係するReducerの処理は先に終了するため、そのReducerが終了した時点で後続ジョブを開始できるようになる。
【００５１】
図９、図１１に示すジョブネット内のジョブＩＤが「job1」のジョブでは、フローＩＤが「0」のフローがＳ１０６の条件を満たすため、このフローのパーティション番号が小さい値になるようにする。従って、図１４に示したように、上記フローのパーティション番号の開始番号を「0」とし、上記フローのパーティション数が20であることから、もう片方のフロー（fid=1のフロー）のパーティション番号の開始番号を「20」とする。そして、このジョブ全体のパーティション数は全フローのパーティション数の合計である「40」となる。そして、図１４に示した修正して再定義した後のジョブ定義ファイル（Ｆ１）が作成される（Ｓ１０７）。
【００５２】
一方、Ｓ１０６の条件を満たさない場合（Ｓ１０６；Ｎｏ）、後続ジョブの入力データを作成するReducerとそれ以外のデータを作成するReducerとが、同一フローから入力データを受け取ることになる。この場合、フロー毎にパーティション番号の範囲を調整することで、後続ジョブの入力ファイルの作成を先に完了する上記方式を実行することができない。そこで、一部のパーティション番号については後続ジョブの入力データを作成するReducerのみ先に実行し、そのReducerの入力データを一時的にＨＤＤ上に残しておき、上記Reducerが完了したら、ＨＤＤ上に残しておいたデータを使って、他のReducerを実行するという方式でReduce処理を行う。
【００５３】
これによって、ジョブが終了する前に後続ジョブの入力データ生成を完了することが可能になる。ＨＤＤ上に残しておいたデータを使って実行するReducerは、最初にそのデータを読み込んで実行するReducerとは異なるタスクとして実行する。ここでは、後からＨＤＤ上のデータを読み込んで実行するReduceタスクを「拡張タスク」と呼ぶ。また、「拡張タスク」の入力データを最初に読み込んで処理するタスクを「キャッシュ処理タスク」と呼ぶ。
【００５４】
JobConfigurator（１０１）は、全パーティションのうちＨＤＤにデータを残すパーティションの数を決定し、その数を「flow」タグの「partcache」属性として登録する。拡張タスクの数は「partcache」の数に等しくなり、当該ジョブのパーティション数は「partcache」の数だけ増える。これを踏まえて、JobConfigurator（１０１）は、ジョブ定義ファイルの内容を修正して再定義する。
【００５５】
例えば、図１０、図１２に示したジョブネット内のジョブＩＤが「job1」のジョブに対してこの処理を行うことになるが、ＨＤＤにデータを残すパーティションの数を「5」と決定した場合、ジョブ定義ファイル（Ｆ２）を図１５に示すように、「flow」タグに「partcache」属性として5を設定する。各パラメタの内容については、図１４に示した場合と同様であるためここではそれらの説明を省略するが、ジョブ全体のパーティション数は、「flow」タグ内に指定されたパーティション数「20」から「partcache」の数だけ増えて「25」になる。
【００５６】
拡張タスクのシーケンス番号は、フローのパーティション番号の最後の部分が割り当てられる。上の例の場合、fid=0のフローはパーティション数が20であり、「partcache」を5に設定したので、シーケンス番号が0〜14のReduceタスクは、通常の方式で実行する。シーケンス番号が15〜19のReduceタスクは「キャッシュ処理タスク」であるためReduceタスク実行後も入力データをＨＤＤに残しておく。シーケンス番号が20〜24のReduceタスクは「拡張タスク」であるため、ＨＤＤに残したデータを入力データとして使って処理する(Ｓ１０８)。
【００５７】
このように、キャッシュタスクは、フローのパーティション番号のうちのあらかじめ定められた最後の部分が割り当てられ、拡張タスクは、フローのパーティション番号のReduceタスクのうちのキャッシュタスクと同じ数のReduceタスクであって、フローのパーティション番号より後の番号が割り当てられ、拡張タスクが、フローのパーティション番号のReduceタスクとは異なるタスクとして実行される。
【００５８】
続いて、Child（２０３）におけるMapタスクの処理フロー詳細を図１６に示す。以下では、図１６を参照してMapタスクの実行手順を説明する。
【００５９】
Mapタスクは、入力ファイルパスとそのデータの範囲（オフセット位置の範囲）とが指定されて起動される。複数種類の入力データが存在するジョブでは、異なる種類の入力データは異なるファイルとなるため、1つのMapタスクは必ず1つの種類の入力データを処理することになる。Mapタスク起動時にジョブ定義ファイル、ジョブＩＤ、及び処理対象となる入力ファイルのパス名がChild（２０３）に渡される。ジョブ定義ファイルには入力ファイルのパス名とそれを処理するMapperの対応関係が記載されており、共通処理部（４２）は、指定された入力ファイルのパス名と上記対応関係から動作すべきMapperを決定する（Ｓ２０１）。
【００６０】
共通処理部（４２）は、入力ファイルをレコード単位で読み取り、Ｓ２０１で選択したすべてのMapperに渡し、Mapperは、処理結果レコードを共通処理部に返す（Ｓ２０３）。共通処理部（４２）は、受け取ったレコードの種類に基づき、当該レコードのデータＩＤ（「map」タグ内の「out」属性に指定される情報）を求める。さらに、ジョブ定義ファイルの「flow」タグ内に記載された情報を参照し、上記データＩＤに基づきそのデータが送信されるフローのフローＩＤを求める。さらに、「flow」タグ内の「data」タグに記載された情報から、グルーピングのkeyとして使うべきフィールドを求める。共通処理部（４２）は、Mapperから受け取ったレコードの上記フィールドに記載された情報（keyデータ）から、以下の演算によってパーティション番号を決定する。
【００６１】
パーティション番号＝[Hash(keyデータ)] mod [パーティション数] + [フローのパーティション数]
上記計算式において、Hashはハッシュ関数であり、「mod」は割り算の余りを意味する。「フローのパーティション数」は当該「flow」タグ内の「part」属性に指定された値である。共通処理部（４２）は、Mapperから受け取ったレコードに対して、上で求めたパーティション番号、keyデータ、データＩＤを付与する。（Ｓ２０４）
共通処理部（４２）は、各種データを付与したレコードをメモリ上に蓄積し、データサイズが一定量を超えると、パーティション番号毎にkeyデータとデータＩＤに基づきソートし、ＨＤＤに一時ファイルとして出力する。尚、ソートの優先度は、まずkeyデータでソートしkeyデータが同じレコードはデータＩＤでソートする（Ｓ２０５）。
【００６２】
入力ファイルに記録されたすべてのレコードの処理が終わると、共通処理部（４２）は、メモリ内に残されたデータをパーティション番号毎にソートし、いったんＨＤＤ上に出力した後、パーティション番号毎にマージを行う。マージはkeyデータとデータＩＤに基づいたマージソートを行う。これによって、パーティション番号毎にソートされたデータが作成される（Ｓ２０６）。
【００６３】
続いて、Child（２０３）におけるReduceタスクの処理フロー詳細を図１７に示す。以下では、図１７を参照してReduceタスクの実行手順を説明する。Reduceタスクの起動時に、Child（２０３）にはタスクのシーケンス番号が渡される。ジョブ定義ファイルの「flow」タグには、「part」、「partmin」、「partcache」というパーティションに関する情報が記載されているが、ReduceTask（４４）内の共通処理部（４５）は、これらを使って自タスクの種類（「通常タスク」、「キャッシュ処理タスク」、「拡張タスク」）を判定する。
【００６４】
すなわち、「partcache」属性が存在し、シーケンス番号が([partmin]+[part]-[partcache])〜([partmin]+[part]-1)の間の値であれば「キャッシュ処理」タスクである。また、「partcache」属性が存在し、シーケンス番号が([partmin]+[part])〜([partmin]+[part]+[partcache]-1)の間の値であれば「拡張タスク」である。それ以外の場合は「通常タスク」である。フローが複数存在する場合、すべてのフローについてチェックを行う（Ｓ３０１）。
【００６５】
拡張タスクでないと判定された場合（Ｓ３０１；Ｎｏ）、共通処理部（４５）は、他のスレーブノード上のChild（２０３）がMap処理時に作成したデータを取得する。このとき、共通処理部（４５）は、パーティション番号がReduceタスクのシーケンス番号に一致するもののみ取得する。（Ｓ３０２）そして、共通処理部（４５）は、すべてのデータを取得したら、データに含まれるkeyデータとデータＩＤに基づきマージソートを行う。これによって、当該Reduceタスクの入力データ全体がソートされたデータが生成される（Ｓ３０３）。
【００６６】
Ｓ３０１で拡張タスクと判定された場合（Ｓ３０１；Ｙｅｓ）、共通処理部（４５）は、対応するキャッシュ処理タスクのシーケンス番号を求める。具体的には自タスクのシーケンス番号から「partcache」に記載された値を減算することで求める。そして、その番号に対応するパーティション番号に対応するデータをＨＤＤから取得する（Ｓ３０４）。
【００６７】
次に、共通処理部（４５）は、ジョブ定義ファイルに記載された情報と、実行中のタスクのシーケンス番号に基づき、実行中タスクの処理に対応するフローを求める。さらに、共通処理部（４５）は、そのフローで送受信されるデータを入力データとするReducer一覧を求める。「通常タスク」の場合、上記条件を満たすすべてのReducerを処理対象として選択する。「キャッシュ処理タスク」の場合、上記Reducerの中で後続ジョブの入力ファイルを生成するすべてのReducerを処理対象として選択する。「拡張タスク」の場合、上記Reducerの中で後続ジョブの入力ファイルを生成しないすべてのReducerを処理対象として選択する（Ｓ３０５）。
【００６８】
次に、共通処理部（４５）は、処理対象データから1レコードずつＳ３０５で選択したReducerに渡す。各Reducerの処理結果は分散ファイルシステムに格納される（Ｓ３０６、Ｓ３０７）。
【００６９】
処理対象データに記録された全レコードの処理が終了したら（Ｓ３０６；Ｙｅｓ）、共通処理部（４５）は、現在実行中タスクが「キャッシュ処理タスク」であるか否かを判定し（Ｓ３０８）、現在実行中タスクが「キャッシュ処理タスク」の場合（Ｓ３０８；Ｙｅｓ）、実行を完了したキャッシュ処理タスクのシーケンス番号に「partcache」の値を加算することで拡張タスクのシーケンス番号を求め、拡張タスクのタスクＩＤ（ジョブID_タスク種別_シーケンス番号）を自スレーブノードのTaskTrackerに、実行可能な拡張タスクＩＤとして登録する（Ｓ３０９）。一方、共通処理部（４５）は、現在実行中タスクが「キャッシュ処理タスク」でない場合（Ｓ３０８；Ｎｏ）、何もせずに処理を終了する。
【００７０】
JobScheduler（１０２）は、JobTracker（１０３）から定期的にタスクの実行状況を受け取り、実行中のタスク数がシステム全体で同時実行可能なタスク数を下回った場合、実行可能な別のジョブを探す。ジョブが実行可能となるには入力データが揃っている必要があるため、JobScheduler（１０２）は、ジョブネット状態テーブル（Ｔ４）の情報を参照して、上記テーブル「入力データ」から当該ジョブの入力データのパス名を求め、そのデータの生成が完了しているかどうかを確かめる。また、開始可能なジョブは、上記テーブルの「状態フラグ」が「未実行」のものに限られる。この２つの条件を満たすジョブが、実行可能なジョブとなる。
【００７１】
JobScheduler（１０２）は、実行可能なジョブを見つけたら、JobScheduler（１０２）はジョブ定義ファイルを参照し、そのジョブに含まれる拡張タスクのシーケンス番号一覧をＳ３０１で説明した方法で求める。そして、そのジョブの開始をJobTracker（１０３）に指示する。このとき、ジョブＩＤと上記拡張タスクのシーケンス番号一覧とをJobTracker（１０３）に送る。JobTracker（１０３）は、受け取ったシーケンス番号一覧に基づき、タスク管理テーブル（Ｔ２）の「拡張フラグ」を決定する。
【００７２】
このように、本実施例では、複数のReducerを含むジョブ統合型のMapReduceジョブ（第一のジョブ）と、上記MapReduceジョブの出力データを入力データとするMapReduceジョブ（第二のジョブ）が存在するとき、第一のMapReduceのジョブ内で実行されるReducerのうち、第二のMapReduceジョブの入力データを生成するReducerと、第二のMapReduceジョブの入力データを生成しないReducerを異なるタスクとして実行し、第二のMapReduceジョブの入力データを生成するReducerが、第二のMapReduceジョブの入力データを生成しないReducerよりも先に完了するように、上記タスクの実行順序を制御する。したがって、MapReduceにおいてジョブ統合を行った場合でも、統合前と同様にジョブの同時実行を行うことが可能となり、これによってサーバリソースの稼働率低下を抑止し、結果的にジョブネットの実行時間を短縮することができる。
【００７３】
ジョブネットの実行監視を行うジョブ監視システムのモニタリング画面の例を、図１８に示す。図１８に示す例では、実行対象となるジョブ４０１、４０２が画面上に表示され、これらのジョブを繋ぐ矢印によって、ジョブの依存関係（実行順序）が示されている。また、各ジョブの下には、Map及びReduceの進捗率を示す情報４０３、４０４が、それぞれ表示している。この例では、Job1のReduceの進捗率が95%、Job2のMapの進捗率が5%となっており、このことからJob1の終了を待たずにJob1の次のジョブ（Job2）が開始されていることが分かる。
【００７４】
ジョブ監視システムはマスタノード上で稼働し、JobScheduler（１０２）から実行中のジョブネットを構成するジョブの依存関係（すなわちジョブネット定義テーブル（Ｔ３）の内容）を取得する。また、ジョブ監視システムは、JobTracker（１０３）から各ジョブの進捗状況（すなわちジョブ管理テーブル（Ｔ１）のMap進捗およびReduce進捗）を取得する。そして、取得した情報に基づき、図１８に示したモニタリング画面を作成する。Mapの進捗率は、（終了したMapタスク数）/（全Mapタスク数）によって算出する。Reduceの進捗率についても同様である。尚、ジョブ監視システムは、マスタノードと通信可能なサーバであればどのサーバ上で稼働してもよく、別の好適な構成としてマスタノードは別にジョブ監視システムを稼働するためのノードを追加する構成が考えられる。
【００７５】
なお、本実施例では、上述した各処理を行うプログラムがＣＰＵ（３０１）によってＨＤＤ（３０９）から読み出されてメモリ（３０３）上に展開、実行されることとして説明した。しかし、これらのプログラムを、インストール可能な形式又は実行可能な形式のファイルでＣＤ−ＲＯＭ、フレキシブルディスク（ＦＤ）、ＣＤ−Ｒ、ＤＶＤ等のコンピュータで読み取り可能な記録媒体に記録して提供することももちろん可能である。さらに、上述した各処理を行うプログラムを、インターネット等のネットワークに接続されたコンピュータ上に格納し、ネットワーク経由でダウンロードや配信することにより提供してもよい。
【符号の説明】
【００７６】
１０  マスタノード
２０  スレーブノード
１０１  ＪｏｂＣｏｎｆｉｇｕｒａｔｏｒ
１０１  ＪｏｂＳｃｈｅｄｕｌｅｒ
１０２  ＪｏｂＴｒａｃｋｅｒ
１０３  ＴａｓｋＳｃｈｅｄｕｌｅｒ
２０３  Ｃｈｉｌｄ
Ｔ１  ジョブ管理テーブル
Ｔ２  タスク管理テーブル
Ｔ３  ジョブネット定義テーブル
Ｔ４  ジョブネット状態テーブル
Ｆ１、Ｆ２  ジョブ定義ファイル。
【図１】
2015095096.tif 000003
【図２】
2015095096.tif 000004
【図３】
2015095096.tif 000005
【図４】
2015095096.tif 000006
【図５】
2015095096.tif 000007
【図６】
2015095096.tif 000008
【図７】
2015095096.tif 000009
【図８】
2015095096.tif 000010
【図９】
2015095096.tif 000011
【図１０】
2015095096.tif 000012
【図１１】
2015095096.tif 000013
【図１２】
2015095096.tif 000014
【図１３】
2015095096.tif 000015
【図１４】
2015095096.tif 000016
【図１５】
2015095096.tif 000017
【図１６】
2015095096.tif 000018
【図１７】
2015095096.tif 000019
【図１８】
2015095096.tif 000020