Hadoopシリーズ-HDFS

Hadoop シリーズ - HDFS#

一、HDFS の起動プロセス#

ファイルのメタ情報を読み込む
ログファイルを読み込む
チェックポイントを設定する
セーフモードに入る。目的は：データブロックのレプリケーション率、冗長性が要件を満たしているかを確認すること

二、HDFS の運用メカニズム#

ユーザーファイルはブロックに分割され、複数の DataNode サーバーに保存され、各ファイルはクラスター全体に複数のレプリカを持ち、データの安全性を向上させることができる

三、基本アーキテクチャ#

NameNode: 全体のファイルシステムの管理ノードで、ファイルがどのようにブロックデータに分割されるかを記録し、これらのデータブロックのストレージ情報を記録する
1. Fsimage: メタデータがハードディスク上に保存されるイメージファイル
2. edits: システム操作のログ記録ファイル
3. Fstime: 最後のチェックポイントの時間を保存する
4. seen_txid: 最後の edits の番号
5. version
Secondary NameNode: 補助バックグラウンドプログラム（NameNode の障害復旧ノードではない）、NameNode と通信し、定期的に HDFS メタデータのスナップショットを保存する
DataNode: データノードで、HDFS データブロックをローカルファイルシステムに読み書きする

HDFS が小さなファイルの保存に適さない理由は、各ファイルがメタ情報を生成し、小さなファイルが多くなるとメタ情報も増え、NameNode に負担をかけるからである

フェデレーション HDFS#

各 NameNode は 1 つの名前空間を維持し、異なる NameNode 間の名前空間は相互に独立している。各 DataNode は各 NameNode に登録する必要がある

複数の NN が 1 つのクラスター DN のストレージリソースを共有し、各 NN は独自にサービスを提供できる。
各 NN はストレージプールを定義し、独自の ID を持ち、各 DN はすべてのストレージプールにストレージを提供する。
DN はストレージプール ID に基づいて対応する NN にブロック情報を報告し、同時に DN はすべての NN にローカルストレージの可用リソース状況を報告する。
クライアントが複数の NN 上のリソースに便利にアクセスする必要がある場合、クライアントマウントテーブルを使用して異なるディレクトリを異なる NN にマッピングすることができるが、NN 上には対応するディレクトリが存在する必要がある

四、NameNode の作業メカニズム#

NameNode 作業メカニズム.png
HDFS の読み書き -> ログをロール -> SN が NN にチェックポイントが必要かどうかを尋ねる -> 時間が来た（60 分）または edits データが満杯になった場合にチェックポイントをトリガー -> SN がチェックポイントの実行をリクエスト -> NN が edits ファイルと fsimage ファイルを SN にコピー -> SN がマージした fsimage を NN に同期する

五、DataNode の作業メカニズム#

DataNode ノード作業メカニズム.jpeg

DataNode は起動時に NN に登録し、その後定期的にブロックデータの情報を報告する
NN と DataNode の間はハートビート検出メカニズムを介して、3 秒ごとに 1 回ハートビートを送信し、10 分以上ハートビートが受信されない場合はノードが使用不可と見なされる

六、HDFS のデータ読み取りプロセス#

HDFS 読み取りプロセス.png

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path file = new Path("demo.txt");
FSDataInputStream inStream = fs.open(file);
String data = inStream.readUTF();
System.out.println(data);
inStream.close();

クライアントはFileSystemオブジェクトを初期化し、open()メソッドを呼び出してDistributedFileSystemオブジェクトを取得する
DistributedFileSystem は RPC を介して NN に最初のバッチのブロックロケーションを取得するようリクエストする
前の 2 つのステップで FSDataInputStream が生成され、このオブジェクトは DFSInputStream オブジェクトにラップされる
クライアントはread()メソッドを呼び出し、DFSInputStream はクライアントに最も近い DataNode を見つけて接続し、このファイルの最初のデータブロックの読み取りを開始する。データは DataNode からクライアントに転送される
最初のデータブロックの読み取りが完了すると、DFSInputStream は接続を閉じ、次のデータブロックの DataNode ノードに接続してデータ転送を続ける
データを読み取る際に DFSInputStream と DataNode データノードの通信に異常が発生した場合、次の DataNode ノードに接続を試み、そのデータブロックを含む DataNode ノードを記録し、残りのブロックを読み取る際にはその DataNode をスキップする
クライアントがデータの読み取りを完了した後、close()メソッドを呼び出して接続を閉じ、データをローカルファイルシステムに書き込む

七、HDFS のデータ書き込みプロセス#

HDFS 書き込みプロセス.png

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path file = new Path("demo.txt");
FSDataOutputStream outStream = fs.create(file);
out.write("Welcome to HDFS Java API !!!".getBytes("UTF-8"));
outStream.close();

クライアントがcreateメソッドを呼び出し、ファイル出力ストリームDFSDataOutputStreamオブジェクトを作成し、NN にファイルのアップロードをリクエストする
DistributedFileSystem は RPC を介して NN にブロックに関連付けられていない新しいファイルエントリを作成するように呼び出す。作成前に NN はファイルが存在するか、作成する権限があるかを検証し、成功した場合は操作を EditLog（WAL，write ahead log）に書き込み、出力ストリームオブジェクトを返す。そうでない場合は IO 例外をスローする
クライアントは最初にファイルを分割する。たとえば、ブロックサイズが 128M の場合、300M のファイルは 3 つのブロックに分割され、2 つの 128M と 1 つの 44M が作成され、NN にどの DataNode サーバーにブロックを転送するかをリクエストする
NN は書き込み可能な DN ノード情報を返し、クライアントと NameNode が割り当てた複数の DataNode がパイプラインを構成して接続を確立し、クライアント側が出力ストリームオブジェクトにデータを書き込む
FSDataOutputStream オブジェクトを介して DN にデータを書き込み、これらのデータは小さなパケットに分割され、データキューに並べられる。クライアントが最初の DataNode にパケットを書き込むと、そのパケットはパイプライン内で次の DataNode、次の DataNode に直接渡される。ブロック全体またはファイル全体を書き込んでから次に配信するのではない
各 DN が 1 つのブロックを書き終えた後、クライアントにack確認情報を応答する。注意すべきは、各パケットを書き終えた後に確認情報を返すわけではない
クライアントがデータの書き込みを完了した後、closeメソッドを呼び出してストリームを閉じる

補足:

クライアントが書き込み操作を実行した後、書き終えたブロックのみが可視であり、書き込み中のブロックはクライアントには見えない。syncメソッドを呼び出すことで、クライアントはそのファイルの書き込み操作がすべて完了したことを確認できる。クライアントがcloseメソッドを呼び出すと、デフォルトでsyncメソッドが呼び出される。手動で呼び出す必要があるかどうかは、プログラムの必要に応じてデータの堅牢性とスループットの間のトレードオフによる

書き込み中のエラーハンドリング:#

書き込みプロセス中に特定の DN レプリカノードにエラーが発生した場合:#

まずパイプラインが閉じられる
すでにパイプラインに送信されたが確認を受け取っていないパケットをデータキューに書き戻し、データの損失を防ぐ
次に、現在正常に動作している DN データノードに新しいバージョン番号が付与される（NameNode のリース情報を利用して最新のタイムスタンプバージョンを取得できる）。故障ノードが復旧した場合、バージョン情報が一致しないため、故障 DataNode は削除される
故障ノードを削除し、正常な DN データノードを選択してパイプラインを再構築し、データの再書き込みを開始する
レプリカが不足している場合、NN は他の DN ノードで新しいレプリカを作成する

書き込みプロセス中にクライアントがクラッシュした場合の解決策:#

データの書き込み中にクライアントが異常終了した場合、同一ブロックデータの異なるレプリカが不一致の状態になる可能性がある。特定のレプリカを主データノードとして選択し、他のデータノードを調整し、NN はリースメカニズムを通じてすべての DN レプリカノードがこのデータブロック情報を持つ最小ブロック長を見つけ、そのデータブロックを最小長に復元する

詳細は参照: HDFS 復元プロセス 1

八、HDFS レプリケーションメカニズム#

最初のレプリカ：アップロードノードが DN であれば、そのノードにアップロードされる；アップロードノードが NN であれば、DN がランダムに選択される
2 番目のレプリカ：異なるラックの DN に配置される
3 番目のレプリカ：2 番目のレプリカと同じラックの異なる DN に配置される

九、HDFS セーフモード#

セーフモードは HDFS の作業状態の一つで、セーフモードにある状態では、クライアントにファイルの読み取り専用ビューのみを提供し、名前空間の変更を受け付けない

NN は起動時に最初に fsimage をメモリにロードし、その後 edits ログ記録の操作を実行する。メモリ内でファイルシステムメタデータのマッピングが成功裏に確立された後、新しい fsimage と空の edits 編集ログファイルを作成する。この時点で NN はセーフモードで動作している
この段階で NN は DN から情報を収集し、各ファイルのデータブロックを統計し、最小レプリカ条件を満たすことが確認されると、一定の割合のデータブロックが最小レプリカ数に達すると、セーフモードを終了する。満たされない場合は、DN に対してレプリカ数が不足しているデータブロックの複製を行うように手配し、最小レプリカ数に達するまで続ける
フォーマットされたばかりの HDFS を起動すると、データブロックがないためセーフモードには入らない

セーフモードを終了する：hdfs namenode -safemode leave

十、HA 高可用メカニズム#

参照: Hadoop NameNode 高可用 (High Availability) 実現解析

基本アーキテクチャの実現#

HDFS の HA 高可用性は zk によって保証され、基本アーキテクチャ:

アクティブNameNodeとスタンバイNameNode: 主従 NameNode ノードで、アクティブな主 NameNode ノードのみが外部にサービスを提供する
共有ストレージシステム: NN の運用中に生成されたメタデータを保存し、主従 NN は共有ストレージシステムを介してメタデータの同期を実現し、主従切り替えを行う際には新しい NN がメタデータが完全に同期されていることを確認してから外部にサービスを提供できる
主従切り替えコントローラー ZKFailoverController: ZKFC は独立したプロセスとして実行され、NN の健康状態を迅速に監視できる。主 NN に障害が発生した場合、zk クラスターを利用して主従自動選挙切り替えを実現する
DataNode： DataNode は主従 NN にデータブロック情報を同時にアップロードし、HDFS のデータブロックと DataNode 間のマッピング関係を同期させる必要がある
Zookeeper クラスター：主従切り替えコントローラーに主従選挙のサポートを提供する

主従切り替えの実現#

参照: Hadoop NameNode 高可用 (High Availability) 実現解析

NameNode の主従切り替えは主にZKFailoverController、HealthMonitor、ActiveStandbyElectorの 3 つのコンポーネントが協調して実現する：

主従切り替えコントローラー ZKFailoverController が起動すると、HealthMonitor と ActiveStandbyElector の 2 つの主要な内部コンポーネントが作成され、ZKFailoverController は HealthMonitor と ActiveStandbyElector にそれぞれのコールバックメソッドを登録する。
HealthMonitor は主に NameNode の健康状態を検出し、NameNode の状態が変化した場合、ZKFailoverController の対応するメソッドをコールバックして自動的な主従選挙を行う。
ActiveStandbyElector は主に自動的な主従選挙を完了し、内部に Zookeeper の処理ロジックを封装している。Zookeeper の主従選挙が完了すると、ZKFailoverController の対応するメソッドをコールバックして NameNode の主従状態を切り替える

図のように、プロセス分析:

HealthMonitor が初期化を完了すると、内部スレッドが定期的に対応する NameNode の HAServiceProtocol RPC インターフェースのメソッドを呼び出して、NameNode の健康状態を検出する
NN の状態に変化が検出されると、ZKFailoverController の対応するメソッドがコールバックされ、処理が行われる
ZKFailoverController が主従切り替えが必要であることを検出すると、ActiveStandbyElector を使用して処理を行う
ActiveStandbyElector が ZK と対話して自動選挙を完了し、ZKFailoverController の対応するメソッドをコールバックして現在の NN を通知する
ZKFailoverController が対応する NameNode の HAServiceProtocol RPC インターフェースのメソッドを呼び出して、NameNode をアクティブ状態またはスタンバイ状態に切り替える