(57)【要約】  結果整合性ストレージシステムに対する分散バッチ処理を効率的に行う。  制御ノード（１００）は、実行制御部（１１０）、及び、再実行制御部（１４０）を含む。実行制御部（１１０）は、分散データストア（３００）に格納された、キーに対するデータの読み出しを含む所定の処理を、処理ノード（２００）に実行させる。再実行制御部（１４０）は、所定の処理を処理ノード（２００）に実行させた後で、当該所定の処理実行時に分散データストア（３００）に格納されていた当該キーに対するデータに不整合が生じていた可能性の有無を、当該データの代表値に基づいて判定する。そして、再実行制御部（１４０）は、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、所定の処理を処理ノード（２００）に再実行させる。
2016027455.tif 000002
【特許請求の範囲】
【請求項１】
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させる、実行制御手段と、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、再実行制御手段と、
を備えた、情報処理装置。
【請求項２】
前記所定の処理は、前記結果整合性ストレージシステムに格納された、指定された１以上のキーに対するデータの読み出しを含み、
前記実行制御手段は、複数キーを指定して前記所定の処理を前記処理手段に実行させ、
前記再実行制御手段は、前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無を判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、当該不整合が生じていた可能性のあるキーを指定して前記所定の処理を前記処理手段に再実行させる、
請求項１に記載の情報処理装置。
【請求項３】
前記結果整合性ストレージシステムにおいて、前記複数キーの各々に対するデータは、第１及び第２の領域に複製して格納され、
前記再実行制御手段は、前記第１及び前記第２の領域の各々について生成された、前記複数キーの各々に対するデータの代表値、及び、当該複数キーのすべてのデータに係る代表値を用いて、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無の判定、及び、当該不整合が生じていた可能性があるキーの特定を行う、
請求項２に記載の情報処理装置。
【請求項４】
前記再実行制御手段は、
前記所定の処理を実行させる前の、前記第１の領域における前記複数キーのすべてのデータに係る代表値と、前記第２の領域における前記複数キーのすべてのデータに係る代表値と、が異なる場合、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性があると判定し、
前記複数キーの内の、前記第１及び第２の領域におけるデータの代表値が異なるキーを、前記不整合が生じていた可能性があるキーとして特定する、
請求項３に記載の情報処理装置。
【請求項５】
前記実行制御手段は、
さらに、前記結果整合性ストレージシステムへの、データの書き込みを含む所定の前処理を、前記処理手段に実行させ、
前記処理手段により前記所定の前処理が実行された後に、前記所定の処理を前記処理手段に実行させる、
請求項１乃至４のいずれかに記載の情報処理装置。
【請求項６】
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、
データ処理方法。
【請求項７】
前記所定の処理は、前記結果整合性ストレージシステムに格納された、指定された１以上のキーに対するデータの読み出しを含み、
複数キーを指定して前記所定の処理を前記処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無を判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、当該不整合が生じていた可能性のあるキーを指定して前記所定の処理を前記処理手段に再実行させる、
請求項６に記載のデータ処理方法。
【請求項８】
前記結果整合性ストレージシステムにおいて、前記複数キーの各々に対するデータは、第１及び第２の領域に複製して格納され、
前記第１及び前記第２の領域の各々について生成された、前記複数キーの各々に対するデータの代表値、及び、当該複数キーのすべてのデータに係る代表値を用いて、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無の判定、及び、当該不整合が生じていた可能性があるキーの特定を行う、
請求項７に記載のデータ処理方法。
【請求項９】
前記所定の処理を実行させる前の、前記第１の領域における前記複数キーのすべてのデータに係る代表値と、前記第２の領域における前記複数キーのすべてのデータに係る代表値と、が異なる場合、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性があると判定し、
前記複数キーの内の、前記第１及び第２の領域におけるデータの代表値が異なるキーを、前記不整合が生じていた可能性があるキーとして特定する、
請求項８に記載のデータ処理方法。
【請求項１０】
コンピュータに、
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、
処理を実行させるプログラムを格納する、コンピュータが読み取り可能な記録媒体。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、情報処理装置、データ処理方法、及び、記録媒体に関する。
【背景技術】
【０００２】
近年、スケールアウト型のシステム構成の発展に伴い、結果整合性データストアが幅広く利用されている。結果整合性データストアは、分散システムの一貫性(Consistency)、可用性(Availability)、分断耐性(Partition-tolerance)に係るCAP定理において、一貫性を緩め、可用性と分断耐性を重視した、分散データストアである。分散データストアには、大量のデータが格納され、これら大量のデータの解析や加工を効率的に行うために、分散バッチ処理が利用される。【０００３】
例えば、Amazon S3（登録商標）のような、結果整合性のストレージサービスを利用する場合、ファイル等の大量のデータのAmazon S3への格納、及び、格納したデータに対する加工から成るバッチ処理が、複数のコンピュータで実行される。【０００４】
図９は、このような分散バッチ処理が用いられる、一般的な分散システムの構成を示すブロック図である。【０００５】
図９を参照すると、分散システムは、制御ノード７００、複数の処理ノード８００、及び、分散データストア９００を含む。分散データストア９００は、上述の結果整合性データストアであり、処理対象のファイルに含まれるデータを複製して、複数のデータストアノード９１０に格納する。制御ノード７００は、各処理ノード８００に、分散バッチ処理のジョブの実行を指示する。各処理ノード８００は、制御ノード７００により指定されたデータに対して、分散バッチ処理のジョブを実行する。【０００６】
図１０は、分散バッチ処理の動作定義の例を示す図である。【０００７】
図１０の例では、分散バッチ処理は、２つのジョブ「ジョブ１（Job1）」、「ジョブ２（Job2）」の定義、及び、各ジョブに係る分割実行の定義”split by X”、”split by D”により構成されている。分散バッチ処理は、以下のように実行される。まず、制御ノード７００が、各処理ノード８００に、ファイルに含まれるデータ（ファイルデータ）Ｘ（Ｘ１、Ｘ２、…）を割り当てる（split by X）。各処理ノード８００は、割り当てられたデータＸ（Ｘ１、Ｘ２、…）に対して、ジョブ１を実行する。ジョブ１は、ファイルから割り当てられたファイルデータＸ（Ｘ１、Ｘ２、…）を読み出し（FR(X)）、分散データストア９００にデータＤ（Ｄ１、Ｄ２、…）として書き込む（SW(D)）。処理ノード８００によるすべてのファイルデータＸ（Ｘ１、Ｘ２、…）に対するジョブ１が終了すると、制御ノード７００は、各処理ノード８００に、データＤ（Ｄ１、Ｄ２、…）を割り当てる（split by D）。各処理ノード８００は、割り当てられたデータＤ（Ｄ１、Ｄ２、…）に対して、ジョブ２を実行する。ジョブ２は、分散データストア９００から割り当てられたデータＤ（Ｄ１、Ｄ２、…）を読み出し（SR(D)）、所定の処理を行う（P(D)）。ジョブ２は、処理結果を、分散データストア９００にデータＥ（Ｅ１、Ｅ２、…）として書き込む（SW(E)）。【０００８】
結果整合性データストアでは、データの書き込み直後であっても、そのデータの読み込みの一貫性が保証されない。例えば、処理ＡがデータＸを更新したことで、あるデータストアノード９１０のデータＸの値が更新された場合、当該更新の結果が、データＸのレプリカを格納する他のデータストアノード９１０に反映される（同期される）までには、ある程度の時間を要する。このため、処理ＡによるデータＸの更新の直後に、他の処理ＢがデータＸを読み出したとしても、他のデータストアノード９１０からデータＸが読み出された場合には、更新されたデータＸの値とは異なる値が読み出される可能性がある。【０００９】
図１１は、図９の分散システムにおける、分散バッチ処理により処理されるデータの例を示す図である。【００１０】
図１１の例では、データＤ６’のように、ジョブ２において読み出されるデータが、ジョブ１による更新（同期）前のデータである可能性がある。この場合、誤った処理結果Ｅ６’が、分散データストア９００に書き込まれる。【００１１】
このような、結果整合性データストアにおけるデータの読み込みの一貫性に係る問題を解決するための技術が、例えば、特許文献１に開示されている。特許文献１の技術では、各データストアノードにおける、コマンド受信時の論理クロック値を用いて、データが確定している時点を特定する。【００１２】
また関連技術として、特許文献２には、結果整合性データストアにおいて、オペレーションの受信時刻に基づいて、オペレーションの実行順序の誤りを検出し、正しい順序で再実行する技術が開示されている。
【先行技術文献】
【特許文献】
【００１３】
【特許文献１】特許第５４８８１７８号公報
【特許文献２】特開２０１１−２４８７５２号公報
【発明の概要】
【発明が解決しようとする課題】
【００１４】
結果整合性データストアに対して上述の分散バッチ処理を行う場合、ジョブ１の終了後、例えば、上述の特許文献１のような技術により、データストアが一貫性を保証した状態になったことを確認し、ジョブ２を開始する必要がある。あるいは、ジョブ２におけるデータストアからのデータの読み出しのたびに、読み出し前に当該データの整合性を確認する必要がある。このため、ジョブの処理速度が低下する。【００１５】
本発明の目的は、上述の課題を解決し、結果整合性ストレージシステムに対する分散バッチ処理を効率的に行うことができる情報処理装置、データ処理方法、及び、記録媒体を提供することである。
【課題を解決するための手段】
【００１６】
本発明の一態様における情報処理装置は、結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させる、実行制御手段と、前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、再実行制御手段と、を備える。【００１７】
本発明の一態様におけるデータ処理方法は、結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる。【００１８】
本発明の一態様におけるコンピュータが読み取り可能な記録媒体は、コンピュータに、結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、処理を実行させるプログラムを格納する。
【発明の効果】
【００１９】
本発明の効果は、結果整合性ストレージシステムに対する分散バッチ処理を効率的に行うことができることである。
【図面の簡単な説明】
【００２０】
【図１】本発明の実施の形態の特徴的な構成を示すブロック図である。
【図２】本発明の実施の形態における、分散システムの構成を示すブロック図である。
【図３】本発明の実施の形態における、コンピュータにより実現された制御ノード１００の構成を示すブロック図である。
【図４】本発明の実施の形態における、ハッシュツリーの例を示す図である。
【図５】本発明の実施の形態における、ノード情報１２１の例を示す図である。
【図６】本発明の実施の形態における、制御ノード１００の処理を示すフローチャートである。
【図７】本発明の実施の形態における、分散システムの他の構成を示すブロックである。
【図８】本発明の実施の形態における、分散システムのさらに他の構成を示すブロックである。
【図９】一般的な分散システムの構成を示すブロック図である。
【図１０】分散バッチ処理の動作定義の例を示す図である。
【図１１】図９の分散システムにおける、分散バッチ処理により処理されるデータの例を示す図である。
【発明を実施するための形態】
【００２１】
本発明の実施の形態を、上述の図１０のような分散バッチ処理を実行する分散システムを例に説明する。
【００２２】
はじめに、本発明の実施の形態の構成を説明する。
【００２３】
図２は、本発明の実施の形態における、分散システムの構成を示すブロック図である。
【００２４】
図２を参照すると、分散システムは、制御ノード１００、１以上の処理ノード（または、処理手段）２００「ＰＮ１」、「ＰＮ２」、…、及び、分散データストア（または、分散ストレージシステム）３００を含む。なお、符号の後の「」内は、当該符号で表される構成要素の識別子を示す。
【００２５】
制御ノード１００、処理ノード２００、及び、分散データストア３００は、ネットワーク等により接続される。制御ノード１００は、本発明の情報処理装置の一実施形態である。処理ノード２００は、例えば、ＣＰＵ（Central Processing Unit）を備えたコンピュータである。また、分散データストア３００は、上述の結果整合性データストアである。
【００２６】
分散データストア３００は、複数のデータストアノード（または、複数の領域）３１０「ＤＮ１」、「ＤＮ２」、…を含む。各データストアノード３１０は、複数のキーの各々に対するデータと、ハッシュツリーとを格納する。ハッシュツリーは、データストアノード３１０に格納されているデータのハッシュ値を表す。
【００２７】
図４は、本発明の実施の形態における、ハッシュツリーの例を示す図である。
【００２８】
ハッシュツリーの葉ノードには、データストアノード３１０が格納しているデータのキーと、当該キーに対するデータを代表する値（代表値）として、ハッシュ値が設定される。また、葉ノードより上位の各ノードには、当該ノードをルートノードとするサブツリーの全葉ノードのデータに係る代表値として、当該ノードの全子ノードのハッシュ値を連結した値のハッシュ値（連結ハッシュ値）が設定される。
【００２９】
図４の例では、データストアノード３１０「ＤＮ１」のハッシュテーブルにおいて、葉ノード「Ｎ１１１」に、キー「１」、及び、キー「１」に対するデータ「Ｄ１」のハッシュ値「Ｈ１１１＝Ｈ（Ｄ１）」が設定されている。同様に、葉ノード「Ｎ１１２」に、キー「２」、及び、キー「２」に対するデータ「Ｄ２」のハッシュ値「Ｈ１１２＝Ｈ（Ｄ２）」が設定されている。また、葉ノード「Ｎ１１１」、「Ｎ１１２」の親ノード「Ｎ１１」には、葉ノード「Ｎ１１１」、「Ｎ１１２」のハッシュ値「Ｈ１１１」、「Ｈ１１２」を連結した値のハッシュ値「Ｈ１１＝Ｈ（Ｈ１１１，Ｈ１１２）」が設定されている。さらに、ノード「Ｎ１１」、「Ｎ１２」の親ノード「Ｎ１」には、ノード「Ｎ１１」、「Ｎ１２」のハッシュ値を連結した値のハッシュ値「Ｈ１＝Ｈ（Ｈ１１，Ｈ１２）」が設定されている。
【００３０】
なお、サブツリーの全葉ノードのデータを代表する値であれば、連結ハッシュ値以外のハッシュ値が用いられてもよい。
【００３１】
また、データを代表する値であれば、ハッシュ値以外を代表値として用いてもよい。さらに、複数のキーから成る集合と当該集合に属するすべてのキーのデータに係る代表値とを関連付けることができれば、ハッシュツリー以外の方法で、当該複数のキーから成る集合を表してもよい。
【００３２】
分散データストア３００では、各キーに対するデータが複製され、複数のデータストアノード３１０に格納される。本発明の実施の形態では、ハッシュツリーの所定の階層以下のサブツリーの単位で、当該サブツリーに属する複数キーに対するデータが複製され、２つのデータストアノード３１０に格納される。
【００３３】
図４の例では、キー「５〜８」に対するデータが、データストアノード３１０「ＤＮ１」、「ＤＮ２」に格納されている。これらのデータは、データストアノード３１０「ＤＮ１」のハッシュテーブルにおけるノード「Ｎ２」以下のサブツリー、データストアノード３１０「ＤＮ２」のハッシュテーブルにおけるノード「Ｎ１」以下のサブツリーに対応する。
【００３４】
なお、データストアノード３１０が格納しているデータ値が更新されると、当該データストアノード３１０、あるいは、分散データストア３００内の制御部（図示せず）等により、ハッシュツリーのハッシュ値も更新される。
【００３５】
また、処理ノード２００が分散データストア３００上のデータにアクセス（書き込み、読み出し）する場合、アクセス先のデータストアノード３１０は、処理ノード２００により指定されず、分散データストア３００内の制御部等により決定される。また、上述の通り、分散データストア３００は結果整合性データストアであるため、あるキーに対するデータの書き込み直後の、当該キーに対するデータの読み出しの一貫性は保証されない。
【００３６】
制御ノード１００は、実行制御部１１０、ノード情報記憶部１２０、チャンク抽出部１３０、及び、再実行制御部１４０を含む。
【００３７】
チャンク抽出部１３０は、データストアノード３１０に格納されたハッシュツリーから、複数のキーから成る集合（以下、チャンクと呼ぶ）を抽出する。ここで、チャンクは、例えば、上述のデータが複製されるサブツリーの単位で抽出される。
【００３８】
実行制御部１１０は、複数の処理ノード２００に、処理対象のファイルデータを指定して、分散バッチ処理のジョブ１の実行を指示する。また、すべてのファイルデータに対するジョブ１が終了すると、実行制御部１１０は、複数の処理ノード２００に、チャンク抽出部１３０で抽出されたチャンクを処理対象のチャンクとして指定し、ジョブ２（所定の処理）の実行を指示する。
【００３９】
ノード情報記憶部１２０は、ノード情報１２１を記憶する。ノード情報１２１は、ハッシュツリーにおける、ジョブ実行の指示済みのチャンクに係る情報を示す。
【００４０】
図５は、本発明の実施の形態における、ノード情報１２１の例を示す図である。
【００４１】
ノード情報１２１は、「キー範囲」、「ノード識別子」、及び、「ハッシュ値」を含む。ここで、「キー範囲」は、実行指示済みのジョブ２において指定したチャンクのキー集合（範囲）を示す。「ノード識別子」は、チャンクに対応するサブツリーのルートノードを示す。「ノード識別子」には、ハッシュツリーにおける当該チャンクに対応するサブツリーのルートノードの識別子が、当該ハッシュツリーが格納されたデータストアノード３１０の識別子とともに設定される。「ハッシュ値」は、チャンクに対応するサブツリーのルートノードに設定された連結ハッシュ値（以下、チャンクのハッシュ値と呼ぶ）を示す。
【００４２】
再実行制御部１４０は、処理対象のチャンクのキー集合のいずれかに対するデータの不整合（コンフリクト）が、ジョブ２実行時に生じていた可能性の有無を判定する。コンフリクトが生じていた可能性ありと判定された場合、再実行制御部１４０は、コンフリクトが生じていた可能性があるキーを特定する。そして、再実行制御部１４０は、当該特定したキーについてのコンフリクトの解消（当該キーに対するデータの同期化）が確認された時点で、当該特定したキーを指定して、処理ノード２００に、ジョブ２の再実行を指示する。
【００４３】
ここで、２つのデータストアノード３１０に格納されている、同じキーに対するデータのコンフリクトが、ジョブ２実行時に生じていた可能性の有無は、以下のようにして判定される。
【００４４】
例えば、ジョブ１終了後であってジョブ２実行前の、一方のデータストアノード３１０におけるあるキーに対するデータ値と、他方のデータストアノード３１０における当該キーに対するデータ値とが、同じであったと仮定する。この場合、ジョブ２実行前に、ジョブ１による当該キーに対するデータの更新が、データストアノード３１０間で反映されており、ジョブ２実行時にコンフリクトは発生していない。したがって、ジョブ２において読み出されるデータは、ジョブ１による更新後のデータである。
【００４５】
一方、ジョブ１終了後であってジョブ２実行前の、一方のデータストアノード３１０におけるあるキーに対するデータ値と、他方のデータストアノード３１０における当該キーに対するデータ値とが異なると仮定する。この場合、ジョブ２実行前に、ジョブ１による当該データの更新が、データストアノード３１０間で反映されておらず、ジョブ２実行時にコンフリクトが発生していた可能性がある。したがって、ジョブ２において読み出されるデータは、更新後、及び、更新前のいずれかのデータである。
【００４６】
本発明の実施の形態では、再実行制御部１４０は、このような、コンフリクトが生じていた可能性の有無の判定を、以下のように、チャンクのハッシュ値を用いて、チャンクのキー集合に対して一括して行う。再実行制御部１４０は、ジョブ２実行前に取得された、一方のデータストアノード３１０のチャンクのハッシュ値（ノード情報１２１のハッシュ値）と、他のデータストアノード３１０の同じキー集合のチャンクのハッシュ値とを比較する。そして、チャンクのハッシュ値が異なる場合、再実行制御部１４０は、読み出し対象の同じチャンクのキー集合のいずれかのキーに対するデータのコンフリクトが、ジョブ２実行時に生じていた可能性があると判定する。
【００４７】
処理ノード２００は、処理実行部２１０を含む。処理実行部２１０は、制御ノード１００の実行制御部１１０により指定されたファイルデータを用いて、ジョブ１を実行する。また、処理実行部２１０は、実行制御部１１０により指定されたチャンクのキー集合の各キーに対するデータを用いて、ジョブ２を実行する。また、処理実行部２１０は、再実行制御部１４０より指定されたキーに対するデータを用いて、ジョブ２を再実行する。複数の処理ノード２００は、指定されたジョブを並列に実行してもよい。
【００４８】
なお、制御ノード１００は、ＣＰＵとプログラムを記憶した記憶媒体を含み、プログラムに基づく制御によって動作するコンピュータであってもよい。
【００４９】
図３は、本発明の実施の形態における、コンピュータにより実現された制御ノード１００の構成を示すブロック図である。制御ノード１００は、ＣＰＵ１０１、ハードディスクやメモリ等の記憶手段（記憶媒体）１０２、他の装置等とデータ通信を行う通信手段１０３、キーボード等の入力手段１０４、及び、ディスプレイ等の出力手段１０５を含む。
【００５０】
ＣＰＵ１０１は、実行制御部１１０、チャンク抽出部１３０、及び、再実行制御部１４０の機能を実現するためのコンピュータプログラムを実行する。記憶手段１０２は、ノード情報記憶部１２０のデータ（ノード情報１２１）を記憶する。通信手段１０３は、分散データストア３００から、各データストアノード３１０のハッシュテーブルを受信する。また、通信手段１０３は、処理ノード２００へ分散バッチ処理のジョブの実行を指示する。入力手段１０４は、利用者等から、分散バッチ処理の実行の指示を受け付ける。出力手段１０５は、利用者等へ、分散バッチ処理の実行結果を出力する。
【００５１】
また、図２に示した制御ノード１００の各構成要素は、独立した論理回路でもよい。
【００５２】
次に、本発明の実施の形態の動作を説明する。
【００５３】
図６は、本発明の実施の形態における、制御ノード１００の処理を示すフローチャートである。
【００５４】
各処理ノード２００におけるジョブ１の実行が終了すると、ジョブ実行制御部１１０が、ジョブ２に係る分割実行の定義”split by D”に従って、チャンク抽出部１３０にチャンクの抽出を指示する。
【００５５】
チャンク抽出部１３０は、分散データストア３００を構成するデータストアノード３１０の識別子のリストを取得する（ステップＳ１０１）。
【００５６】
チャンク抽出部１３０は、取得したリストに含まれる各データストアノード３１０から、ハッシュツリーを取得する（ステップＳ１０２）。
【００５７】
チャンク抽出部１３０は、取得したリストに含まれるデータストアノード３１０の内の一つを、処理対象のデータストアノード３１０（対象データストアノード）として選択する（ステップＳ１０３）。
【００５８】
チャンク抽出部１３０は、対象データストアノードのハッシュツリーから、チャンクを抽出する（ステップＳ１０４）。
【００５９】
実行制御部１１０は、抽出されたチャンクの内の一つを処理対象のチャンク（対象チャンク）として選択する（ステップＳ１０５）。
【００６０】
実行制御部１１０は、対象チャンクのキー集合と同じキー集合が、ノード情報１２１に存在するかどうかを判定する（ステップＳ１０６）。
【００６１】
ステップＳ１０６で、対象チャンクのキー集合と同じキー集合がノード情報１２１に存在しない場合（ステップＳ１０６／Ｎ）、実行制御部１１０は、対象チャンクを複数の処理ノード２００の内のいずれかに割り当てる。そして、実行制御部１１０は、当該処理ノード２００に、対象チャンクを指定して、ジョブ２の実行を指示する（ステップＳ１０７）。
【００６２】
ジョブ２の実行を指示された処理ノード２００は、指定された対象チャンクに対して、ジョブ２を実行する。ジョブ２では、分散データストア３００に格納された、対象チャンクのキー集合の各キーに対するデータの読み出しが行われる。
【００６３】
実行制御部１１０は、ノード情報１２１に、ジョブ実行を指示したチャンクに係るキー集合、ノード識別子、及び、ハッシュ値を設定する（ステップＳ１０８）。ここで、ノード識別子、及び、ハッシュ値には、対象データストアノードのハッシュツリーにおける、対象チャンクに対応するサブツリーのルートノードの識別子、及び、ルートノードに設定された連結ハッシュ値（対象チャンクのハッシュ値）が設定される。実行制御部１１０は、ステップＳ１１５の処理に進む。
【００６４】
ステップＳ１０６で、対象チャンクのキー集合と同じキー集合がノード情報１２１に存在する場合（ステップＳ１０６／Ｙ）、再実行制御部１４０は、対象データストアノードのハッシュツリーから、対象チャンクのハッシュ値を抽出する。すなわち、再実行制御部１４０は、対象チャンクに対応するサブツリーのルートノードに設定されている連結ハッシュ値を抽出する。そして、再実行制御部１４０は、抽出された対象チャンクのハッシュ値と、ノード情報１２１に設定されている、対象チャンクと同じキー集合に対するハッシュ値とが一致するかどうかを判定する（ステップＳ１０９）。
【００６５】
ハッシュ値が一致する場合（ステップＳ１０９／Ｙ）、再実行制御部１４０は、対象チャンクのキー集合の各キーについて、コンフリクトが生じていた可能性がないと判定する（ステップＳ１１０）。すなわち、再実行制御部１４０は、対象チャンクのキー集合の各キーについて、データストアノード３１０間でのコンフリクトが、ジョブ２の実行時に生じていた可能性はないと判定する。再実行制御部１４０は、ステップＳ１１５の処理に進む。
【００６６】
一方、ステップＳ１０８でハッシュ値が一致しない場合（ステップＳ１０９／Ｎ）、再実行制御部１４０は、対象チャンクのキー集合のいずれかのキーについて、コンフリクトが発生していた可能性があると判定する（ステップＳ１１１）。すなわち、再実行制御部１４０は、対象チャンクのキー集合のいずれかのキーについて、データストアノード３１０間でのコンフリクトが、ジョブ２の実行時に生じていた可能性があると判定する。
【００６７】
再実行制御部１４０は、コンフリクトが生じていた可能性があるキーを特定する（ステップＳ１１２）。ここで、再実行制御部１４０は、対象データストアノードのハッシュツリーにおける対象チャンクのサブツリーと、ノード情報１２１における対象チャンクと同じキー集合に対するノード識別子で示されるサブツリーとの間で、ハッシュ値を比較する。そして、再実行制御部１４０は、ハッシュ値が異なる葉ノードのキーを、コンフリクトが生じていた可能性があるキーとして特定する。
【００６８】
再実行制御部１４０は、特定したキーについてのコンフリクトの解消（当該キーに対するデータの同期化）が確認されるまで待機する（ステップＳ１１３）。ここで、再実行制御部１４０は、コンフリクトが生じているキーの葉ノード（ハッシュ値が異なる葉ノード）間で、ハッシュ値が同一になるまで待機する。
【００６９】
コンフリクトの解消が確認された場合（ステップＳ１１３／Ｙ）、再実行制御部１４０は、特定したキーを複数の処理ノード２００の内のいずれかに割り当てる。そして、再実行制御部１４０は、当該処理ノード２００に、特定したキーを指定して、ジョブ２の再実行を指示する（ステップＳ１１４）。ここで、再実行制御部１４０は、例えば、ステップＳ１０７で実行を指示した、当該特定したキーを含むチャンクに対するジョブ２が終了した時点で、当該特定したキーを指定したジョブ２の再実行を指示する。
【００７０】
ジョブ２の実行を指示された処理ノード２００は、指定されたキーに対して、ジョブ２を再実行する。ジョブ２では、分散データストア３００に格納された、指定されたキーに対するデータの読み出しが行われる。
【００７１】
以下、ステップＳ１０４で抽出された全てのチャンクについて、ステップＳ１０５からの処理が繰り返される（ステップＳ１１５）。
【００７２】
また、ステップＳ１０１で取得した全てのデータストアノード３１０について、ステップＳ１０３からの処理が繰り返される（ステップＳ１１６）。
【００７３】
次に、具体例を説明する。ここでは、ジョブ１の実行の結果、分散データストア３００のデータストアノード３１０「ＤＮ１」、「ＤＮ２」に、図４のような、ハッシュツリーが格納されていると仮定する。すなわち、分散データストア３００のデータストアノード３１０「ＤＮ１」に、キー集合「１〜４」、「５〜８」に対するデータが格納されており、データストアノード３１０「ＤＮ２」に、キー集合「５〜８」、「９〜１２」に対するデータが格納されている。キー「６」について、データストアノード３１０「ＤＮ２」のデータは「Ｄ６」に更新されたが、データストアノード３１０「ＤＮ１」のデータは更新前のデータ「Ｄ６’」のままであり、コンフリクトが生じている。また、ハッシュツリー上のルートノードから２番目の階層以下のサブツリーに属するキーの集合を、チャンクとして抽出すると仮定する。
【００７４】
はじめに、チャンク抽出部１３０は、データストアノード３１０のリストとして、「ＤＮ１」、「ＤＮ２」を取得する。
【００７５】
チャンク抽出部１３０は、データストアノード３１０「ＤＮ１」、「ＤＮ２」から、図４のようなハッシュツリーを取得する。
【００７６】
チャンク抽出部１３０は、データストアノード３１０「ＤＮ１」を対象データストアノードとして選択する。
【００７７】
チャンク抽出部１３０は、データストアノード３１０「ＤＮ１」のハッシュツリーから、２番目の階層のノード「Ｎ１」、「Ｎ２」以下のサブツリーに属するキー集合「１〜４」、「５〜８」を、チャンクとして抽出する。
【００７８】
実行制御部１１０は、キー集合「１〜４」を対象チャンクとして選択する。
【００７９】
キー集合「１〜４」はノード情報１２１に存在しないため、実行制御部１１０は、キー集合「１〜４」を指定して、処理ノード２００「ＰＮ１」に、ジョブ２の実行を指示する。
【００８０】
処理ノード２００「ＰＮ１」は、キー集合「１〜４」に対して、ジョブ２を実行する。
【００８１】
実行制御部１１０は、図５のように、ノード情報１２１に、キー集合「１〜４」、ノード識別子「ＤＮ１−Ｎ１」、及び、当該ノードのハッシュ値を設定する。
【００８２】
次に、実行制御部１１０は、キー集合「５〜８」を対象チャンクとして選択する。
【００８３】
キー集合「５〜８」もノード情報１２１に存在しないため、実行制御部１１０は、キー集合「５〜８」を指定して、処理ノード２００「ＰＮ２」に、ジョブ２の実行を指示する。
【００８４】
処理ノード２００「ＰＮ２」は、キー集合「５〜８」に対して、ジョブ２を実行する。
【００８５】
実行制御部１１０は、図５のように、ノード情報１２１に、キー集合「５〜８」、ノード識別子「ＤＮ１−Ｎ２」、及び、当該ノードのハッシュ値を設定する。
【００８６】
次に、チャンク抽出部１３０は、データストアノード３１０「ＤＮ２」を対象データストアノードとして選択する。
【００８７】
チャンク抽出部１３０は、データストアノード３１０「ＤＮ２」のハッシュツリーから、２番目の階層のノード「Ｎ１」、「Ｎ２」以下のサブツリーに属するキー集合「５〜８」、「９〜１２」を、チャンクとして抽出する。
【００８８】
実行制御部１１０は、キー集合「５〜８」を対象チャンクとして選択する。
【００８９】
キー集合「５〜８」は、図５のように、ノード情報１２１に存在する。また、対象データストアノード「ＤＮ２」のハッシュツリーにおけるノード「Ｎ１」のハッシュ値は、図５のノード情報１２１の当該キー集合「５〜８」に対するハッシュ値と異なる。したがって、再実行制御部１４０は、キー集合「５〜８」のいずれかのキーについて、コンフリクトが発生していた可能性があると判定する。再実行制御部１４０は、対象データストアノード「ＤＮ２」のハッシュツリーにおけるノード「Ｎ１」以下のサブツリーと、データストアノード３１０「ＤＮ１」のハッシュツリーにおけるノード「Ｎ２」以下のサブツリーとを比較する。そして、再実行制御部１４０は、ハッシュ値がＨ（Ｄ６）、Ｈ（Ｄ６’）のように異なる葉ノードのキー「６」を、コンフリクトが生じていた可能性があるキーとして特定する。この場合、ジョブ２は、キー「６」に対して、更新後のデータ「Ｄ６」を読み出した可能性もあるし、更新前のデータ「Ｄ６’」を読み出した可能性もある。
【００９０】
再実行制御部１４０は、コンフリクトが生じていた可能性があるキー「６」の葉ノード間で、ハッシュ値が同一になるまで待機する。
【００９１】
データストアノード３１０「ＤＮ１」のキー「６」のデータ「Ｄ６’」が更新され、「Ｄ６」になった時点で、キー「６」の葉ノードのハッシュ値は同一（Ｈ（Ｄ６））となる。
【００９２】
再実行制御部１４０は、キー「６」を指定して、処理ノード２００「ＰＮ３」に、ジョブ２の再実行を指示する。
【００９３】
処理ノード２００「ＰＮ３」は、キー「６」に対して、ジョブ２を再実行する。ジョブ２は、キー「６」に対して、更新後のデータ「Ｄ６」を読み出す。
【００９４】
以上により、本発明の実施の形態の動作が完了する。
【００９５】
次に、本発明の実施の形態の特徴的な構成を説明する。図１は、本発明の実施の形態の特徴的な構成を示すブロック図である。
【００９６】
図１を参照すると、本発明の実施の形態における制御ノード１００（情報処理装置）は、実行制御部１１０、及び、再実行制御部１４０を含む。
【００９７】
実行制御部１１０は、分散データストア３００（結果整合性ストレージシステム）に格納された、キーに対するデータの読み出しを含む所定の処理を、処理ノード２００（処理手段）に実行させる。
【００９８】
再実行制御部１４０は、所定の処理を処理ノード２００に実行させた後で、当該所定の処理実行時に分散データストア３００に格納されていた当該キーに対するデータに不整合が生じていた可能性の有無を、当該データの代表値に基づいて判定する。そして、再実行制御部１４０は、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、所定の処理を処理ノード２００に再実行させる。
【００９９】
本発明の実施の形態によれば、結果整合性ストレージシステムに対する分散バッチ処理を効率的に行うことができる。その理由は、処理ノード２００に所定の処理を実行させた後に、再実行制御部１４０が、分散データストア３００に格納されていたデータに不整合が生じていた可能性の有無を判定し、不整合の解消が確認された時点で、所定の処理を再実行させるためである。
【０１００】
一般的な結果整合性データストアでは、同一キーに対するデータへのアクセスの並列度が高い場合や、データストアノード間の通信障害、データストアノードの障害が発生したときに、一貫性が損なわれる可能性が高い。しかしながら、上述の分散バッチ処理のように、ジョブ１によるデータの書き込み完了後、ジョブ２によるデータの読み込みを行うようなケースでは、ジョブ２における読み出し時に、一貫性が損なわれる可能性が低い。したがって、本発明の実施の形態のように、ジョブ２を実行させた後、実行時にコンフリクトが発生していた可能性があるデータに対してジョブ２を再実行させた方が、ジョブ２の実行前や各読み出し前に整合性を確認するより、全体的なスループットは高くなる。
【０１０１】
また、本発明の実施の形態では、ジョブ２をハッシュツリー上のチャンク単位で実行させ、チャンク単位でコンフリクトが発生していた可能性の有無を判定し、コンフリクトが発生していた可能性のあるキーのみに対して、ジョブ２を再実行させる。これにより、全体的なスループットをさらに向上させることができる。
【０１０２】
以上、実施形態を参照して本願発明を説明したが、本願発明は上記実施形態に限定されるものではない。本願発明の構成や詳細には、本願発明のスコープ内で当業者が理解し得る様々な変更をすることができる。
【０１０３】
例えば、本発明の実施の形態では、制御ノード１００と処理ノード２００とを異なる装置としたが、複数の処理ノード２００のいずれかが、制御ノード１００の各構成要素を含んでいてもよい。
【０１０４】
図７は、本発明の実施の形態における、分散システムの他の構成を示すブロックである。
【０１０５】
図７の構成では、処理ノード２００「ＰＮ１」が、処理実行部２１０に加えて、実行制御部１１０、ノード情報記憶部１２０、チャンク抽出部１３０、及び、再実行制御部１４０を含む。この場合、実行制御部１１０は、自処理ノード２００「ＰＮ１」を含む複数の処理ノード２００のいずれかに、チャンクを割り当て、ジョブ２を実行させる。
【０１０６】
さらに、処理ノード２００が一つの場合、当該処理ノード２００が、制御ノード１００の各構成要素を含んでいてもよい
図８は、本発明の実施の形態における、分散システムのさらに他の構成を示すブロックである。
【０１０７】
図８の構成では、処理ノード２００が一つであり、当該処理ノード２００が、実行制御部１１０、ノード情報記憶部１２０、チャンク抽出部１３０、及び、再実行制御部１４０を含む。この場合、実行制御部１１０は、自処理ノード２００の処理実行部２１０に、チャンクを順番に指定し、ジョブ２を実行させる。ここで、各チャンクに対する処理は、シーケンシャルに実行されてもよいし、複数のプロセスにより並列に実行されてもよい。
【０１０８】
また、本発明の実施の形態では、各キーに対するデータが複製され、２つのデータストアノード３１０に格納されたが、３つ以上のデータストアノード３１０に格納されてもよい。この場合、再実行制御部１４０は、ステップＳ１１１で、コンフリクトが生じていた可能性ありと判定後、ステップＳ１１３で、ハッシュ値の一致（同期化）が確認されたときに、ノード情報１２１におけるハッシュ値を、同期化後の新たなハッシュ値で更新する。これにより、各キーに対するデータが３つ以上のデータストアノード３１０に格納されていても、全データストアノード３１０間での不整合の解消時に、ジョブ２を再実行できる。
【０１０９】
この出願は、２０１４年８月２１日に出願された日本出願特願２０１４−１６８２５２を基礎とする優先権を主張し、その開示の全てをここに取り込む。
【符号の説明】
【０１１０】
１００    制御ノード
１０１    ＣＰＵ
１０２    記憶手段
１０３    通信手段
１０４    入力手段
１０５    出力手段
１１０    実行制御部
１２０    ノード情報記憶部
１２１    ノード情報
１３０    チャンク抽出部
１４０    再実行制御部
２００    処理ノード
２１０    処理実行部
３００    分散データストア
３１０    データストアノード
７００    制御ノード
８００    処理ノード
９００    分散データストア
９１０    データストアノード
【図１】
2016027455.tif 000003
【図２】
2016027455.tif 000004
【図３】
2016027455.tif 000005
【図４】
2016027455.tif 000006
【図５】
2016027455.tif 000007
【図６】
2016027455.tif 000008
【図７】
2016027455.tif 000009
【図８】
2016027455.tif 000010
【図９】
2016027455.tif 000011
【図１０】
2016027455.tif 000012
【図１１】
2016027455.tif 000013
【手続補正書】
【提出日】平成29年2月13日(2017.2.13)
【手続補正２】
【補正対象書類名】特許請求の範囲
【補正対象項目名】全文
【補正方法】変更
【補正の内容】
【特許請求の範囲】
【請求項１】
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させる、実行制御手段と、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、再実行制御手段と、
を備えた、情報処理装置。
【請求項２】
前記所定の処理は、前記結果整合性ストレージシステムに格納された、指定された１以上のキーに対するデータの読み出しを含み、
前記実行制御手段は、複数キーを指定して前記所定の処理を前記処理手段に実行させ、
前記再実行制御手段は、前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無を判定し、当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、当該不整合が生じていた可能性のあるキーを指定して前記所定の処理を前記処理手段に再実行させる、
請求項１に記載の情報処理装置。
【請求項３】
前記結果整合性ストレージシステムにおいて、前記複数キーの各々に対するデータは、第１及び第２の領域に複製して格納され、
前記再実行制御手段は、前記第１及び前記第２の領域の各々について生成された、前記複数キーの各々に対するデータの代表値、及び、当該複数キーのすべてのデータに係る代表値を用いて、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無の判定、及び、当該不整合が生じていた可能性があるキーの特定を行う、
請求項２に記載の情報処理装置。
【請求項４】
前記再実行制御手段は、
前記所定の処理を実行させる前の、前記第１の領域における前記複数キーのすべてのデータに係る代表値と、前記第２の領域における前記複数キーのすべてのデータに係る代表値と、が異なる場合、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性があると判定し、
前記複数キーの内の、前記第１及び第２の領域におけるデータの代表値が異なるキーを、前記不整合が生じていた可能性があるキーとして特定する、
請求項３に記載の情報処理装置。
【請求項５】
前記実行制御手段は、
さらに、前記結果整合性ストレージシステムへの、データの書き込みを含む所定の前処理を、前記処理手段に実行させ、
前記処理手段により前記所定の前処理が実行された後に、前記所定の処理を前記処理手段に実行させる、
請求項１乃至４のいずれかに記載の情報処理装置。
【請求項６】
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、
データ処理方法。
【請求項７】
前記所定の処理は、前記結果整合性ストレージシステムに格納された、指定された１以上のキーに対するデータの読み出しを含み、
複数キーを指定して前記所定の処理を前記処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無を判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、当該不整合が生じていた可能性のあるキーを指定して前記所定の処理を前記処理手段に再実行させる、
請求項６に記載のデータ処理方法。
【請求項８】
前記結果整合性ストレージシステムにおいて、前記複数キーの各々に対するデータは、第１及び第２の領域に複製して格納され、
前記第１及び前記第２の領域の各々について生成された、前記複数キーの各々に対するデータの代表値、及び、当該複数キーのすべてのデータに係る代表値を用いて、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性の有無の判定、及び、当該不整合が生じていた可能性があるキーの特定を行う、
請求項７に記載のデータ処理方法。
【請求項９】
前記所定の処理を実行させる前の、前記第１の領域における前記複数キーのすべてのデータに係る代表値と、前記第２の領域における前記複数キーのすべてのデータに係る代表値と、が異なる場合、前記複数キーの内のいずれかに対するデータに不整合が生じていた可能性があると判定し、
前記複数キーの内の、前記第１及び第２の領域におけるデータの代表値が異なるキーを、前記不整合が生じていた可能性があるキーとして特定する、
請求項８に記載のデータ処理方法。
【請求項１０】
コンピュータに、
結果整合性ストレージシステムに格納された、キーに対するデータの読み出しを含む所定の処理を、処理手段に実行させ、
前記所定の処理を前記処理手段に実行させた後で、当該所定の処理実行時に前記結果整合性ストレージシステムに格納されていた前記キーに対するデータに不整合が生じていた可能性の有無を、前記データの代表値に基づいて判定し、
当該不整合が生じていた可能性があると判定された場合に、当該不整合の解消が確認された時点で、前記所定の処理を前記処理手段に再実行させる、
処理を実行させるプログラム。
【国際調査報告】
2016027455.tif 000014
2016027455.tif 000015
2016027455.tif 000016