HDFSの現状と今後 (その1)


まず、Apache Hadoop 3.0.0リリースおめでとうございます!

* 公式の記事: https://blogs.apache.org/foundation/entry/the-apache-software-foundation-announces26

ここ最近の大きめのHDFSの機能について、いくつか語ってみたいと思います

* 3.0.0でリリースされた機能

  * Erasure Coding
  * Router based federation (2.9.0にもbackportされた)

* 3.1.0に入ることが確定した機能 (trunkにマージ済)

  * Tiered storage

* 3.1.0に入るかわからない機能 (trunkにマージ前)

  * Ozone: Object store for HDFS

## Erasure Coding

JIRA: https://issues.apache.org/jira/browse/HDFS-7285

Apache Hadoop 3.0.0の中でも目玉機能はHDFS Erasure Coding(以下、EC)かと思われます。これにより、耐障害性はそのまま、実データ量を半分程度に抑えられるようになりました。Facebookは、HDFSにECの仕組みが実装されないから独自にforkした、という話を聞いたことがありますが、ついにApache Hadoopにも実装されたということになります。ここでFacebookのエンジニアがHDFSの開発に戻ってくるとは考えにくいですが。

* HBaseにはFacebook用のbranchがあったけど、公式のリリースブランチに戻ってきた、という話はある
* そういえば、ほとんど米Yahoo!専用になっていたApache Hadoop 0.23系がありましたね

### ECが活かせる環境

ECには、大規模環境においては非常に有効ですが、小規模な環境においては、あまり恩恵を受けられません。例えば、RS(10,4)のエンコーディングを利用する場合、DataNodeは最低14台必要ですし(RS(6,3)でも9台)、故障時のレプリケーション先のことも考えると、最低でも20台くらいは必要になるかと思います。また、20台程度の規模ではECによるディスク容量の節約による恩恵を受けづらいかと思います。設定や運用の手間が(若干ですが)増え、トラブルシューティング時の解析が困難になるのは目に見えているので、その分のコストがECによるディスク容量の削減でペイできないと思われます。

### なぜ今になってECなのか

以下の4点かなぁと思います。(他にもあると思います)

* そもそもHDFSのような大規模分散ストレージにECを適用する、という論文が出だしたのが2012年くらいになってからだった
* [ISA-L](https://github.com/01org/isa-l)がないとエンコード/デコードに時間がかかって実用的でない。これが出始めたのが2014年あたり。

  * HDFSの暗号化について、AES-NIがないとお話にならない、という話と同じ。

* 10G NICが当たり前になるなど、昔と比べてNWが強化され、HDFS上のデータを処理するにあたってデータローカリティを必ずしも意識する必要がなくなってきた (現状のHDFS-EC(striping layout)にはデータローカリティがない)
  * データローカリティがあったほうがよいのは間違いないが、10Gのネットワークごしなら充分、というパターンは普通に考えられる

* HDFSが処理用途だけではなく、アーカイブ用途にも使われることが増えてきた

  * Archival Storageという、アーカイブ用のデータを分けて保存するための機能がHadoop 2.6で追加された
  * 過去に書いた解説: https://www.slideshare.net/hadoopxnttdata/hdfs-new-feature-2015-cwt2015/7

開発が始まったのが2014年なので、ISA-Lが出た頃と一致してるかなと思います。

### ECのトレードオフ

長所はディスク容量が削減されることで、誰にとってもわかりやすいのですが、短所についてはあまりわかりやすくないかと思うので、いくつか紹介します。

#### 故障時の再構築が遅い

通常のレプリケーションだと、故障した分だけそのまま再レプリケーションすればよいので、ちょうど故障した分だけ読み込めばよいです。ただし、Erasure Coding(RS(10,4))の場合は、1つのdata blockもしくはparity blockを再構築するために、10個のblockを読む必要があります。なんと10倍です。

これを少し減らすために、Hitchhiker XORというアルゴリズムが提案され、Apache Hadoopでも実装されています。

* JIRA: https://issues.apache.org/jira/browse/HADOOP-11828
* 最近書いた解説: https://www.slideshare.net/hadoopxnttdata/introduction-of-apache-hadoop-security-isssue-and-hitchhiker-erasure-coding-algorithm/29

#### データローカリティがない

Erasure Codingにおいて、データの配置パターンが2種類考えられました。

* striping layout
* contiguous layout

  * JIRA: https://issues.apache.org/jira/browse/HDFS-8030

このうち、データローカリティのない、striping layoutが現状利用可能です。striping layoutとcontiguous layoutの違いについてこれから説明していきます。

![Screen Shot 2017-12-23 at 16.44.14.png](https://qiita-image-store.s3.amazonaws.com/0/32443/4a46781d-eff2-90b7-38ba-afe27df57be4.png "Screen Shot 2017-12-23 at 16.44.14.png")


striping layoutでは、上図([Design Document](https://issues.apache.org/jira/secure/attachment/12697210/HDFSErasureCodingDesign-20150206.pdf)より抜粋)のようにデータを1024KB単位のcellに分けて、複数のDataNodeに配置させています。逆に、contiguous layoutでは、このcellが128MB程度と大きくなります(HDFSのブロックサイズと同じになります)

* デフォルトのcell sizeは64KBから1024KBに変更になりました
* JIRA: https://issues.apache.org/jira/browse/HDFS-12303

ここで、parityも1つあたり1024KBとなります。striping layoutでは、このcell sizeが比較的小さいことにより、書き込むデータ量を比較的少なく抑えられることが特長です。例えばRS(10,4)で10MBのファイルを書き込む場合、cell sizeが1024KBだと10個の1MBのdataと4個の1MBのparityに分かれます。書き込む合計容量は14MB。ところが、cell sizeが仮に5MBだと、2個の5MBのdataと8個の0Bのdataと4個の5MBのparityに分かれます。書き込む合計容量は30MBです。これではECにするメリットがありません。HDFS-ECの開発者が可能な範囲でHDFSの利用状況について統計を取ったところ、このような小さめのファイルのほうが多かったので、先にstriping layoutが実装されることになりました。

ただし、striping layoutには、あるファイルの0B〜128MBまでを処理したいと考えたとき、複数のDataNodeからデータを取ってくる必要があるため、データローカリティは存在しない事に注意が必要です。

このあたりの話についてはもっと良い記事があるので読んでみてください。HDFS-ECの知識がより深まると思います。

* Cloudelaの開発者による記事: https://blog.cloudera.co.jp/introduction-to-hdfs-erasure-coding-in-apache-hadoop-c13910ba15d4
* Yahoo! Japanの開発者による記事: https://techblog.yahoo.co.jp/infrastructure/hdfs_erasure_coding/

#### データの読み書きが遅くなる?

実は、遅くなるとは限りません。エンコード/デコードの分、CPU使用量は増えますが、通常のレプリケーションだとパイプラインを作ってクライアントから1台のデータノードにデータを書き、そこから他のDataNodeに伝搬させていくのに対し、ECでは複数のDataNodeにblockを並列に書き込むようになっています(読み込みも並列になる)。例えば、書き込み先のHDDの速度がボトルネックになっているような場合では、ECのほうが速く書き込みができるかもしれません。

また、エンコード/デコードに時間がかかる件は、ISA-Lの利用によってかなり改善されています。

#### いくつかの機能が制限されている

現状、hflush, hsync, appendはサポートしていません。これらの機能が必要な場合は、通常のレプリケーションを利用してください。

## Router based federation

JIRA: https://issues.apache.org/jira/browse/HDFS-10467

ざっくり言うと、新しいHDFS NameNode Federationの実装です。HDFSは、DataNodeを増やせば増やすだけ容量が増え、I/Oの総スループットも向上するというスケールアウト型の設計になっているのですが、DataNodeの台数が4桁規模になってくると、マスタとなるNameNodeがボトルネックとなるケースがあります。このボトルネックをNameNodeを増やすことで解決するというのがNameNode Federationです。この機能自体はApache Hadoop 2系で既に実装され、そのときは、以下のような形で実装されていました。

![Screen Shot 2017-12-24 at 16.25.26.png](https://qiita-image-store.s3.amazonaws.com/0/32443/f4ba858c-24e9-6288-9ba8-56cfc3abb722.png "Screen Shot 2017-12-24 at 16.25.26.png")

出典: http://hadoop.apache.org/docs/r3.0.0/hadoop-project-dist/hadoop-hdfs/Federation.html

ここで、複数のNameNodeに対してそれぞれDataNodeのグループがいるのではなく、各DataNodeは複数のNameNodeと通信することになります。また、あるNameNodeを1つのHDFSのディレクトリにマウントしたように見せる仕組み(ViewFileSystem)と合わせることで、1つのHDFSとして見せることができます。

![Screen Shot 2017-12-24 at 16.29.18.png](https://qiita-image-store.s3.amazonaws.com/0/32443/a6293e2a-8ea3-c664-2ba6-09128cb0b3f2.png "Screen Shot 2017-12-24 at 16.29.18.png")

出典: http://hadoop.apache.org/docs/r3.0.0/hadoop-project-dist/hadoop-hdfs/ViewFs.html

しかしながら、この実装には以下の問題がありました。

* どのNameNodeをどのディレクトリにマウントさせるか考える必要がある

  * 各NameNodeに負荷が分散されるような設計にしないといけない
  * 間違えた場合のやり直しが困難

* どのNameNodeをどのディレクトリにマウントするかという設定をHDFSの全クライアントに配布する必要がある

その問題を解決するために実装されたのが、Router based federationです。実は、このFederationが実装される前にYARN ResourceManagerでrouter based federationが実装され([YARN-2915](https://issues.apache.org/jira/browse/YARN-2915))、それがHDFS NameNode側にも移植された、という経緯で実装されています。以下に概念図を示します。(出典: http://hadoop.apache.org/docs/r3.0.0/hadoop-project-dist/hadoop-hdfs/HDFSRouterFederation.html)

![Screen Shot 2017-12-24 at 16.49.14.png](https://qiita-image-store.s3.amazonaws.com/0/32443/ceec2ede-ec93-a5dc-09d2-e0e12dc13cb9.png "Screen Shot 2017-12-24 at 16.49.14.png")

ここで、'R'は'Router'です。全体を統括するState Storeが存在し、各NameNodeとRouterは同居しています。クライアントがHDFS上のあるファイルを読みたいときは、まず、Routerにリクエストを投げます。すると、RouterはState Storeと通信し、どのNameNodeがそのファイルを管理しているか問い合わせます。その後、RouterはそのNameNodeに対し、どのDataNodeがそのファイルのblockを保持しているか問い合わせ、クライアントにその結果を返す、という形になります。

State Storeは、どのNameNodeがどのディレクトリを管理しているかというrouting tableを保持しており、adminはAPI経由でrouting tableを設定することになります。そのため、どのNameNodeをどのディレクトリにマウントさせるか考える必要があることには変わりません、しかしながら、マウントの設定をクライアントに配布する必要はなくなります。リバランスについては、議論はされていますがまだ実装されていないように見えます。リバランスは是非欲しい機能なので、これからの開発に期待です。(とは言っても、NameNode Federationが必要になるほどの大規模クラスタって世界にいくつあるんだ... って感じですが)

## まとめ

ここまで、Erasure CodingとRouter based federationについて紹介しました。長くなってきたので、Tiered StorageとOzoneについてはまた今度にします。そもそも、Tiered Storageについてはまだdesign documentをしっかりと読み込めていない...