(57)【要約】【課題】  ジョブ内に複数回実行されるタスクが含まれている場合、ジョブを高速に処理すると共に、過去に行ったことがある処理を高速化する。【解決手段】  本発明は、資源管理ノードが、入力データを保持している計算ノードに該入力データのハッシュ値を要求し、実行ファイルと共に取得し、入力データを保持している計算ノードにタスクを割り当てる。計算ノードが割り当てられたタスクを実行し、その結果を実行ファイル、入力データと関連付けて計算結果一時記憶手段に格納し、資源管理ノードが、実行結果として、実行ファイル、入力データを取得して、該計算ノードと関連付けて計算結果記憶手段に格納する。資源管理ノードが、計算ノードにタスクを割り当てる際に、入力データのハッシュ値と実行ファイルをキーとして、計算結果記憶手段を検索し、該計算ノードによる該タスクの計算結果が存在している場合には、当該タスクの処理をスキップする。【選択図】  図６
2012242975.tif 000002
【特許請求の範囲】
【請求項１】
分散並列実行フレームワークにおいて実行するプログラムのフローが循環有向グラフとして表現されている場合に、タスクを割り当てる資源管理ノード及びタスクを割り当てられる計算ノードから構成される並列キャッシュシステムであって、
前記資源管理ノードは、
タスクの入力データを有する計算ノードに対して入力データのハッシュ値の計算を要求し、実行ファイルと共に取得するハッシュ値取得手段と、
タスクを実行した計算ノードから取得した入力データのハッシュ値、実行ファイル、入力ファイル取得し、該計算ノードと関連付けて計算結果記憶手段に格納するタスク管理手段と、
前記入力データのハッシュ値と前記実行ファイルを検索キーとして、前記計算結果記憶手段を検索し、該キーに該当する計算ノードのデータがある場合には、計算をスキップし、ない場合にはタスクを割り当てる計算ノードを決定するタスク割当手段と、
を有し、
前記計算ノードは、
前記資源管理ノードから要求された前記入力データのハッシュ値を求め、該資源管理ノードに返却するハッシュ値計算手段と、
前記資源管理ノードにより割り当てられたタスクを実行し、その結果を実行ファイル、入力データと関連付けて計算結果一時記憶手段に格納すると共に、該資源管理ノードに送信するタスク実行手段と
を有することを特徴とする並列キャッシュシステム。
【請求項２】
前記資源管理ノードは、
ジョブが終了すると、前記検索キーに基づいて前記計算結果記憶手段から計算結果を保持している計算ノードを抽出し、該計算ノードと同じ物理計算機上で動作しているデータノードに対して、該計算結果を送信し、格納する手段を更に有する
請求項１記載の並列キャッシュシステム。
【請求項３】
分散並列実行フレームワークにおいて実行するプログラムのフローが循環有向グラフとして表現されている場合に、タスクを割り当てる資源管理ノード及びタスクを割り当てられる計算ノードから構成される並列キャッシュ方法であって、
前記資源管理ノードのハッシュ値取得手段が、入力データを保持している計算ノードに該入力データのハッシュ値を要求し、実行ファイルと共に取得するハッシュ値取得ステップと、
前記資源管理ノードのタスク割当手段が、入力データを保持している計算ノードにタスクを割り当てるタスク割り当てステップと、
前記計算ノードのタスク実行手段が、前記資源管理ノードから割り当てられたタスクを実行し、その結果を実行ファイル、入力データと関連付けて計算結果一時記憶手段に格納するタスク実行ステップと、
前記資源管理ノードの実行結果管理手段が、前記計算ノードからタスクの実行結果として、実行ファイル、入力データを取得して、該計算ノードと関連付けて計算結果記憶手段に格納する実行結果管理ステップと、
を行い、
前記タスク割当ステップにおいて、
前記計算ノードにタスクを割り当てる際に、前記入力データのハッシュ値と前記実行ファイルをキーとして、前記計算結果記憶手段を検索し、該計算ノードによる該タスクの計算結果が存在している場合には、当該タスクの処理をスキップする
ことを特徴とする並列キャッシュ方法。
【請求項４】
前記資源管理ノードにおいて、
ジョブが終了すると、前記キーに基づいて前記計算結果記憶手段から計算結果を保持している計算ノードを抽出し、該計算ノードと同じ物理計算機上で動作しているデータノードに対して、該計算結果を送信し格納する
請求項３記載の並列キャッシュ方法。
【請求項５】
分散並列実行フレームワークにおいて実行するプログラムのフローが循環有向グラフとして表現されている場合に、タスクを計算ノードに割り当てる資源管理ノードであって、
タスクの入力データを有する計算ノードから取得した入力データのハッシュ値、実行ファイル、入力データを取得し、該計算ノードと関連付けて前記計算結果記憶手段に格納するタスク管理手段と、
前記計算ノードに対して入力データのハッシュ値の計算を要求し、ハッシュ値と実行ファイルを取得するハッシュ値取得手段と、
前記入力データのハッシュ値と前記実行ファイルを検索キーとして、前記計算結果記憶手段を検索し、該キーに該当するデータがある場合には、計算をスキップし、ない場合にはタスクを割り当てる計算ノードを決定するタスク割当手段と、
を有することを特徴とする資源管理ノード。
【請求項６】
コンピュータを、
請求項５記載の資源管理ノードの各手段として機能させるための並列キャッシュプログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、分散並列処理キャッシュ装置及び方法及び資源管理ノード及びプログラムに係り、特に、分散処理フレームワークにおける分散並列処理を行う際に、入力データを有する計算ノードにタスクを割り当てるためのキャッシュ装置及び方法及び資源管理ノード及びプログラムに関する。
【背景技術】
【０００２】
分散処理フレームワークにおいて、PCクラスタ上にファイルを分割して保存する分散記憶装置を前提としたデータ解析基盤ソフトウェアがある。これは、図１に示すように、有向グラフの形式で動作フローを記述することで、互いに依存関係のない入力データに対して並列計算を行うプログラムを実行基盤とするものである。図１において、辺がデータの流れであり、頂点が行う処理内容を意味する。頂点上で実行される処理をタスクという。実行時は、分散ファイルシステムにアクセスし、入力ファイルを処理単位に分割する初期化処理を行い、分割したファイルが保存されているデータノードに近い計算ノードに、処理内容が記述されたプログラムを配置し、計算が完了したら結果を分散ファイルシステムに書き込むものである。なお、「計算ノード」とは、実際にタスクを実行する計算機を指す。
【０００３】
このような環境において、１つ前の頂点の処理結果を分割し、入力データとして処理する方法として、MapReduceを拡張し、ループ処理に特化させたシステムがある（例えば、非特許文献１、２参照）、Map処理、Reduceで記述されたプログラムを分散並列処理するシステム（例えば、特許文献1参照 ）や、非循環有向グラフ処理で記述されたプログラムを分散並列処理するシステム（特許文献２参照）がある。
【０００４】
以下に従来の技術における分散並列処理キャッシュ処理を説明する。
【０００５】
図２は、従来技術の概略フローチャートである。
【０００６】
資源管理ノードがキャッシュを考慮しながら計算ノードにタスクを割り当て（ステップ２１０）、計算ノードが割り当てられたタスクを実行する（ステップ２２０）。タスクの実行結果を計算ノードに一時ファイルとして保存し（ステップ２３０）、ジョブが終了していない場合はステップ２１０に戻る。ジョブが終了したら計算ノードの一時ファイルを削除する。なお、ここで、「ジョブ」とは、動作フローを記述したプログラムが行う一連の処理を指し、各頂点で実行される処理と、データの流れを表現するデータフローで構成される。
【０００７】
上記の計算ノードにタスクを割り当てるステップ２１０の処理を図３に示す。
【０００８】
特許文献１の技術に基づいて資源管理ノードが次に実行すべきタスクを決定する（ステップ４１０）。次に、資源管理ノードが次に実行ジョブ中における再利用するデータの宣言を確認する（ステップ４２０）。ステップ４１０で選択したタスクの入力は再利用可能かを判定し、再利用可能であり、１回目の実行である場合は、計算結果を保持するノードで処理が終了したと見做し、計算をスキップし（ステップ４４０）、ステップ４１０の処理に戻る。一方、再利用ができない、または、２回目以降の実行である場合は、特許文献１の技術に基づいて実行する計算ノードを決定する（ステップ４６０）。
【０００９】
上記のステップ２３０では、図４に示すように、完了したタスクの計算結果を計算ノードの外部記憶装置（ローカルディスク）に保存する（ステップ６１０）。
【００１０】
最後に、ステップ２４０では、図５に示すように、完了したタスクの計算結果を計算ノードの外部記憶装置（ローカルディスク）に保存する（ステップ６１０）。
【先行技術文献】
【特許文献】
【００１１】
【特許文献１】United States Patent 7,650,331, System and method for efficient large-scale data processing (Map Reduce) Google, January 19, 2010.
【特許文献２】USPTO Applicaton #20080082644, Distributed Parallel Computing（Dryad)，  Microsoft Corporation，September 29, 2006.
【非特許文献】
【００１２】
【非特許文献１】HaLoop: Efficient Iterative Data Processing on Large Clusters, Yingyi Bu, Bill Howe, Magdalena Balazinska, Michael D. Ernst. In VLDB'10: The 36the International Conference on Very Large Data Bases, Singapore, 24-30, September, 2010.
【非特許文献２】Twister: A Runtime for Iterative MapReduce Jaliya Ekanayake, Hui Li, Bingjing Zhang, Thilina Gunarathne, Seung-Hee Bae, Judy Qiu, Geoffrey Fox, Twister: A Runtime for Iterative MapReduce. The first International Workshop on MapReduce and its Applications (MAPREDUCE'10) - HPDC2010.
【発明の概要】
【発明が解決しようとする課題】
【００１３】
しかしながら、上記特許文献１，２、非特許文献１，２のシステムのよい箇所を組み合わせたシステムを構築した場合でも、以下のような問題がある。
【００１４】
（１）  ジョブ内で過去に行われた計算結果を再利用するためには、再利用するデータをユーザが宣言する必要がある。
【００１５】
（２）  ジョブが複数回実行される場合、過去に行ったことがあるタスクであっても再度同じ計算、転送を行う必要がある。詳しくは、
・過去に行ったことがある計算が実行中のジョブに含まれていたとしても、複数ジョブを実行する際には再度同じ計算を行う必要がある。
【００１６】
・過去に行ったことがある計算が実行中のジョブに含まれていたとしても、計算ノード間における転送が発生してしまう。
【００１７】
本発明は、上記の点に鑑みなされたもので、ジョブ内に複数回実行されるタスクが含まれている場合、ジョブを高速に処理すると共に、過去に行ったことがある処理を高速化することが可能な分散並列処理キャッシュ装置及び方法及びプログラムを提供することを目的とする。
【課題を解決するための手段】
【００１８】
上記の課題を解決するため、本発明（請求項１）は、分散並列実行フレームワークにおいて実行するプログラムのフローが循環有向グラフとして表現されている場合に、タスクを割り当てる資源管理ノード及びタスクを割り当てられる計算ノードから構成される並列キャッシュシステムであって、
前記資源管理ノードは、
タスクの入力データを有する計算ノードに対して入力データのハッシュ値の計算を要求し、実行ファイルと共に取得するハッシュ値取得手段と、
タスクを実行した計算ノードから取得した入力データのハッシュ値、実行ファイル、入力ファイル取得し、該計算ノードと関連付けて計算結果記憶手段に格納するタスク管理手段と、
前記入力データのハッシュ値と前記実行ファイルを検索キーとして、前記計算結果記憶手段を検索し、該キーに該当する計算ノードのデータがある場合には、計算をスキップし、ない場合にはタスクを割り当てる計算ノードを決定するタスク割当手段と、
を有し、
前記計算ノードは、
前記資源管理ノードから要求された前記入力データのハッシュ値を求め、該資源管理ノードに返却するハッシュ値計算手段と、
前記資源管理ノードにより割り当てられたタスクを実行し、その結果を実行ファイル、入力データと関連付けて計算結果一時記憶手段に格納すると共に、該資源管理ノードに送信するタスク実行手段と、を有する。
【００１９】
また、本発明（請求項２）は、前記資源管理ノードにおいて、
ジョブが終了すると、前記検索キーに基づいて前記計算結果記憶手段から計算結果を保持している計算ノードを抽出し、該計算ノードと同じ物理計算機上で動作しているデータノードに対して、該計算結果を送信し、格納する手段を更に有する。
【発明の効果】
【００２０】
上記のように、本発明によれば、入力データを保持する計算ノードにおいて実行されたタスクの実行結果を資源管理ノードで記録しておくことにより、
（１）ジョブ内に複数回実行されるタスクが含まれている場合、ジョブの高速化を自動的に行うことができる。
【００２１】
（２）ジョブが複数回実行される場合において、過去に行ったことのある処理を高速化できる。詳しくは、過去に行ったことのある計算が実行中のジョブに含まれていた場合、その結果を入力ブロック単位で再利用し計算をスキップすることができる。
【図面の簡単な説明】
【００２２】
【図１】有向グラフ形式の動作フローである。
【図２】従来技術の概略フローチャートである。
【図３】図２のＳ２１０を詳細化したフローチャートである。
【図４】図２のＳ２３０を詳細化したフローチャートである。
【図５】図２のＳ２４０を詳細化したフローチャートである。
【図６】本発明の一実施の形態におけるシステム構成図である。
【図７】本発明の一実施の形態における全体動作のフローチャートである。
【図８】本発明の一実施の形態における図７のＳ３１０を詳細化したフローチャートである。
【図９】本発明の一実施の形態における図７のＳ３３０を詳細化したフローチャートである。
【図１０】本発明の一実施の形態における図７のＳ３４０を詳細化したフローチャートである。
【図１１】本発明の一実施の形態における全体動作のフローチャートである。
【発明を実施するための形態】
【００２３】
以下図面と共に、本発明の実施の形態を説明する。
【００２４】
図６は、本発明の一実施の形態におけるシステム構成を示す。
【００２５】
同図に示すシステムは、資源管理ノード１０と計算ノード２１、分散ファイルシステムのノードであり、データを保存するデータノード２２を有する複数の物理計算機２０から構成される。
【００２６】
資源管理ノード１０は、どの計算ノードにどのタスクを割り当てるかを管理する計算機であり、タスク管理部１１、検索部１２、ノード管理ＤＢ１３を有し、外部記憶として計算結果記憶部１（ローカルディスク）が接続されている。ノード管理ＤＢ１３は、物理計算機毎に、計算ノード、データノード、及び、タスク用の入力データを関連付けて保持している。また、計算結果記憶部１は、計算ノード毎にハッシュ値及び実行ファイルが格納される。以下では外部記憶装置として説明するが、資源管理ノード内のディスク装置やメモリとして構築されていてもよい。
【００２７】
物理計算機２０の計算ノード２１は、ハッシュ計算部２１１を有し、計算結果一時記憶部２を有するローカルディスクが接続されている。データノード２２はデータ記憶部２２１を有する。
【００２８】
上記の構成のシステムの処理について説明する。
【００２９】
図７は、本発明の一実施の形態における全体動作のフローチャートである。
【００３０】
ステップ３１０）  資源管理ノード１０のタスク管理部１１は、キャッシュを考慮しながら計算ノード２１にタスクを割り当てる。詳細については図８で後述する。
【００３１】
ステップ３２０）  タスクを割り当てられた計算ノード２１は、資源管理ノード１０の指示によりタスクを実行する。
【００３２】
ステップ３３０）  計算ノード２１は、ステップ３２０の結果を実行ファイル、入力データと関連付けて一時ファイルとして計算結果一時記憶部２（ローカルディスク）に保存すると共に、資源管理ノード１０に計算結果を送信し、ジョブが終了すればステップ３４０に移行し、終了していなければステップ３１０に戻る。詳細については図９にて後述する。
【００３３】
ステップ３４０）  ジョブが終了したら、資源管理ノード１０のタスク管理部１１は、計算結果を再利用できるように計算結果記憶部１に保存し、永続化する。詳細は図１０にて後述する。
【００３４】
次に、上記のステップ３１０について説明する。
【００３５】
図８は、本発明の一実施の形態における図７のＳ３１０の詳細化のフローチャートである。
【００３６】
ステップ５１０）  資源管理ノード１０のタスク管理部１１は、特許文献１の技術に基づいて、次に実行すべきタスクを決定する。
【００３７】
ステップ５２０）  資源管理ノード１０のタスク管理部１１は、ステップ５１０で選択したタスクの入力データを保持している計算ノード２１をノード管理ＤＢ１３から検索し、当該計算ノード２１に対して入力データのハッシュ値を計算するように要求する。これにより計算ノード２１は、入力データのハッシュ値を計算してハッシュ値と実行ファイルを資源管理ノード１０に送信する。
【００３８】
ステップ５３０）  資源管理ノード１０のノード検索部１２は、ステップ５２０において、計算ノード２１に要求して取得した、入力データのハッシュ値と実行ファイルをキーとして、計算結果記憶部１から過去の計算結果の有無と当該計算結果の保存場所を検索する。計算結果記憶部１に入力データのハッシュ値に該当する計算結果が存在する場合には、すでに計算が実行されているものとしてステップ５４０に移行し、存在しない場合はまだ計算が実行されていないものとしてステップ５５０に移行する。
【００３９】
ステップ５４０）  タスク管理部１１は、計算結果を保持する計算ノードで処理が終了したと見做し、計算をスキップし、ステップ５１０に移行する。
【００４０】
ステップ５５０）  タスク管理部１１は、特許文献１の技術に基づいて、タスクを実行する計算ノードを決定する。
【００４１】
次に、図７のタスク終了時の処理（ステップ３３０）について説明する。
【００４２】
図９は、本発明の一実施の形態における図７のＳ３３０の詳細化のフローチャートである。
【００４３】
ステップ７１０）  計算ノード２１は、実行完了したタスクの計算結果（入力データのハッシュ値、実行ファイル、実行結果）を自ノードに接続された計算結果一時記憶部２に保存すると共に、資源管理ノード１０に送信する。
【００４４】
ステップ７２０）  資源管理ノード１０のタスク管理部１１は、計算ノードからキー（入力データのハッシュ値と実行ファイル）と実行結果を取得し、計算ノードと関連付けて計算結果記憶部１に格納する。
【００４５】
次に、図７のジョブ終了時の処理（ステップ３４０）について説明する。
【００４６】
図１０は、本発明の一実施の形態における図７のＳ３４０のフローチャートである。
【００４７】
ステップ９１０）  ジョブが終了すると、資源管理ノード１０のノード検索部１２は、検索キー（入力データのハッシュ値と実行ファイル）に基づいて計算結果記憶部１を検索し、当該キーに対応する計算結果を保持している計算ノードを検索する。
【００４８】
ステップ９２０）  ノード検索部１２は、ステップ９１０で検索された計算ノードと同じ物理計算機上で動作しているデータノードをノード管理ＤＢ１３から検索する。タスク管理部１１は、当該データノード２２に計算結果を送信する。これにより物理計算機２０のデータノード２２は当該計算結果をデータ記憶部２２１に格納する。
【００４９】
図１１は、本発明の一実施の形態における全体動作のシーケンスチャートである。同図におけるステップ番号は前述のフローチャートのステップ番号に対応する。
【００５０】
上記のような処理を行うことにより、入力データが時系列に従って増えていくような処理を含む計算を高速に行うことができる。
【００５１】
これにより、
（１）検索エンジンの更新が高速に行えるようになる。
【００５２】
（２）ユーザへのレコメンデーションが高速に更新できるようになる。
【００５３】
（３）ログ解析を高速に行うことができる。
【００５４】
なお、上記の資源管理ノード及び計算ノードの動作をプログラムとして構築し、資源管理ノード、計算ノードとして利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【００５５】
本発明は、上記の実施の形態に限定されることなく特許請求の範囲内において種々変更・応用が可能である。
【符号の説明】
【００５６】
１  計算結果記憶部
２  計算結果一時記憶部
１０  資源管理ノード
１１  タスク管理部
１２  検索部
１３  ノード管理ＤＢ
２０  物理計算機
２１  計算ノード
２２  データノード
２１１  ハッシュ計算部
２２１  データ記憶部
【図１】
2012242975.tif 000003
【図２】
2012242975.tif 000004
【図３】
2012242975.tif 000005
【図４】
2012242975.tif 000006
【図５】
2012242975.tif 000007
【図６】
2012242975.tif 000008
【図７】
2012242975.tif 000009
【図８】
2012242975.tif 000010
【図９】
2012242975.tif 000011
【図１０】
2012242975.tif 000012
【図１１】
2012242975.tif 000013