この記事はリクルートライフスタイル Advent Calendar 2016の25日目の記事です。
データエンジニアリンググループ・ゆるふわAwesome機械学習エンジニアのtomomotoです。主にデータ分析や機械学習を中心に、データ活用を推進したり、システム開発したり、フリー素材になったりしています。
本記事では、今更ではありますが、KVSの仕組み概要について、HBaseとCassandraを例にして紹介したいと思います。（仕事、クリスマス、結婚記念日、何も関係ないです！）
KVSとは
そもそもKVSとは、なんでしょうか？名前の通りKeyValueStoreであることが条件だとすると、KeyからValueをとってくるDataBaseということになりますが、それで十分なのでしょうか？そうするとファイル名(Key)でファイル情報(Value)をとってくるファイルシステムもKVSということになってしまいますが、それではしっくりきません。
一般的には、CAP定理におけるRDBが該当するCA型以外の型、つまりAP/CP型がKVSと言われています。CAP定理とは、簡単にいうと下記のCAP定理の３要素を同時に満たすことは難しく、どれかの要素を犠牲にする必要があるという定理です。正確には、この３要素間のトレードオフ関係ということを表しており、必ずしも２要素を選ぶというわけではありません。
- CAP定理の３要素
- Consistency(一貫性)
- Availability(可用性)
- Partition Tolerance(ネットワーク分断耐性)
またKVSが強く注目されたのは、2006年のGoogleBigTableの論文や2007年のAmazonDynamoDBの論文公開からだと思われます。GoogleBigTableは、可用性を犠牲にした代わりに強い一貫性と高い負荷分散性能を、AmazonDynamoDBは、一貫性を犠牲にした代わりにスケーラビリティと高可用性を実現していました。この論文をきっかけに、BigTableからHBase、DynamoDBからCassandra(一部BigTableの影響もあり)、といったOSSが開発されました。
以上のようにKVSには色々なルーツや定義がありますが、私は下記のような問題点を許容するしたものこそがKVSだと考えています。
-
CP型のKVS:HBase
- 一時的なダウンタイムを許容することによって、強い一貫性と高い負荷分散性能を実現しているDataBase
-
AP型のKVS:Cassandra
- 強い一貫性を保てないことを許容するが、EventualConsistencyによってある程度の一貫性を保つ、高い可用性と高い負荷分散性能を実現しているDataBase
KVSの仕組み
AP型であれ、CP型であれ、KVSは、高い負荷分散性能を実現するために多数のノードでデータ管理をする必要があります。しかし、ノードの数が増えるほど、いずれかのノードに障害が発生する確率は高まり、データロストや可用性低下につながりやすくなります。この問題の解決のためには、レプリケーション（複数のノードで同じデータを管理する）を行う必要があります。しかし、複数のノードで同じデータを扱っていると一貫性は保持することは難しくなります。以下では、このトレードオフの関係を、CP/AP型のKVSではどのようにバランスをとっているかを、CP型の代表としてHBaseを、AP型の代表としてCassandraを、例に説明します。
HBaseの仕組み
HBaseでは、データのKeyをRangeで分割したデータの塊であるRegionという単位でデータを取り扱います。Keyの値が1-5000だとして1000でデータを区切るとしたら、Keyが1001-2000の範囲に収まるデータが1つのRegionとなります。このような分割をSharding(シャーディング)といい、HBaseでは自動Shardingが取り入れられています。
RegionServerは、１つ以上のRegionを管理します。また、各RegionServerが管理しているRegionの情報がZookeeperが管理しており、ZookeeperがRegionの再配置によるリバランスなどを行っています。そのためClientはZookeeperにRegionServerの位置を教えてもらい、RegionServerと直接データのやり取りを行います。さらにHBaseは、データの行レベルの一貫性を担保するために、各Regionは、必ず１つのRegionServerによって管理されています。しかし、これは同時にデータロストの危険性が増し、さらに可用性も下げてしまいます。
RegionServerでは、RegionはMemostoreとしてメモリ上に展開されています。しかし、Memostoreの時点ではレプリケーションされておらず、さらにメモリ上だけではServerが落ちたときにデータをロストしてしまいます。そこでHBaseでは、変更内容をHLogとしてDiskに書き込むようになっています。このような仕組みを一般的にはWAL(WriteAheadLog, ログ先行書き込み)といいます。
Memostore/HFileは、操作内容や対象のレコードなどの情報がまとまったログが集まったようなデータです。そのためWriteでもUpdateでも、Memstoreに追記していく形になります。またReadの時は、Memstoreから対象のレコードの最新の情報を確認し、データを読み込みます。もしMemstoreに十分な情報がなかった場合は、HFileも参照します。
Memostoreのサイズが大きくなった場合などに、MemostoreのデータをHDFSにHFileとして書き込みます。通常HDFSでは、リプリケーションが行われるので、HFileとして書き出された時点のデータのロストの可能性は非常に低くなります。HFile数が多くなってきた場合などには、HFileの集約処理を行うCompaction(コンパクション)が行なわれます。集約処理の内容は、概に後続の処理で必要でなくなっているログの削除を行うことです。
以上のような仕組みから、一部データロストの危険性が高い部分やMasterやRegionServerの復旧のために可用性が下がる可能性がありますが、RegionServerを増やすことによってスケールしつつも、行レベルの一貫性を保つことができています。
Cassandraの仕組み
Cassandraでは、Consistent Hashing(コンシステントハッシュ法)を用いて各データをどのNodeに配置するか決定します。Consistent Hashingとは、「データのキーのハッシュ値」と「データを管理するNode」をハッシュリング(ハッシュを数値にして、十分大きな数字で割った余りをリング上に並べるイメージです。)上にマッピングし、その位置関係から担当Nodeをきめる手法です。例えば、図のDataAは、KeyのHashが星の位置に該当し、レプリケーション数が3なので、担当Nodeは5,6,7になります。Consistent Hashingによって、Nodeの追加や削除時のデータの再配置コストを減らすことできます。また、Masterを持たず、各Nodeが自分の担当範囲のデータを自律的に管理することも実現でき、可用性の向上も実現できています。
Consistent Hashingを利用するためには、各NodeはClientから問い合わせされた時のために、自分以外も含めた各Nodeの担当範囲を知っておく必要があります。この情報をゆるく共有するために、Gossip Protocol(ゴシッププロトコル)という仕組みを取り入れています。Gossip Protocolとは、各Nodeがランダムに他のNodeと情報共有することで、時間が経てば全てのNodeが最新情報を保持できるようになる仕組みです。
Cassandraのある程度の一貫性は、主にQuorum Protocol(クォーラムプロトコル)によって支えられています。Quorum Protocolとは、【レプリケーション数】より、【読み込み成功ノード必要数】と【書き込み成功ノード必要数】の合計の方が大きいことを満たすことによって、データ読み込み成功時には、必ず最新Versionのデータが読み込み範囲に含まれていることを保障する仕組みです。また、【読み込み成功ノード必要数】と【書き込み成功ノード必要数】によって、読み込み性能重視にしたり、書き込み性能重視に変更することもできます。
Quorum Protocolに加えて、読み込まれたデータVersionの前後関係が判定できれば、ある程度の一貫性が保障されることになります。この前後関係の判定に利用するのがVector Clock(ベクタークロック)です。Vector Clockとは、論理クロックの１つで、NodeとCounterのペアの集合体です。通常のクロックを利用するためには、全Nodeの時計の同期を常時維持する必要があります。しかし、Node数が多くなると困難であるため、Vector Clockを利用することがあります。しかし、Cassandraでは読み込まれたデータVersionの前後関係を厳密に扱うことを諦め、VectorClockを利用していません。（参考：http://www.datastax.com/dev/blog/why-cassandra-doesnt-need-vector-clocks）
※ここで、ある程度の一貫性と書いている理由は、書き込みが違うNode経由にほぼ同時に起きた場合にVector Clock上で競合が起きてデータの前後判定が不可能になったり、書き込みと読み込み違うNode経由でほぼ同時に起きた場合にタイミング次第で違う値が返ってくることがあるからです。
最後に、レプリケーション数:3、書き込み成功必要数:2、読み込み成功必要数:2の時の、書き込みと読み込みの処理の流れを説明します。（ただし、VectorClockの説明を行う為、Cassandraでは導入されていないVectorClockが使われていた場合について説明します。）
書き込みを行う時は、最初にCassandraを構成するいずれかのNodeに接続し、書き込み要求の情報を渡します。接続された仲介Nodeは、保持しているVector Clockのカウントアップを行い、データの担当Node全てに対して、Vector Clockの情報を付与して、書き込み要求を投げます。書き込み先のNodeは、保持しているデータのVector Clockより渡されたVector Clockの方が後になると判定されたら、データを更新し、書き込み成功したことを仲介Nodeに返します。データの担当Nodeが落ちているなど、接続できない場合は、他のNodeに書き込み処理の復旧用データ(Hint)を書き込んでおきます。これにより、データの担当Nodeが復旧した時に、Hintをもらって書き込み処理を再現することができます。これはHinted Hand Off(ヒンティッド ハンドオフ)という仕組みです。
仲介Nodeは、帰ってきた書き込み成功数がQuorum Protocolとして設定した書き込み成功必要数に達した場合に、Clientに書き込み成功と返します。また、タイムアウトや競合によって書き込みないNodeが多く書き込み成功必要数に達しない場合は、Clientに失敗と返します。
読み込みを行う時も同様に、最初にCassandraを構成するいずれかのNodeに接続し、読み込み要求の情報を渡します。接続された仲介Nodeは、データの担当Node全てに対して、読み込み要求を投げます。読み込み先のNodeは、保持しているデータをVector Clockを含めて、仲介Nodeに返します。
仲介Nodeは、帰ってきた読み込み成功数がQuorum Protocolとして設定した読み込み成功必要数に達した場合に、Vector Colockの比較によって最新のデータが判明した場合は、読み込み成功として最新データを返します。（同時に古いデータを保持していたデータの担当Nodeに対して、最新データへの更新要求を出します。これは、Read Repair(リードリペア)という仕組みです。）Vector Colockの比較によって競合によって最新のデータが複数あると判断した場合は、読み込み失敗として複数の最新のデータを返します。また、タイムアウトなどにより読み込み成功必要数に達しない場合は、Clientに失敗と返します。
データの担当Node内での動作は、ほぼHBaseと同様で、メモリとWALとダンプファイル(HFileのこと)によってデータを取り扱います。唯一違う点は、ダンプファイルの保存先がHDFSの代わりにローカルのファイルシステムを使う点です。
以上のような仕組みから、タイミングなどによっては一貫性が保証されないところがあるものの、Masterといった特別なServerが存在せず、高い可用性を実現できています。
まとめ
HBaseとCassandraを例にして、KVSの仕組みを簡単に紹介してみました。KVSの設計思想の理解に役立てば幸いです。ちなみに私は、Cassandraの仕組みが好きですが、使うのはHBaseの方が好きです。
また、本記事で紹介している仕組みはコア機能の一部であり、実際には紹介したもの以外にも多くの機能があります。ご自身でぜひ高みへ上り詰めてください！