Hadoop_21_ZOOKEEPER.md

著者 の GitHub のソースコード：https://github.com/tomwhite/hadoop-book

ZooKeeper

Hadoop の分散コーディネーションサービスである ZooKeeper の紹介。

一般的に分散システムを構築するのは難しい。主な理由の一つに partial failure がある。

partial failure とはデータの送り元が、データが正しく送られたかどうかが分からないような状態を指す（通信中にネットワークに問題が生じた場合、送り主は受け手が正しくデータを受け取ったか分からない、など）。これは分散システムを構築する上で本質的に発生しうるもので、完全に解消したり隠蔽することはできない。

この partial failure を扱いやすくするツールとして ZooKeeper が開発されている。

ZooKeeper の特徴：

• ZooKeeper is simple
  • ファイルシステムとして見た場合、最小の機能しかもたない
• ZooKeeper is expressive
  • 分散キュー、分散ロック、リーダー選出など様々なアプリケーションが構築できる
• ZooKeeper is highly available
  • 複数台サーバで運用され、高可用性を担保する。ZooKeeper を利用することでアプリケーションが SPoF（単一障害点）を持つことを回避できる。
• ZooKeeper facilitates loosely coupled interactions
  • ??
• ZooKeeper is a library
  • ZooKeeper を利用した様々な「レシピ」があり、これを利用することができる。

1. Installing and Running ZooKeeper

Stand Alone モードについて：

• tar ボールをダウンロード・解凍
• 環境変数を設定
• 設定ファイル（ /etc/zookeeper/zoo.cfg ）を設定。最小は3つ
• 起動

（メモ）MacOS なら brew でも 3.4.6 がインストールできる。MacOS で tmux してると $ zkServer start できない場合があるので注意。

mntr コマンドの他に JMX, 3.5.0 以降であれば http://localhost/8080/`commands` でブラウザからも同等の内容が確認できる。

2. An Example

あるサービスを、複数台のサーバが提供する場合に…

• クライアントが接続できるサーバがどれなのか、 list を管理する必要がある
• サーバを取り除く場合、誰が取り除くのか？（取り除かれるべきサーバは not running のため、自分自身を取り除けない）

active distributed data structure として、外部イベントに反応してステータスを変更していく仕組みが必要 => ZooKeeper がこれを実装している。

2-1. Group Membership in ZooKeeper

ZooKeeper を high-availabilty filesystem と理解することができる：

• ただしファイルもディレクトリも持たない
• znode というものがあり、これがファイル・ディレクトリ両方の役割を果たす

2-2. Creating the Group

（コードを含むためPDF参照）

2-3. Joining a Group

（コードを含むためPDF参照）

2-4. Listing Members in a Group

（コードを含むためPDF参照）

2-5. Deleting a Group

（コードを含むためPDF参照）

3. The ZooKeeper Service

ZooKeeper の仕組みなどについての解説

3-1. Data Model

• znode と呼ばれる階層的なデータ構造

• 各 znode には ACL(AccessControlList) が付与でき、細かい制御が可能

• 大きいデータを対象としないため、各 znode あたり 1MB までしか保存できない

• Read/Write は atomic（中途半端なデータが読めたり、書き込みが中途半端に行われることは無い）

• データサイズが小さいため、追記処理はない（データを更新する場合は置き換える）

• znode はパスと呼ばれるスラッシュ区切りの文字列で参照される。ただし、UNIX のファイルシステムとは異なり必ず絶対パスで指定を行い、 ., .. などは意味を持たない。パスではないため、Java 上では java.lang.String オブジェクトで扱われる。

3-1-1. Ephemeral Nodes

znode には ephemeral / persistent の2種類がある。作成時に指定され、後から変更はできない。

ephemeral znode はセッションが切れたタイミングで削除される（セッションの定義は後述）。ephemeral znode は子を持てない（子が ephemeral だったとしてもダメ）。

ephemeral znode は作成したユーザーのセッションに依存するが、他のユーザーから参照することはできる。

3-1-2. Sequence numbers

末尾に連番を付与した sequencial znode を作成可能(P.634 参照)

3-1-3. Watches

イベントハンドラのようなもの。ZooKeeper サービスを呼ぶときに付与でき、他のサービスが呼ばれたときに発火する。 繰り返し発火させることはできない。

例えば、exists メソッドを呼ぶときに watch を付与しておけば、そのときに znode が存在しなかったとしても（= false が帰ってくる）、znode が作成されたタイミングで付与した watch がコールされる。P.627 の例を参照。

3-2. Operations

9つの基本的なオペレーション：

• create
  • Creates a znode (the parent znode must already exist)
• delete
  • Deletes a znode (the znode must not have any children)
• exists
  • Tests whether a znode exists and retrieves its metadata
• getACL, setACL
  • Gets/sets the ACL for a znode
• getChildren
  • Gets a list of the children of a znode
• getData,setData
  • Gets/sets the data associated with a znode
• sync
  • Synchronizes a client’s view of a znode with ZooKeeper
  • (Zookeeper は 大多数の サーバに値が反映されることを保証するため、見ている値が最新でない場合がある。その場合に sync メソッドを呼ぶと、クライアントを最新の情報に更新する。)

delete, setData（更新オペレーション）にはバージョンナンバーが必要で、exists メソッドから取得する。バージョンが一致しなければ（＝他のクライアントが更新を先に行ってしまったら）失敗となる。

3-2-1. Multiupdate

複数の更新をまとめて atomic にする。

3-2-2. APIs

コアライブラリのC, Javaと、拡張ライブラリの Perl, Python, REST API がある。

いずれも同期・非同期の API があり、構築するアプリケーションや処理内容によって使い分ける。例えば exists は以下の２種類：

public Stat exists(String path, Watcher watcher) throws KeeperException, InterruptedException

public void exists(String path, Watcher watcher, StatCallback cb, Object ctx)

StatCallback インタフェースは次のメソッドを持つ：

public void processResult(int rc, String path, Object ctx, Stat stat);

• int rc
  • 処理結果ステータス。0以外であればエラー。この場合、Stat オブジェクトは null になる

3-2-3. Watch triggers

読み込み処理の exists, getChildren, getData の3つには watcher を付与できる。watcher は書き込み処理の create, delete, setData で発火する。

（Table 21-3 を参照）。

3-2-4. ACLs

ZooKeeper 自体への認証と、各 znode へのアクセス制御の仕組みがある。

ZooKeeper 自体への認証設定はオプショナルで、digest(ID/PW), sasl(Kerberos), ip の3種類。

ACL は CREATE, READ, WRITE, DELETE, ADMIN の5種類ある。setACL できるのは ADMIN のみ。全ユーザー exists は叩ける。

3-3. Implementation

• Standalone モードと、Replicated モードの2種類がある
• Replicated モードは emsemble と呼ばれるクラスタで動く
  • ZooKeeper は emsemble の過半数が稼働している限りサービス提供し続ける
  • 「過半数」ということから、emsemble を組む台数は奇数にする

？？？（P.620）

Conceptually, ZooKeeper is very simple: all it has to do is ensure that every modification to the tree of znodes is replicated to a majority of the ensemble. If a minority of the machines fail, then a minimum of one machine will survive with the latest state. The other remaining replicas will eventually catch up with this state.

ZooKeeper はこれを実現するため、Zab というプロトコルを利用する（Paxos アルゴリズムに似るが異なる）：

• Phase 1: Leader election
  • emsemble 内で leader を選出する。残りのサーバは follower となる。このフェーズは過半数（か quorum） の followers が leader と同期を取れたら完了する。
• Phase 2: Atomic broadcast
  • 書き込み処理は全て leader へ転送され、leader から followers へ書き込み命令が broadcast される。
  • 過半数の followers の書き込みが完了したら leader は commit し、クライアントへ応答を返す
  • 2-phase commit に類似

書き込みはメモリより先にディスクに書き込む。読み込みはメモリからのみ行うため高速。（書き込んだ直後に読み込んでも最新のものは得られない？）

3-4. Consistency

ZooKeeper は以下の性質を担保する：

Sequential consistency

Updates from any particular client are applied in the order that they are sent. This means that if a client updates the znode z to the value a, and in a later operation, it updates z to the value b, then no client will ever see z with value a after it has seen it with value b (if no other updates are made to z).

Atomicity

Updates either succeed or fail. This means that if an update fails, no client will ever see it.

Single system image

A client will see the same view of the system, regardless of the server it connects to. This means that if a client connects to a new server during the same session, it will not see an older state of the system than the one it saw with the previous server. When a server fails and a client tries to connect to another in the ensemble, a server that is behind the one that failed will not accept connections from the client until it has caught up with the failed server.

Durability

Once an update has succeeded, it will persist and will not be undone. This means updates will survive server failures.

Timeliness

The lag in any client’s view of the system is bounded, so it will not be out of date by more than some multiple of tens of seconds. This means that rather than allow a client to see data that is very stale, a server will shut down, forcing the client to switch to a more up-to-date server.

3-5. Sessions

emsemble 内の1つのサーバとコネクションが確立すると、セッションが開始される。

• セッションにはタイムアウトがある。これはアプリケーション側で設定する。
• セッションはクライアントが ping を打って保つ（これは ZooKeeper client が自動で行っている）
• セッションが切れると ephemeral znodes が消える
• つないでいるサーバが（読み込みタイムアウト時などに）failover されることがあるが、この場合はセッションは引き継ぎ、ephemeral znodes も消えない。

3-5-1. Time

tick time オプションで、ZooKeeper の各アクションの基本単位時間を決める。多くの場合は 2000(msec=2秒)。

セッションタイムアウトは 2ticks 〜 20ticks 内で設定されなければならない。1tick=2000msec の場合、4秒〜40秒にあたる。

適切な値はアプリケーションの性質や JMX などのモニタリング値を見て定める。emsemble の台数が増えると、タイムアウトなどは長めに設定する必要がある。

3-6. States

CONNECTING, CONNECTED, CLOSED の3種類。各遷移で Watcher イベントを発火できる。

4. Building Applications with ZooKeeper

4-1. A Configuration Service

（PDF 参照）

4-2. The Resilient ZooKeeper Application

「分散コンピューティングの落とし穴」(https://ja.wikipedia.org/wiki/%E5%88%86%E6%95%A3%E3%82%B3%E3%83%B3%E3%83%94%E3%83%A5%E3%83%BC%E3%83%86%E3%82%A3%E3%83%B3%E3%82%B0%E3%81%AE%E8%90%BD%E3%81%A8%E3%81%97%E7%A9%B4) が述べている事項に、「ネットワークは信頼できる」というものがある。

ZooKeeper でネットワーク問題含め、エラーが生じたときにどのように対処できるかを検討する。

ZooKeeper の Java API は次の2つの例外をスローする：InterruptedException, KeeperException

4-2-1. InterruptedException

一般的に、スレッドに対して interrupt() が呼ばれることがあり、blocking methods をキャンセルすることができる。ZooKeeper もこれに習い、interrupt() が呼ばれると InterruptedException を投げるため、処理のキャンセルが可能。

これはエラーというよりオペレーションがキャンセルされたことを表すため、アプリケーションが処理を止めるように例外を伝達しなければならない。

4-2-2. KeeperException

ZooKeeper サーバとのコミュニケーションにエラーが生じた際に発生する。原因によりいくつかのサブクラスが存在する。例えば、存在しない znode にオペレーションを発行した場合、KeeperException.NoNodeException がスローされる。

KeeperException のサブクラスはエラーコードを持ち、合わせて handler へ渡される。例えば先ほどの例外のエラーコードは enum の KeeperException.Code.NONODE が定義している。

クライアントはエラーコードを確認するか、エラークラスをチェックすることで例外処理を行う。

KeeperException は種類によってさらに3種類に分類できる。

1) State exceptions

znode への処理に対し、znode のステートが不正な場合に発生する。

例

• setData, delete 時にバージョンナンバーが古い KeeperException.BadVersionException
• ephemeral znode に子nodeを追加しようとした KeeperExcep tion.NoChildrenForEphemeralsException

2) Recoverable exceptions

セッション内で回復できる例外を表す KeeperException.ConnectionLossException が該当。

コネクション自体はリトライすることで回復でき、その場合にはセッション情報は保持される。しかし、KeeperException.ConnectionLossException が発生した際に更新処理を行っていた場合は、それが成功したか失敗したかはクライアントは判断できない。

べき等な処理であればリトライを行い、そうでなければ、更新が成功したか失敗したかを判断できるようなパス・データの作り方をした方が良い。

3) Unrecoverable exceptions

セッションが切れたり、認証が失敗した場合。KeeperException.SessionExpiredException, KeeperException.AuthFailedException。

セッションはなくなり、ephemeral znode も消える。クライアントから改めてコネクションを張り、セッションを新しく作り直す必要がある。

4-2-3. A reliable configuration service

（P.632 参照）

4-3. A Lock Service

複数プロセス間で排他制御を行う場合、任意の時点で、ロックを持つのはただ1つのプロセスとなるようにしたい。分散システムでリーダーを選ぶ場合などに利用される（ZooKeeper の leader 選出の話ではないことに注意）。

ZooKeeper でこれを実現する場合、以下の様にする：

• ロック用の znode を作成する（例えば /leader）
• ロック取得したいクライアントは、/leader の下に sequential な ephemeral znode を作成する。
• sequential number がもっとも 小さい クライアントがロックを獲得する。
  • クライアントA が /leader/lock-1、クライアントB が /leader/lock-2　の場合、クライアントA がロックを獲得する。
• ロックの解放は ephemeral znode を自ら削除するか、クライアントプロセスが死ぬ際にセッションが解放されて自動的に削除される。
  • /leader の getChildren に watcher を登録しておくことで、削除されたときに検知でき、ロック獲得時の処理を記述できる

以下、この処理の問題点と回避方法が紹介されている。

4-3-1. The herd effect

上で説明した方法だと、 /leader の下の ephemeral znode が削除されるたびに、ロックを獲得しようとしている全クライアントの watcher が発火する。これはクライアント数が増えるとトラフィックを圧迫する原因になる（the herd effect）。

これを回避するためには、子nodeの削除を検知するのではなく、「直前のロックファイルが削除された」という特定の場合に対して watcher を設定するようにする。上の例では、lock-2 は /leader の子node削除に対して watcher を張るのではなく、/leader/lock-1 が削除されたことに対して watcher を設定する。

4-3-2. Recoverable exceptions

create 中にコネクションが�切れてしまった場合、ephemeral znode が作成されたかどうかが判断できない。このため、deadlock の原因となってしまう。

パス名にセッションID を入れることで回避する。コネクションが切れた場合、自分のセッションIDを持つ znode ができているかどうかを確認する。

lock-<session_id>-<sequence>

4-3-3. Unrecoverable exceptions

セッションが切れてしまった場合は ephemeral znode が消えるため、アプリケーション側で再度ロック獲得をする必要がある。これは ZooKeeper の管掌ではない。

4-3-4. Implementation

ZooKeeper comes with a production-quality lock imple‐ mentation in Java called WriteLock that is very easy for clients to use.

4-4. More Distributed Data Structures and Protocols

（ZooKeeper が有効なアプリケーション例やツールの紹介）

5. ZooKeeper in Production

5-1. Resilience and Performance

マシン・ネットワークの故障を考慮し、サーバはラック・電源・スイッチに対して分散して配置するのがよい。

低レイテンシを必要とするサービスは同じデータセンタ内に emsemble を構築し、そうではない場合は複数のデータセンターにまたいで emsemble を構築するのがよい（リーダーの選出や、分散ロックに ZooKeeper を利用する場合などが該当）。

高可用性を保つため、ZooKeeper マシンは ZooKeeper だけ動かすべき。

スナップショットとトランザクションログは異なる物理ディスクに書き込む。dataDir だけを設定するとどちらもこのディレクトリに書き込まれるが、dataLogDir を設定するとトランザクションログはこちらに書き込まれる。こうすることで、ログを追記するときにディスクを seek する手間がなくなる。全ての書き込みは leader から行われるため、データの書き込みは可能な限り早くしたい。

SWAP を使うとパフォーマンスに影響がでるため、Java Heap Size を調整して SWAP を利用しないようにする。

5-2. Configuration

（P.639）