(57)【要約】【課題】  データを並列分散処理する際に、データの欠損を許すことによって他より時間がかかっているデータ処理タスクを早期に強制終了し、欠損のあるサンプルデータとして結果を高速に得る。【解決手段】  本発明は、タスク処理サーバと通信を行うことによりタスクの状況を取得して、該タスクの集合であるジョブ毎にタスク統計記憶手段に格納し、タスク終了命令手段が、タスク統計記憶手段から、タスクの統計を得るために最低限必要なサンプル数のタスクの状況を取得して、所定の条件を満たす場合に当該タスクを強制的に終了させ、強制終了したことを該タスク統計記憶手段に反映させる。【選択図】  図２
2014010500.tif 000002
【特許請求の範囲】
【請求項１】
並列分散データ処理において、タスクを実行するタスク処理サーバに割り当てる入力ブロックや、タスクの情報を管理し、スケジューリングやタスクの強制終了を行うデータ処理装置であって、
前記タスク処理サーバと通信を行うことによりタスクの状況を取得して、該タスクの集合であるジョブ毎にタスク統計記憶手段に格納するタスク監視手段と、
前記タスク統計記憶手段から、タスクの統計を得るために最低限必要なサンプル数のタスクの状況を取得して、所定の条件を満たす場合に当該タスクを強制的に終了させ、強制終了したことを該タスク統計記憶手段に反映させるタスク終了命令手段と、
を有することを特徴とするデータ処理装置。
【請求項２】
前記タスク終了命令手段は、
タスクの実行時間が所定の閾値より大きい場合には、該タスクを強制的に終了させる手段を含む
請求項1記載のデータ処理装置。
【請求項３】
前記タスク終了命令手段は、
前記所定の条件として、
タスクのタスクＩＤに対してハッシュ関数を適用し、その値が所定の数値以下であった場合に該タスクを強制的に終了させる、
または、
タスクに割り当てられた入力データのサイズが一定の値よりも大きいまたは、小さい場合に該タスクを強制的に終了させる、
または、
タスク割り当て時に乱数を生成し、該乱数の値が所定の範囲である場合に、該当するタスクを強制的に終了させる、
または、
タスクを実行しているタスク処理サーバの負荷が所定の閾値より高い場合に、該タスクを強制的に終了させる、
または、
タスクのジョブ開始時刻から所定の時間を経過した時点で全てのタスクを強制的に終了させる、
のいずれか１つまたは、複数組み合わせる
請求項1記載のデータ処理装置。
【請求項４】
並列分散データ処理において、タスクを実行するタスク処理サーバに割り当てる入力ブロックや、タスクの情報を管理し、スケジューリングやタスクの強制終了を行うデータ処理方法であって、
タスクの状況を格納するタスク統計記憶手段と、
タスク監視手段と、
タスク終了命令手段と、を有する装置において、
前記タスク監視手段が、前記タスク処理サーバと通信を行うことによりタスクの状況を取得して、該タスクの集合であるジョブ毎に前記タスク統計記憶手段に格納するタスク監視ステップと、
前記タスク終了命令手段が、前記タスク統計記憶手段から、タスクの統計を得るために最低限必要なサンプル数のタスクの状況を取得して、所定の条件を満たす場合に当該タスクを強制的に終了させ、強制終了したことを該タスク統計記憶手段に反映させるタスク終了命令ステップと、
を行うことを特徴とするデータ処理方法。
【請求項５】
前記タスク終了命令ステップにおいて、
タスクの実行時間が所定の閾値より大きい場合には、該タスクを強制的に終了させる
請求項４記載のデータ処理方法。
【請求項６】
前記タスク終了命令ステップにおいて、
前記所定の条件として、
タスクのタスクＩＤに対してハッシュ関数を適用し、その値が所定の数値以下であった場合に該タスクを強制的に終了させる、
または、
タスクに割り当てられた入力データのサイズが一定の値よりも大きいまたは、小さい場合に該タスクを強制的に終了させる、
または、
タスク割り当て時に乱数を生成し、該乱数の値が所定の範囲である場合に、該当するタスクを強制的に終了させる、
または、
タスクを実行しているタスク処理サーバの負荷が所定の閾値より高い場合に、該タスクを強制的に終了させる、
または、
タスクのジョブ開始時刻から所定の時間を経過した時点で全てのタスクを強制的に終了させる、
のいずれか１つまたは、複数組み合わせる
請求項４記載のデータ処理方法。
【請求項７】
コンピュータを、
請求項1乃至３のいずれか1項に記載のデータ処理装置の各手段として機能させるためのデータ処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、データ処理装置及び方法及びプログラムに係り、特に、並列分散データ処理の進行状況に応じて適応的にサンプリングを適用し、処理結果にかかる時間を短縮するデータ処理装置及び方法及びプログラムに関する。
【背景技術】
【０００２】
近年証券取引、ソーシャルメディア、生命科学、医療データ、宇宙科学等、様々な領域において、より精度が高い予測や有益な法則の発見を目的として、膨大なデータの蓄積・分析が行われている。そのような膨大なデータを効率的に処理するための技術として大規模な計算機システムによる並列分散処理が有効である。
【０００３】
データの並列分散処理では、入力となるデータを小さな単位(便宜上、これを入力ブロックと呼ぶことにする)に分割し、入力ブロックを複数のサーバで並列分散処理することによって処理効率を高めることができる。
【０００４】
各々のサーバは複数のブロックを順次処理していく。サーバ内では複数のブロックを並列処理することも可能である。
【０００５】
入力ブロックを各々のサーバで処理した結果得た中間的な出力データ(中間出力ブロック)は最終的な処理結果を得るために集約する必要がある。サーバがブロック、あるいは中間出力ブロックを処理する単位を便宜上タスクと呼ぶことにする。
【０００６】
このようなタスクによって並列分散処理を行う場合、あるタスクの処理が遅れたがために全体の処理時間が遅延するという問題がある。例えば前段の入力ブロックの処理を行うタスクの処理が遅れると、後段の中間出力ブロックのタスクも遅れる。あるいは同じサーバ内であっても、ひとつ前のタスクが遅れれば、次にスケジュールされたタスクの実行が遅れる。
【０００７】
こうした問題を解決するために、特に時間がかかるタスクを同じサーバにて再スタートするか、別のサーバで同じタスクを実行させ、早く完了したタスクを採用するか、あるいは両者を併用する方式がある（例えば、非特許文献１参照）。同方式ではタスクをコピーするサーバの選択においてネットワークのリソース利用状況を勘案する。さらに同方式では時間がかかると見込まれる大きなサイズのタスクを優先的に実行する戦略をとる。
【先行技術文献】
【非特許文献】
【０００８】
【非特許文献１】Reining in the Outliers in Map-Reduce Clusters using Mantri (OSDI 2010)  http://www.usenix.org/event/osdi10/tech/
【発明の概要】
【発明が解決しようとする課題】
【０００９】
上述の従来の方式は所与のデータ処理を行うタスクを最適に実行・配置するための方式であり、すべてのデータを処理することが前提となっている。しかしながら、大規模なデータ処理においてある程度のデータ欠損は許容しつつも極力データ処理にかかる時間を短縮するという用途には適さない。
【００１０】
本発明は上記の課題を鑑みてなされたものであり、データを並列分散処理する際に、データの欠損を許すことによって他より時間がかかっているデータ処理タスクを早期に強制終了し、欠損のあるサンプルデータとして結果を高速に得る方法を実現することが可能なデータ処理装置及び方法及びプログラムを提供することを目的とする。
【課題を解決するための手段】
【００１１】
上記の課題を解決するため、本発明（請求項１）は、並列分散データ処理において、タスクを実行するタスク処理サーバに割り当てる入力ブロックや、タスクの情報を管理し、スケジューリングやタスクの強制終了を行うデータ処理装置であって、
前記タスク処理サーバと通信を行うことによりタスクの状況を取得して、該タスクの集合であるジョブ毎にタスク統計記憶手段に格納するタスク監視手段と、
前記タスク統計記憶手段から、タスクの統計を得るために最低限必要なサンプル数のタスクの状況を取得して、所定の条件を満たす場合に当該タスクを強制的に終了させ、強制終了したことを該タスク統計記憶手段に反映させるタスク終了命令手段と、を有する。
【００１２】
また、本発明（請求項２）は、前記タスク終了命令手段において、
タスクの実行時間が所定の閾値より大きい場合には、該タスクを強制的に終了させる手段を含む。
【００１３】
また、本発明（請求項３）は、前記タスク終了命令ステップにおいて、
前記所定の条件として、
タスクのタスクＩＤに対してハッシュ関数を適用し、その値が所定の数値以下であった場合に該タスクを強制的に終了させる、
または、
タスクに割り当てられた入力データのサイズが一定の値よりも大きいまたは、小さい場合に該タスクを強制的に終了させる、
または、
タスク割り当て時に乱数を生成し、該乱数の値が所定の範囲である場合に、該当するタスクを強制的に終了させる、
または、
タスクを実行しているタスク処理サーバの負荷が所定の閾値より高い場合に、該タスクを強制的に終了させる、
または、
タスクのジョブ開始時刻から所定の時間を経過した時点で全てのタスクを強制的に終了させる、
のいずれか１つまたは、複数組み合わせる。
【発明の効果】
【００１４】
上記のように本発明によれば、タスクの統計を得るために最低限必要なサンプル数のタスクの状況に基づいて、データの欠損を許すことによって他より時間がかかっているデータ処理タスクを早期に終了させ、サンプルされたデータとして結果を高速に得ることが可能となる。
【図面の簡単な説明】
【００１５】
【図１】本発明の一実施の形態におけるシステム構成図である。
【図２】本発明の一実施の形態におけるタスク管理サーバの構成図である。
【図３】本発明の一実施の形態におけるタスク統計ＤＢの構成例である。
【図４】本発明の一実施の形態におけるタスク終了命令部の処理のフローチャートである。
【図５】本発明の一実施の形態における実行時間の閾値の算出のフローチャートである。
【図６】本発明の一実施の形態におけるタスクを強制終了する条件を示す図（その１）である。
【図７】本発明の一実施の形態におけるタスクを強制終了する条件を示す図（その２）である。
【図８】本発明の一実施の形態におけるタスクを強制終了する条件を示す図（その３）である。
【図９】本発明の一実施の形態におけるタスクを強制終了する条件を示す図（その４）である。
【図１０】本発明の一実施の形態におけるタスクを強制終了する条件を示す図（その５）である。
【発明を実施するための形態】
【００１６】
以下、図面と共に本発明の実施の形態を説明する。
【００１７】
本発明は、複数のデータブロックを並列で処理し、この処理結果を中間出力として集計処理して最終結果を得るようなデータ処理を考えた場合に、ある程度のデータの欠損を許すことによって、他より時間がかかっているデータ処理のタスクを早期に終了し、サンプリングされたデータから最終結果を高速に得る方法である。
【００１８】
図１は、本発明の一実施の形態におけるシステム構成を示す。
【００１９】
同図に示すように、データを小さな単位に分割した入力ブロックを処理し、中間出力ブロックを出力するタスクおよび中間出力を処理し、最終出力データを出力するタスクを実行するタスク処理サーバ２０１，２０２と、タスク処理サーバ２０に割り当てる入力ブロックや、タスクの情報を管理し、スケジューリングやタスクの強制終了を行うタスク管理サーバ１０からなる。
【００２０】
図２は、本発明の一実施の形態におけるタスク管理サーバの構成を示す。
【００２１】
同図に示すようにタスク管理サーバ１０は、タスク監視部１１、タスク終了命令部１２、タスク統計ＤＢ１３，タスク終了閾値計算部１４を有する。
【００２２】
タスク管理サーバ１０では各々のタスク処理サーバ２０と通信を行い、タスクの状況をタスク監視部１１で監視し、タスク統計DB１３に管理する。タスク統計DB１３の構成例を図３に示すタスク統計DB１３ではタスクの集合であるジョブ毎にテーブルを有し、ジョブを構成するタスクの情報をテーブルで管理する。テーブルを構成するカラムは、各々のタスクに割り当てられるユニークなタスクID、タスクの開始時刻、完了時刻、現在まで、あるいは完了するまでの実行時間、タスクの状態、などから成る。タスク状態のカラムはタスクの状態、すなわち「実行中」、「自然完了」、「強制完了」といった情報を格納するために用いる。
【００２３】
図４は、本発明の一実施の形態におけるタスク終了命令部の処理のフローチャートである。
【００２４】
タスク管理サーバ１０のタスク終了命令部１２では、タスク統計DB１３の情報を元に、一定期間(例えば1秒)が到来する毎に現在実行中のタスクを対象とし、当該タスクの開始から現在までの実行時間tを得る（ステップ１０１）。実行時間が閾値を超えていた場合に該当タスクを強制的に終了する（ステップ１０４）。またその情報をタスク統計DB１３に反映させる（ステップ１０５）。この際、タスクの統計を得るために最低限必要なサンプル数をKとし、K個以上のタスクサンプルが集まった場合にタスク統計を計算する（ステップ１０２）。該当となるタスクの現在までの実行時間 t と後述する方法で計算する閾値τを比較し（ステップ１０３）、t > τであったらその当該タスクを強制的に終了する命令をタスク処理サーバ２０の該当タスクに送信し、該当タスクはその命令を受信した時点で処理を強制終了する（ステップ１０４）。タスク処理サーバ２０は、強制終了が完了したらタスク管理サーバ１０に通知し、タスク管理サーバ１０はタスク管理DBを更新する（ステップ１０５）。
【００２５】
前記の閾値τの計算は任意の方法で行うことが可能であるが、その一例を図５のフローチャートで示す。累積実行完了タスク数がＫ個以上である場合に、ここではスライディングウィンドウにより、直近K個の自然完了タスクのタスク実行時間の移動平均mおよび移動標準偏差σを計算し、閾値を m+nσとする。n は任意に設定が可能な数値である。ここで「自然完了タスク」とは強制的に完了したタスクではなく、処理の終了に伴い完了したタスクを示す。
【００２６】
上記では、タスク完了時間に基づいてタスクを強制終了する方法を示したが、以下では、タスク完了時間によらず、タスクを強制終了する方法を示す。
【００２７】
図６〜図１０に、タスクの強制終了方法の条件を示す。タスク管理サーバ１０は、いずれもその条件が決定した時点で強制終了を実施する。
【００２８】
図６では、タスクIDに対してハッシュ関数を適用し、その値がある数値以下であったとき、例えばハッシュ関数により0から2^128までの任意の整数値が出力される場合、その値が 2^127 以下であったときに該当タスクを強制終了する。この場合、約半数のタスクは強制終了される。
【００２９】
図７では、タスクに割り当てられた入力データのサイズが一定値よりも大きい、あるいは小さい場合にそのタスクを強制終了する。
【００３０】
図８では、タスク割り当て時に乱数を生成し、その値がある範囲である際に該当タスクを強制終了する（タスクを実施しない）。これはランダムサンプリングに相当する。
【００３１】
図９では、該当タスクを実行しているサーバの負荷に応じ、負荷が高い場合にその該当タスクを強制終了してしまう方法である。負荷の高低を決定する閾値は任意の値に定めることができる。
【００３２】
図１０では、全体の実行時間に応じてタスクを終了する方法であり、ジョブ開始時刻
からある一定の時間を経過した時点ですべてのタスクを強制終了する。
【００３３】
図５〜図１０で述べたタスク強制終了方法はそれぞれ単独で使ってもよいし、組み
合わせて使うことも可能である。
【００３４】
なお、図２に示すタスク管理サーバ１０の各構成要素の処理をプログラムとして構築し、タスク管理サーバとして利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【００３５】
本発明は、上記の実施の形態に限定されることなく、特許請求の範囲内において、種々変更・応用可能である。
【符号の説明】
【００３６】
１０  タスク管理サーバ
１１  タスク監視部
１２  タスク終了命令部
１３  タスク統計ＤＢ
１４  タスク終了閾値計算部
２０  タスク処理サーバ
【図１】
2014010500.tif 000003
【図２】
2014010500.tif 000004
【図３】
2014010500.tif 000005
【図４】
2014010500.tif 000006
【図５】
2014010500.tif 000007
【図６】
2014010500.tif 000008
【図７】
2014010500.tif 000009
【図８】
2014010500.tif 000010
【図９】
2014010500.tif 000011
【図１０】
2014010500.tif 000012