(57)【要約】【課題】同時にMapタスクまたはReduceタスクを実行するMapReduceジョブから構成されるジョブネットの実行時間を短縮する。【解決手段】同時にMapタスクまたはReduceタスクを実行して複数のMapReduceジョブを実行するMapReduceジョブ管理システムであって、複数のMapReduceジョブ同士の関係とMapReduceジョブを構成するMapタスクまたはReduceタスクとが記述された定義ファイルに基づいて、互いに関係のない第一のジョブと第二のジョブとを複数のMapReduceジョブの中から特定し、MapタスクまたはReduceタスクの実行状態に基づいて、第一のジョブを構成するMapタスクまたはReduceタスクのうち、未実行のMapタスクまたはReduceタスクが所定値より少なくなったか否かを判定し、未実行のMapタスクまたはReduceタスクが所定値より少なくなったと判定した場合に、第二のジョブを開始するジョブスケジュール部、を備える。【選択図】  図１１
2015090688.tif 000002
【特許請求の範囲】
【請求項１】
同時にMapタスクまたはReduceタスクを実行して複数のMapReduceジョブを実行するMapReduceジョブ管理システムであって、
前記複数のMapReduceジョブ同士の関係とMapReduceジョブを構成するMapタスクまたはReduceタスクとが記述された定義ファイルに基づいて、互いに関係のない第一のジョブと第二のジョブとを前記複数のMapReduceジョブの中から特定し、前記MapタスクまたはReduceタスクの実行状態に基づいて、前記第一のジョブを構成するMapタスクまたはReduceタスクのうち、未実行のMapタスクまたはReduceタスクが所定値より少なくなったか否かを判定し、前記未実行のMapタスクまたはReduceタスクが所定値より少なくなったと判定した場合に、前記第二のジョブを開始するジョブスケジュール部、
を備えることを特徴とするMapReduceジョブ管理システム。
【請求項２】
前記未実行のMapタスクまたはReduceタスクをタスクキューにキューイングし、優先度が高いジョブを構成するMapタスクまたはReduceタスクを、前記タスクキューから優先的にコンピュータに割り当てるタスクスケジュール部、
をさらに備えることを特徴とする請求項１に記載のMapReduceジョブ管理システム。
【請求項３】
前記ジョブスケジュール部は、前記複数のMapReduceジョブの実行中に、実行中ジョブを構成する未実行Mapタスク数又は未実行Reduceタスクの数が前記所定値より少なくなっている間、前記優先度を、他のジョブよりも高い優先度に変更する、
ことを特徴とする請求項２に記載のMapReduceジョブ管理システム。
【請求項４】
前記ジョブスケジュール部は、前記第一のジョブの出力データを入力データとする前記第二のジョブの数が多いジョブを、優先的に開始する第一のジョブとして特定する、
ことを特徴とする請求項１〜３のいずれか１項に記載のMapReduceジョブ管理システム。
【請求項５】
前記ジョブスケジュール部は、後続のジョブフローの経路長が最大のジョブを優先的に開始する前記第二のジョブとして特定する、
ことを特徴とする請求項１〜４のいずれか１項に記載のMapReduceジョブ管理システム。
【請求項６】
同時にMapタスクまたはReduceタスクを実行して複数のMapReduceジョブを実行するMapReduceジョブ管理方法であって、
ジョブスケジュール部により、前記複数のMapReduceジョブ同士の関係とMapReduceジョブを構成するMapタスクまたはReduceタスクとが記述された定義ファイルに基づいて、互いに関係のない第一のジョブと第二のジョブとを前記複数のMapReduceジョブの中から特定する第一のステップと、
ジョブスケジュール部により、前記MapタスクまたはReduceタスクの実行状態に基づいて、前記第一のジョブを構成するMapタスクまたはReduceタスクのうち、未実行のMapタスクまたはReduceタスクが所定値より少なくなったか否かを判定する第二のステップと、
ジョブスケジュール部により、前記未実行のMapタスクまたはReduceタスクが所定値より少なくなったと判定された場合に、前記第二のジョブを開始する第三のステップと、
を含むことを特徴とするMapReduceジョブ管理方法。
【請求項７】
タスクスケジュール部により、前記未実行のMapタスクまたはReduceタスクをタスクキューにキューイングし、優先度が高いジョブを構成するMapタスクまたはReduceタスクを、前記タスクキューから優先的にコンピュータに割り当てる第四のステップ、
をさらに含むことを特徴とする請求項６に記載のMapReduceジョブ管理方法。
【請求項８】
前記第三のステップでは、前記ジョブスケジュール部により、前記複数のMapReduceジョブの実行中に、実行中ジョブを構成する未実行Mapタスク数又は未実行Reduceタスクの数が前記所定値より少なくなっている間、前記優先度を、他のジョブよりも高い優先度に変更する、
ことを特徴とする請求項７に記載のMapReduceジョブ管理方法。
【請求項９】
前記第一のステップでは、前記ジョブスケジュール部により、前記第一のジョブの出力データを入力データとする前記第二のジョブの数が多いジョブを、優先的に開始する第一のジョブとして特定する、
ことを特徴とする請求項６〜８のいずれか１項に記載のMapReduceジョブ管理方法。
【請求項１０】
前記第一のステップでは、前記ジョブスケジュール部により、後続のジョブフローの経路長が最大のジョブを優先的に開始する前記第二のジョブとして特定する、
ことを特徴とする請求項６〜９のいずれか１項に記載のMapReduceジョブ管理方法。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、並列分散ジョブから構成されるジョブネットのジョブをスケジューリングして管理するジョブ管理システム、ジョブ管理方法に関する。
【背景技術】
【０００２】
近年、大規模データを処理するための並列分散型フレームワークとしてMapReduce（非特許文献１）が提案されている。MapReduceは、MapとReduceという2つの並列処理から構成される処理モデルである。Mapは、互いに依存関係のない入力データを並列処理し、処理結果をkey-value形式のデータとして出力する。出力結果は、key毎にグルーピングされ、Reduceへの入力になる。Reduceは、このグループ毎に並列処理し結果を出力する。原理的には、MapやReduceの入力データの単位（レコード）毎に並列化可能であるが、細かく分割しすぎると処理のオーバーヘッドが大きくなるため、通常、複数のレコードを束ねたタスク毎に並列処理する。
【０００３】
MapReduceのOSS(Open Source Software)としてHadoopがあるが、Hadoopのバージョン1.xではサーバ毎に同時実行可能なタスク数の上限を、Map処理及びReduce処理それぞれに対して設定する。従って、Map処理を行うタスク（Mapタスク）とReduce処理を行うタスク（Reduceタスク）両方が十分に存在すれば、両方のタスクを上限までサーバ上で同時実行可能である。一方、実行すべきタスクがMapタスクのみ、もしくはReduceタスクのみという状況では、片方の種類（MapまたはReduce）のタスクのみ上限まで同時実行することになる。従って、両方の種類のタスクを実行する場合と、1種類のタスクのみ実行する場合とで計算機負荷が異なり、結果的に1種類のタスクのみ実行する場合に計算機リソースの空き時間が発生するという課題がある。
【０００４】
これに対して、特許文献1では、各ジョブのMapタスク及びReduceタスクの処理時間を予測し、予測に基づきMapタスクとReduceタスクの両方がなるべく実行されるようにジョブの実行順序をスケジューリングしている。これによって、なるべく両方の種類のタスクが実行されている状態を維持する。また、別の対応方法として、Hadoopのバージョン2.xでは、各サーバで実行可能なタスク数の上限をMapタスクとReduceタスクの合計数とすることで、実行すべきタスクがMapタスクもしくはReduceタスクのみであってもサーバリソースの稼働率の低下を防ぐようにしている。すなわち、MapタスクとReduceタスク両方を実行する状況でも、片方を実行する状況でも、実行中タスク数が一定であるため、サーバ負荷の変動を少なくすることができ、適切なチューニングによってサーバリソースの空き時間が少ない状態を維持することができる。
【０００５】
通常1つのMapReduceジョブでやりたい処理が実行できるわけではなく、複数のMapReduceジョブを定められた順に実行する必要がある。このようなジョブ間の依存関係が定義された一連のジョブ群（ジョブネット）の実行管理を行うシステムは、ジョブ管理システムと呼ばれる。複数のMapReduceジョブから構成されるジョブネットを考えることができ、このようなジョブネットの実行にもジョブ管理システムを利用することができる。
【先行技術文献】
【特許文献】
【０００６】
【特許文献１】ＵＳ２０１３／０１６７１５１
【非特許文献】
【０００７】
【非特許文献１】MapReduce: Simplified Data Processing on Large Clusters, OSDI’04
【発明の概要】
【発明が解決しようとする課題】
【０００８】
MapReduceでは、複数のMap/Reduceタスクを並列処理するが、通常、システム全体で同時実行可能なタスクの数に比べて処理対象のMap/Reduceタスクの数は多い。そこで、MapReduceでは、実行中ジョブのタスクをキューで管理し、キューから余力があるサーバへ順次タスクを割り当てる方式を採る。また、MapReduceの原理的な制約として、すべてのMapタスクが完了しないとReduceタスクが開始できないという特性がある。
【０００９】
このようなシステムにおいて、Mapの終盤では少数の実行中のMapタスクの完了を待つため、ジョブの多重度が低下し、多くのサーバでリソースの空き時間が発生する。また、後続ジョブを開始するには実行中のジョブが完了する必要があるため、同様にReduceの終盤においてもジョブの多重度が低下し、サーバリソースの空き時間が発生する。このようにMapReduceにはサーバリソースの利用効率に関する課題がある。例えば、図１４に示すように、Mapタスクを開始時からしばらくの間は、高い多重度で効率よくジョブが実行されるが、Mapタスクの終盤では徐々に多重度が低下し、リソースの空き時間が生じているサーバが多くなる時間が発生する（Ｔ１）。同様に、Reduceタスクを開始時からしばらくの間は、高い多重度で効率よくジョブが実行されるが、Reduceタスクの終盤では徐々に多重度が低下し、リソースの空き時間が生じているサーバが多くなる時間が発生する（Ｔ２）。
【００１０】
特許文献1の技術は、異なるジョブのMapとReduceフェーズを同時実行することでサーバリソースの空き時間発生の抑止を意図しているが、MapとReduceの実行時間に偏りがある場合には、サーバリソースの空き時間が発生してしまう。Hadoopバージョン2.xでは、この課題は解消されているが、MapもしくはReduceの終盤におけるジョブの多重度低下という課題は依然として残っている。
【００１１】
これに対して、Hadoopバージョン2.xにおいて、特許文献1で実行しているように複数のジョブを同時実行することで、MapもしくはReduceの終盤におけるジョブの多重度低下を抑止する方式が考えられる。しかし、実行中のジョブが同時にMapの終盤もしくはReduceの終盤を実行する状況になった場合、サーバリソースの空き時間が発生してしまう。
【００１２】
また、MapReduceジョブから構成されるジョブネットを実行する場合、ジョブ間の依存関係の制約の中で同時実行可能なジョブを選択していくことになる。このとき、実際のジョブのスケジューリング方法によっては、途中で同時実行可能なジョブがなくなる場合があり、同時実行可能なジョブを常に確保できるようなジョブのスケジューリングを行う必要がある。上記特許文献1ではこの課題については考慮されていない。
【００１３】
本発明は、上記に鑑みてなされたものであって、同時にMapタスクまたはReduceタスクを実行するMapReduceジョブから構成されるジョブネットの実行時間を短縮することが可能なジョブ管理システム、ジョブ管理方法を提供することを目的とする。
【課題を解決するための手段】
【００１４】
上述した課題を解決し、目的を達成するために、本発明にかかるMapReduceジョブ管理システムは、同時にMapタスクまたはReduceタスクを実行して複数のMapReduceジョブを実行するMapReduceジョブ管理システムであって、前記複数のMapReduceジョブ同士の関係とMapReduceジョブを構成するMapタスクまたはReduceタスクとが記述された定義ファイルに基づいて、互いに関係のない第一のジョブと第二のジョブとを前記複数のMapReduceジョブの中から特定し、前記MapタスクまたはReduceタスクの実行状態に基づいて、前記第一のジョブを構成するMapタスクまたはReduceタスクのうち、未実行のMapタスクまたはReduceタスクが所定値より少なくなったか否かを判定し、前記未実行のMapタスクまたはReduceタスクが所定値より少なくなったと判定した場合に、前記第二のジョブを開始するジョブスケジュール部、を備えることを特徴とするMapReduceジョブ管理システムとして構成される。
【００１５】
また、本発明は、上記MapReduceジョブ管理システムで行われるMapReduceジョブ管理方法としても把握される。
【発明の効果】
【００１６】
本発明によれば、同時にMapタスクまたはReduceタスクを実行するMapReduceジョブから構成されるジョブネットの実行時間を短縮することが可能となる。
【図面の簡単な説明】
【００１７】
【図１】システム全体の構成を示す図である。
【図２】マスタノードにおけるソフトウェア構成を示す図である。
【図３】計算機のハードウェア構成を示す図である。
【図４】ジョブネットの概念説明を示す図である。
【図５】ジョブ管理テーブルの一例を示す図である。
【図６】タスク管理テーブルの一例を示す図である。
【図７】ジョブネット定義テーブルの一例を示す図である。
【図８】ジョブネット状態テーブルの一例を示す図である。
【図９】実行中ジョブ状態テーブルの一例を示す図である。
【図１０】ジョブ履歴テーブルの一例を示す図である。
【図１１】ジョブスケジューリング制御の全体を説明するフローを示す図である。
【図１２】ジョブスケジューリング制御における投入ジョブ決定処理の詳細を説明するフローを示す図である。
【図１３】ジョブスケジューリング制御におけるジョブ優先度変更処理の詳細を説明するフローを示す図である。
【図１４】従来技術において、ジョブの多重度が低下し、多くのサーバでリソースの空き時間が発生する様子を示す図である。
【発明を実施するための形態】
【００１８】
以下に添付図面を参照して、本発明にかかるジョブ管理システム、ジョブ管理方法の実施の形態を詳細に説明する。
【００１９】
図１は、本発明をMapReduceのOSS(Open Source Software)の１つであるHadoopに適用した場合のシステム構成図である。本システムは、１台のマスタノード（１０）、４台のスレーブノード（２０）、1台のユーザ端末（３０）から構成される（ノードとはサーバと同義である）。マスタノード（１０）およびスレーブノード（２０）は、並列分散処理を実現するサーバ群であり、ユーザ端末（３０）は、ユーザがジョブを投入するための端末である。尚、本実施例では、スレーブノードを４台としているが、本発明を適用する上で、スレーブノードの台数は1台以上の任意の台数であってもよい。
【００２０】
マスタノード（１０）では、ジョブネット全体の実行制御を行うJobScheduler（１０１）、MapReduceジョブの実行制御を行うJobTracker（１０２）、分散ファイルシステムにおけるメタデータ管理を行うNameNode（１０３）が稼働する。スレーブノード（２０）では、Mapタスク・Reduceタスクの実行管理を行うTaskTacker(２０１)、分散ファイルシステムにおけるデータアクセス処理を行うDataNode（２０２）、タスクの実体であるChild（２０３）が稼働する。Child（２０３）は、MapReduceジョブ実行中のみ動作するプロセスであり、TaskTracker（２０１）から起動される。JobTracker（１０２）、NameNode(１０３)、TaskTracker（２０１）、DataNode(２０２)、Child（２０３）は相互に連携してMapReduceジョブの実行機能を提供する。JobScheduler（１０１）は、JobTracker（１０２）に対してMapReduceジョブの投入を指示する。
【００２１】
図２に、マスタノード（１０）内のソフトウェア構成の詳細を示す。尚、図２では、本発明とは関係の低いNameNode（１０３）は省略している。JobTracker（１０２）は、MapReduceジョブの進捗を管理するジョブ管理テーブル（Ｔ１）、タスクの実行状態を管理するタスク管理テーブル（Ｔ２）を保持する。JobScheduler（１０１）は、ジョブネットの構成を定義するジョブネット定義テーブル（Ｔ３）、ジョブネットの実行状態を管理するジョブネット状態テーブル（Ｔ４）、実行中ジョブの状態を管理する実行中ジョブ状態テーブル（Ｔ５）、ジョブ履歴を管理するジョブ履歴テーブル（Ｔ６）を保持する。
【００２２】
JobTracker（１０２）は、実行中のMapReduceジョブを構成するタスク群を、キューにより管理し、キューから余力があるサーバへキューイングして順次スレーブノード（２０）に割り当てる。TaskScheduler（１０４）は、この割り当て処理を行うモジュールである。本実施例では、TaskScheduler(１０４)の機能として、ジョブ毎にタスク割り当ての優先度が設定できることを想定する。すなわち、TaskScheduler(１０４)は、ジョブ毎に優先度を設定でき、優先度が異なるジョブが同時実行されている場合、優先度が高いジョブのタスクを優先的にスレーブノードに割り当てる。これは、本システムを実行する上で必要な機能である。尚、この優先度はスレーブノードへのタスク割り当てに関する優先度であって、ＯＳにおけるプロセスの実行優先度とは異なる。
【００２３】
また、本実施例では、同時実行タスク数の制限機能に関して、MapタスクとReduceタスクそれぞれに対して独立に制限値を設けるのではなく、MapタスクとReduceタスクの合計数に対して制限することを仮定する。この仮定は、本実施例では必須ではないが、これによって本システムにおける効果を得やすくなる。
【００２４】
本システムを構成するノード及びユーザ端末のハードウェア構成を、図３に示す。いずれも一般的な計算機やコンピュータである。各ノード及びユーザ端末の内部では、ＣＰＵ（３０１）、ＬＡＮＩ／Ｆ（３０２）、メモリ（３０３）、入出力Ｉ／Ｆ（３０４）、ＨＤＤＩ／Ｆ（３０５）が相互に接続されている。ＬＡＮＩ／Ｆ（３０２）は、ネットワークに接続され、ノード（１０、２０）同士やユーザ端末（３０）とノードの間で通信するときに利用する。ＨＤＤＩ／Ｆ（３０５）には、ＨＤＤ（３０９）が接続されている。
【００２５】
各種プログラムやデータは、ＨＤＤ（３０９）に保存されており、ＨＤＤＩ／Ｆ（３０５）を経由して読み出され、メモリ（３０３）上に展開される。また、展開されたプログラムは、ＣＰＵ（３０１）によって実行される。入出力Ｉ／Ｆ（３０４）には、ディスプレイ（３０６）、キーボード（３０７）、マウス（３０８）が接続されており、ユーザに対するインタフェースを提供する。尚、マスタノード（１０）、スレーブノード（２０）には、入出力Ｉ／Ｆ（３０４）に接続される入出力デバイスは必ずしも必要ではない。
【００２６】
図５は、JobTracker（１０２）が管理するジョブ管理テーブル（Ｔ１）の詳細である。ジョブ管理テーブル（Ｔ１）は、MapReduceジョブ毎に、ジョブのＩＤである「JobID」、Mapタスクの総数である「Map数」、Map処理の進捗状況を表す「Map進捗」、Reduceタスクの総数である「Reduce数」、Reduce処理の進捗状況を表す「Reduce進捗」、ジョブの優先度である「優先度」を対応付けて保持する。MapやReduceの進捗は「実行済みタスク数／実行中タスク数／未実行タスク数」という形式で記載する。図５では、例えば、JOB1のMap数およびReduce数は、それぞれ800および140であり、Mapの実行済みタスク数が20、実行中タスク数が32、未実行タスク数が748であることを示している。また、140あるReduceのタスクの全てが実行済みであることを示している。
【００２７】
図６は、JobTracker（１０２）が管理するタスク管理テーブル（Ｔ２）である。タスク管理テーブル（Ｔ２）は、タスクのIDである「TaskID」とタスクの実行状態を管理する「状態フラグ」を対応付けて保持する。「TaskID」は、「ジョブID_タスク種別_シーケンシャル番号」という形式で記載する。タスク種別によって、Mapタスク（mと記載）とReduce（rと記載）とを区別する。例えば、ジョブIDがJ1、シーケンシャル番号が0001のMapタスクの場合、タスクIDは「J1_m_0001」となる。「状態フラグ」には、未実行、実行済み、実行中の３種類がある。
【００２８】
JobTracker（１０２）は、MapReduceジョブが投入されると、ジョブの設定情報（例えば、ジョブ定義ファイルに定義されているMapタグ、Reduceタグ等に記載されている情報）からMapタスク数、Reduceタスク数、ジョブの優先度を取得し、ジョブ管理テーブル（Ｔ１）に該当するジョブのエントリを追加する。また、JobTracker（１０２）は、タスク管理テーブル（Ｔ２）にジョブを構成する全タスクに関するエントリを追加する。また、JobTracker（１０２）は、ジョブの設定情報からジョブの入出力データパスを取得し、これらの情報に基づき、ジョブを開始する。
【００２９】
ジョブが開始されると、TaskScheduler（１０４）は、ジョブの優先度に応じて各タスクをスレーブノードに割り当てていく。割り当てたタスクは、タスク管理テーブル（Ｔ２）の状態フラグが「実行中」となり、スレーブノード上で実行が完了したら状態フラグは「実行済」になる。ジョブを構成するすべてのタスクの実行が完了したら、ジョブが完了したことになるため、JobTracker（１０２）は、ジョブ管理テーブル（Ｔ１）及びタスク管理テーブル（Ｔ２）から、そのジョブに関するすべてのエントリを削除する。尚、MapReduceはすべてのMapタスクが完了しないと、Reduceタスクを開始することができない処理モデルであるため、TaskScheduler（１０４）は、すべてのMapタスクが完了するまでReduceタスクの割り当てを行わない。
【００３０】
図７に、ジョブネット定義テーブル（Ｔ３）の詳細を示す。本テーブルは、ジョブＩＤである「JobID」、そのジョブの出力データを入力データとして利用するジョブのジョブＩＤである「NextJob」、そのジョブの入力データを出力するジョブのジョブＩＤである「PrevJob」、入力ファイルパスを示す「入力データ」から構成され、これらが互いに対応付けられている。図７に示した内容をDAG（Directed acyclic graph）として図示したものが図４である。
【００３１】
PrevJobが記載されていないジョブ（J1、J2、J3）が最初に実行可能なジョブであり、NextJobが記載されていないジョブ（J4、J6）が最後の方で実行するジョブとなる。図４に示す例では、ジョブJ1、J2が終了するとジョブJ4が実行され、ジョブJ2、J3が終了すると、ジョブJ5が実行され、ジョブJ5が終了すると、ジョブJ6が実行されることを示している。また、図７に示すジョブネット定義テーブル（Ｔ３）では、ジョブJ1の入力データは「/data/j1/」であることを示している。他のジョブについても同様に、それぞれ入力データが定義されている。
【００３２】
図８に、ジョブネット状態テーブル（Ｔ４）の詳細を示す。本テーブルには、ジョブネット定義テーブル（Ｔ３）に記載されたジョブネット全体のうち、実行中及び未実行のジョブに関するジョブの情報のみ記載され、JobScheduler（１０１）は、本テーブルを利用して次に投入するジョブを決定する。本テーブルは、ジョブIDである「JobID」、「NextJob」「PrevJob」、ジョブの実行状態を表す「状態フラグ」、ジョブの入力データのサイズである「入力サイズ」、ジョブの出力データのサイズである「出力サイズ」、ジョブのMapタスク数である「Map数」、ジョブのReduceタスク数である「Reduce数」から構成され、これらが互いに対応付けられている。
【００３３】
「NextJob」、「PrevJob」は、ジョブネット定義テーブル（Ｔ３）と同じ意味である。「状態フラグ」には、「未実行」、「Map」、「Reduce」のいずれかが記載される。「Map」はMap処理を実行中であることを示し、「Reduce」はReduce処理を実行中であることを示す。実行が完了したジョブに関するエントリ（ジョブＩＤが「JobID」に一致するエントリ）は本テーブルから削除されるため、「状態グラフ」が「実行済」となるエントリが記載されることはない。また、実行が完了したジョブのジョブＩＤは「PrevJob」から削除される。
【００３４】
図８に示す例では、ジョブJ1の次にはジョブJ4が実行されるが、現時点では未実行であることを示している。また、ジョブJ1の入力となるデータのサイズは100GBであり、出力となるデータのサイズは70GBであることを示している。さらに、ジョブJ1が実行するMap数およびReduce数は、それぞれ800および140であることを示している。他のジョブについても同様に、それぞれ定義されている。
【００３５】
「出力サイズ」は、過去のジョブ履歴とジョブの入力データのサイズに基づき、推定した値になる。「入力サイズ」は、ジョブの入力データが確定したものはそのデータサイズになるが、入力データが確定していないものは、「PrevJob」の「出力サイズ」に記載された値の合計値とする。「Map数」及び「Reduce数」は、「入力サイズ」及び「出力サイズ」に基づき決定する。本実施例では、「入力サイズ」256MバイトにつきMapタスクを1個、「出力サイズ」512MバイトにつきReduceタスクを1個としている。あるジョブが終了すると、そのジョブの出力サイズが確定するため、そのジョブを「PrevJob」として持つ他のジョブの「入力サイズ」が変化する。これに応じて、「出力サイズ」、「Map数」、「Reduce数」も変化する。実際にジョブ実行に利用される「Map数」、「Reduce数」は、ジョブ開始時に確定する。
【００３６】
図９に、実行中ジョブ状態テーブル（Ｔ５）の詳細を示す。本テーブルには実行中のジョブの状態が記載され、JobScheduler（１０１）は、本テーブルを利用して実行中のジョブの優先度を変更する。本テーブルは、ジョブを識別するためのＩＤである「JobID」、当該ジョブの未実行Map数である「残りMap数」、当該ジョブの未実行Reduce数である「残りReduce数」、ジョブの緊急度を表す「緊急フラグ」から構成され、これらが互いに対応付けられている。
【００３７】
JobScheduler（１０１）は、定期的（例：1秒ごと）にJobTracker（１０２）から実行中ジョブの未実行Map数と未実行Reduce数を受け取り、実行中ジョブ状態テーブル（Ｔ５）に反映する。JobScheduler（１０１）は、Map処理実行中のジョブの残りMap数が基準値以下になった場合、またはReduce処理実行中のジョブの残りReduce数が基準値以下になった場合に、「状態フラグ」を「CRITICAL」にする。これ以外の場合、「緊急フラグ」は「NORMAL」になる。上記基準値は、MapとReduceで異なる値としてもよい。
【００３８】
図１０に、ジョブ履歴テーブル（Ｔ６）の詳細を示す。本テーブルは、ジョブを識別するためのＩＤである「JobID」、当該ジョブの入力データのサイズである「入力サイズ」、当該ジョブの出力データのサイズである「出力サイズ」、ジョブを実行した時刻である「タイムスタンプ」から構成され、これらが互いに対応付けられている。このテーブルには、過去に実行したジョブの履歴が記録され、JobScheduler（１０１）は、このテーブルに記録された情報に基づき、未実行ジョブの出力データサイズを推定する。
【００３９】
具体的には、JobScheduler（１０１）は、「入力サイズ」と「出力サイズ」の比率を本テーブルに記載された情報から計算し、ジョブネット状態テーブル（Ｔ４）の「入力サイズ」に記載された値と上記比率の積を「出力サイズ」に記載すべき値とする。尚、推定方法として、上に挙げた方法以外を用いてもよい。また、JobScheduler（１０１）は、ジョブを実行した後、一定時間が経過した履歴は削除するが、このために「タイムスタンプ」を利用する。
【００４０】
図１１は、JobScheduler（１０１）のジョブスケジューリング機能の詳細な動作を説明するフロー図である。さらに、図１１のＳ１０２（投入ジョブ決定）の詳細を図１２に、図１１のＳ１０６（ジョブ優先度変更）の詳細を図１３に示す。以下ではこれらのフロー図を用いて、JobScheduler（１０１）の動作を説明する。
【００４１】
ジョブネットを実行する際、まず、ユーザは、適当なツール等を用いてジョブネット定義テーブル（Ｔ３）の内容を作成する。そして、ユーザは、ジョブネットの実行をJobScheduler（１０１）に指示する。JobScheduler（１０１）は、ジョブネット定義テーブル（Ｔ３）の「JobID」、「NextJob」、「PrevJob」の情報をジョブネット状態テーブル（Ｔ４）にコピーし、コピーした全エントリに対して、ジョブネット状態テーブル（Ｔ４）の「状態フラグ」を「未実行」に設定する。また、JobScheduler（１０１）は、「PrevJob」が設定されていないジョブについて、入力データファイルパスをジョブネット定義テーブル（Ｔ３）から取得し、実際に存在するファイルにアクセスしてサイズを確認し、その値をジョブネット状態テーブル（Ｔ４）の「入力サイズ」に記載する。
【００４２】
そして、JobScheduler（１０１）は、ジョブ履歴テーブル（Ｔ６）に基づき、「入力サイズ」を記載したジョブの出力サイズを推定し、ジョブネット状態テーブル（Ｔ４）「出力サイズ」に記載する。次に、JobScheduler（１０１）は、「PrevJob」に記載されたジョブの「出力サイズ」が記載されたジョブについて、「入力サイズ」を「PrevJob」の出力サイズの合計として算出し、当該ジョブの「入力サイズ」に記載する。そして上記と同様に、当該ジョブの「出力サイズ」を推定し、記載する。この操作をすべてのジョブの「入力サイズ」と「出力サイズ」が決まるまで続ける。
【００４３】
次に、JobScheduler（１０１）は、ジョブネット状態テーブル（Ｔ４）に記載されたすべてのエントリに対して、「入力サイズ」256Mバイトに対してMapタスク数が1、「出力サイズ」512Mバイトに対して1つのReduceタスク数が1になるように「Map数」、「Reduce数」を計算し、ジョブネット状態テーブル（Ｔ４）に記載する。ここまでで、ジョブネット状態テーブルの初期化が完了する（Ｓ１０１）。
【００４４】
次に、Ｓ１０２に進み、投入ジョブの決定処理を行う。この処理の詳細を、図１２に示す。まず、JobScheduler（１０１）は、投入ジョブの候補として、実際にジョブを実行開始できるものを選択する。具体的には、JobScheduler（１０１）は、ジョブネット状態テーブル（Ｔ４）の「PrevJob」にジョブが記載されていないものを選択する。最初にジョブネットの実行を開始した時点では、「J1」「J2」「J3」が候補になる。そして、JobScheduler（１０１）は、各候補について「NextJob」の数を確認する（Ｓ２０１）。
【００４５】
具体的には、「NextJob」が最も多いジョブが1つであるか否かを判定し（Ｓ２０２）、「NextJob」が最も多いジョブが1つであれば（Ｓ２０２；Ｙｅｓ）、それが投入するジョブになる。今の場合、「J2」の「NextJob」の数が2で候補の中で最大であるため、投入候補として選択される。このように、「NextJob」が多いジョブを投入候補として選択する理由は、そういったジョブはそのジョブの実行が終了した後、より多くのジョブが同時実行可能となるからである。MapReduceジョブを複数同時実行することでサーバのリソース空き時間発生を防止するというジョブスケジューリング制御を行うために、同時実行可能なジョブを増やすことは重要である（Ｓ２０３）。
【００４６】
ジョブ「J2」を実行中にサーバのリソース空き時間が発生しそうになった場合、図１１のループＳ１０９から戻ってきて、Ｓ１０２が再実行される。このとき、Ｓ２０１では「J1」「J3」が次の投入ジョブの候補となるが、これらのジョブは「NextJob」の数が同じであるため、「NextJob」が最も多いジョブが1つでないと判定され（Ｓ２０２；Ｎｏ）、Ｓ２０４に進む。
【００４７】
JobScheduler（１０１）は、候補となったジョブの後続ジョブフローの最長経路長をチェックする。最長経路長は、図４に示したDAG図において、ジョブの実行順に進んだ場合に最も経路が長くなるフローの経路長である。「J1」は「J4」を実行してフローが終了するため最長経路長は1、「J3」は「J5」「J6」を実行してフローが終了するため最長経路長は2となる（Ｓ２０４）。JobScheduler（１０１）は、最長経路長が最大である「J3」を投入ジョブとして決定し、そのジョブを選択する（Ｓ２０５）。
【００４８】
このように最長経路長が最大のジョブを投入候補として優先的に選択する理由は、できるだけ長い期間、同時実行可能なジョブが複数存在するようにするためである。連続するジョブから構成されるフローが1つになった場合、同時実行可能なジョブは1つになる。上記制御を行うことで、フローが複数存在する場合に、各フローの進捗が同程度となるため、フローが1つになる期間が短くなる。
【００４９】
尚、最長経路長の計算方法としては、以下のような改良が可能である。例えば、ジョブネット状態テーブル（Ｔ４）に記載されるMap数、Reduce数が共に基準値（例：システム全体での同時実行可能タスク数）以下であれば、そのジョブのみではリソース空き時間を埋めることは不可能であるため、当該ジョブについては経路長0とカウントする。また、別の方法として、ジョブネット状態テーブル（Ｔ４）に記載されるMap数、Reduce数の合計路経路長として算出する計算方法を採ってもよい。このようにして投入すべきジョブが決定したら、図１１に戻って、JobScheduler（１０１）は、JobTracker（１０２）に対して、そのジョブの開始を指示する（Ｓ１０３）。
【００５０】
JobScheduler（１０１）は、JobTracker（１０２）から定期的（例：1秒毎）にジョブの実行状態（実行中ジョブ一覧、各ジョブの実行フェーズ：すなわちMap/Reduceのいずれを実行中なのか、各ジョブの未実行Map/Reduceタスク数）を取得し、結果をジョブネット状態テーブル（Ｔ４）と実行中ジョブ状態テーブル（Ｔ５）とに反映する。
【００５１】
具体的には、ジョブの実行が完了した場合には、ジョブネット状態テーブル（Ｔ４）と実行中ジョブ状態テーブル（Ｔ５）から「JobID」が当該ジョブのジョブＩＤに一致するエントリを削除し、さらにジョブネット状態テーブル（Ｔ４）の「PrevJob」から当該ジョブのジョブＩＤを削除する。また、実行中ジョブが追加された場合には、実行中ジョブ状態テーブル（Ｔ５）に該当するエントリを追加する。さらに、実行中ジョブの実行フェーズに応じてジョブネット状態テーブル（Ｔ４）の「状態フラグ」を変更し、実行中ジョブの未実行Map/Reduceタスク数を実行中ジョブ状態テーブル（Ｔ５）の「残りタスク数」「残りReduce数」に反映する。
【００５２】
さらに、JobScheduler（１０１）は、実行中ジョブ状態テーブル（Ｔ５）の各エントリに対して、以下の手順で「緊急フラグ」を更新する。まず実行中ジョブ状態テーブル（Ｔ５）に記載された各エントリに対応するジョブについて、「JobID」が一致するエントリをジョブネット状態テーブル（Ｔ４）から探し出し、そのジョブの「状態フラグ」を取得する。「状態フラグ」が「Map」であればMap処理実行中であるため、実行中ジョブ状態テーブル（Ｔ５）の「残りMap数」が基準値（例：システム全体での同時実行タスク数の半分）以下であるかどうかを判定し、基準値以下であれば、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」に「CRITICAL」と記載する。
【００５３】
同様に、ジョブネット状態テーブル（Ｔ４）の「状態フラグ」が「Reduce」であれば、実行中ジョブ状態テーブル（Ｔ５）の「残りReduce数」が基準値以下であるかどうかを判定し、基準値以下であれば、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」に「CRITICAL」と記載する。上記基準値はシステム全体での同時実行タスク数に基づき決定するのが効果的と考えられるが、他の方法を採ってもよい。また、MapとReduceに対して異なる値にしてもよい（Ｓ１０４）。
【００５４】
次に、JobScheduler（１０１）は、ジョブの実行状態が変化したかどうかを判定してチェックする（Ｓ１０５）。具体的には、JobScheduler（１０１）は、実行中ジョブ状態テーブル（Ｔ５）に記載されるジョブ一覧が変化した場合（すなわち「JobID」一覧が変化した場合）と、ジョブ一覧は変化していないが「緊急フラグ」が変化したした場合に、ジョブの実行状態が変化したとみなす。JobScheduler（１０１）は、ジョブの実行状態に変化がないと判定した場合（Ｓ１０５；Ｎｏ）はＳ１０４に戻り、ジョブ状態のモニタリングを続ける。
【００５５】
JobScheduler（１０１）は、ジョブ状態が変化したと判定した場合（Ｓ１０５；Ｙｅｓ）はＳ１０６に進み、実行中ジョブの優先度変更を行う。Ｓ１０６の詳細を図１３に示す。
【００５６】
JobScheduler（１０１）は、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」が「CRITICAL」となっているジョブがないかどうかを判定してチェックする（Ｓ３０１）。JobScheduler（１０１）は、「CRITICAL」となっているジョブがあると判定した場合（Ｓ３０１；Ｙｅｓ）、そのジョブの優先度を「優先度高」に変更するように、JobTracker（１０２）に指示する(Ｓ３０２)。
【００５７】
尚、当該ジョブの優先度が既に「優先度高」であった場合、JobTracker（１０２）は何もしないので、Ｓ１０６を実行する度に指示しても動作上の問題はない。「緊急フラグ」が「CRITICAL」であるジョブは実行の多重度が下がる可能性があるジョブであり、上記優先度制御によって、そのようなジョブの優先度を上げることで、複数ジョブが同時実行しているときにジョブが「CRITIAL」である期間を短くすることができる。これによって、複数ジョブの実行の多重度が同時に下がりサーバリソースの空き時間が発生する可能性を少なくすることができる。
【００５８】
次に、JobScheduler（１０１）は、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」が「NORMAL」となっているジョブが複数あるかどうか判定してチェックする（Ｓ３０３）。
JobScheduler（１０１）は、そのようなジョブが複数存在すると判定した場合（Ｓ３０３；Ｙｅｓ）は、各ジョブについて、Ｓ２０４と同様の手順で後続ジョブフローの最長経路長をチェックする。JobScheduler（１０１）は、最長経路長が最大であるジョブの優先度を「優先度中」とし、他のジョブの優先度を「優先度低」とするようJobTracker（１０２）に指示する（Ｓ３０５）。
【００５９】
JobScheduler（１０１）は、「緊急フラグ」が「NORMAL」であるジョブが1つであると判定した場合（Ｓ３０３；Ｎｏ）、JobScheduler（１０１）は、そのジョブの優先度を「優先度中」とするようJobTracker（１０２）に指示する（Ｓ３０６）。ここで各ジョブの最長経路長に基づき各ジョブの優先度を再設定する理由は、ジョブ実行中にジョブの進捗が想定外の変化した場合に、同時実行可能なジョブを複数確保できるようフィードバック制御を行うためである。
【００６０】
次に、JobScheduler（１０１）は、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」が「NORMAL」であるジョブが存在するかどうかを判定してチェックする（Ｓ１０７）。JobScheduler（１０１）は、そのようなジョブが存在すると判定した場合（Ｓ１０７；Ｙｅｓ）は、サーバリソースの空き時間が発生する可能性はないため、Ｓ１０４に戻り、ジョブ状態のモニタリングを続ける。
【００６１】
一方、JobScheduler（１０１）が、実行中ジョブ状態テーブル（Ｔ５）の「緊急フラグ」が「NORMAL」であるジョブが存在しないと判定した場合（Ｓ１０７；Ｎｏ）、サーバリソースの空き時間が発生する可能性がある。JobScheduler（１０１）は、ジョブネット状態テーブル（Ｔ４）の「状態フラグ」が未実行であるジョブが存在するか否かを判定してその確認をする（Ｓ１０８）。このようなジョブがないと判定された場合（Ｓ１０８；Ｎｏ）、既にすべてのジョブの投入が完了しているため、JobScheduler（１０１）は、処理を終える。
【００６２】
JobScheduler（１０１）は、上記条件を満たすジョブが存在すると判定した場合（Ｓ１０８；Ｙｅｓ）、それらジョブのうち「PrevJob」にジョブＩＤが記載されていないジョブを探し、開始可能なジョブの有無を判定する（Ｓ１０９）。そのようなジョブが見つかったらそのジョブは開始可能である。「PrevJob」にジョブＩＤが記載されているジョブは、前段のジョブが終了していないため開始することはできない。JobScheduler（１０１）は、開始可能なジョブがあると判定した場合）Ｓ１０９；Ｙｅｓ）、Ｓ１０２に進みジョブ投入処理を行う。
【００６３】
一方、JobScheduler（１０１）は、開始可能なジョブがないと判定した場合（Ｓ１０９；Ｎｏ）、Ｓ１０４と同様の手順で、ジョブ状態のモニタリングを行う（Ｓ１１０）。但し、JobScheduler（１０１）は、ここではジョブネット状態テーブル（Ｔ４）の「PrevJob」を監視し、開始可能なジョブが出現するまで次のステップには進まない。開始可能なジョブ（すなわちジョブネット状態テーブル（Ｔ４）の「PrevJob」にジョブＩＤが記載されていないジョブ）が出現したらＳ１０２に進みジョブ投入処理を行う。
【００６４】
このように、本システムでは、複数のMapReduceジョブから構成されるジョブネットから、依存関係のない第一のジョブと第二のジョブを抽出し、第一のジョブの未実行Mapタスク数又は未実行Reduceタスク数が予め定められた数より少なくなった場合、第二のジョブを開始する。これによってサーバリソースの空き時間発生を抑止する。従って、MapReduceジョブから構成されるジョブネットの実行時間を短縮することができる。
【００６５】
尚、上記説明では1つのジョブがMapとReduceを1回含むMapReduce処理を例に取り上げたが、Mapのみから構成されるジョブに対しても本発明を適用することができる。また同様に、Reduceを複数回連続実行するジョブ、すなわちReduceの出力結果に対して、さらにグルーピング対象となるkeyを指定して、2回目以降のReduceを実行するジョブについても本発明を適用することができる。
【００６６】
また、本実施例では、上述した各処理を行うプログラムがＣＰＵ（３０１）によってＨＤＤ（３０９）から読み出されてメモリ（３０３）上に展開、実行されることとして説明した。しかし、これらのプログラムを、インストール可能な形式又は実行可能な形式のファイルでＣＤ−ＲＯＭ、フレキシブルディスク（ＦＤ）、ＣＤ−Ｒ、ＤＶＤ等のコンピュータで読み取り可能な記録媒体に記録して提供することももちろん可能である。さらに、上述した各処理を行うプログラムを、インターネット等のネットワークに接続されたコンピュータ上に格納し、ネットワーク経由でダウンロードや配信することにより提供してもよい。
【符号の説明】
【００６７】
１０  マスタノード
２０  スレーブノード
１０１  ＪｏｂＳｃｈｅｄｕｌｅｒ
１０２  ＪｏｂＴｒａｃｋｅｒ
１０３  ＴａｓｋＳｃｈｅｄｕｌｅｒ
Ｔ１  ジョブ管理テーブル
Ｔ２  タスク管理テーブル
Ｔ３  ジョブネット定義テーブル
Ｔ４  ジョブネット状態テーブル
Ｔ５  実行中ジョブ状態テーブル
Ｔ６  ジョブ履歴テーブル。
【図１】
2015090688.tif 000003
【図２】
2015090688.tif 000004
【図３】
2015090688.tif 000005
【図４】
2015090688.tif 000006
【図５】
2015090688.tif 000007
【図６】
2015090688.tif 000008
【図７】
2015090688.tif 000009
【図８】
2015090688.tif 000010
【図９】
2015090688.tif 000011
【図１０】
2015090688.tif 000012
【図１１】
2015090688.tif 000013
【図１２】
2015090688.tif 000014
【図１３】
2015090688.tif 000015
【図１４】
2015090688.tif 000016