Preferred Networks におけるHadoop

Area

Cluster / Networking

Tag

# Cluster

Kota Uenishi

Engineer

Preferred Networks （以下PFN）では、「現実世界を計算可能にする」「全てのひとにロボットを」という目標のもと、機械学習を始めとしたあらゆる計算技術を用いて研究開発に日々取り組んでいます。その過程では必ずといっていいほど、データの保存や読み出しが必要になります。ここでは、我々がどのようにデータ管理をしているか、また、その過程でどのようにHadoopを利用しているかについて紹介したいと思います。

MN-J storage server photo
写真: PFNカラーリングのストレージサーバー

Hadoop導入の経緯

Hadoopが多く利用されているようなログ分析や、エンタープライズ向けのETL処理やデータサイエンスに比べると、深層学習の分野でよく利用されているような規模のデータは比較的小さいです。よく画像認識のベンチマークとされるImageNetの2012年のコンペで利用されたデータセットは200GB程度です。これはそもそも計算タスクが異なり、同じデータ量に対して必要とされる計算量が異なるためです。

PFNでHDFSのような分散ストレージが必要とされる場面は、世間に比べるとかなり遅れてやってきました。個別のデータセットはそこまで大きくなくとも、データセットの種類は増えていきます。実験をやればやるほど実験結果はたまっていきます。ログやETLに比べると、個別のデータは粒度は荒くサイズも大きいですが、フラットなファイルとディレクトリが多くを占めており、その構成やフォーマットは多種多様です。これに対して、PFNでは昔からストレージサーバーを何台か用意し、NFSを経由してストレージサーバー上にデータを読み書きするという形態でデータを保存・利用してきました。この方法は多くのメリットがあります。

I/O性能: Linuxカーネルに密結合した形で多くの機能が実装されており、ハードウェアが許す限り高い性能が得られる
可搬性: Linux VFS を通じてファイルシステムにマウントされPOSIX互換なAPIが提供されているため、さまざまなアプリケーションが動作する
利便性: ローカルファイルシステムと同様の利用方法であるため、通常のLinux利用の延長ですぐに使い始められる
運用性: インターネット上に多くの情報があり、故障しなければという条件つきではあるが運用しやすい

このような形態が特に楽だったのは、幸い、我々にとっては可用性やデータ信頼性の要件は低かったためです。システム構成を変更したいとかデータロスに備えたいといった要求に対しては、単純に計画停止やバックアップで済んでいました。

ところが、メリットが多い反面いくつかのデメリットもあり、いくつかの側面では運用が限界に達していました。特に拡張性の面で我々は苦労し、ストレージサーバーの残り容量が急に少なくなってパフォーマンスが極端に落ちたりしていました。かといって、学習用のデータや、実験結果を消してディスクを空けるというのは、一時的なワークアラウンドにはなりますが、データの所有者に手間をとらせることになります。結果的に、ストレージサーバーの残り容量が少なくなってきたタイミングで、より大きいストレージサーバーを導入して利用者が移動する、といったことが繰り返されてきました。新しいサーバーを導入するたびに少しずつ運用などの工夫はしてみたものの、データが増え続けることに変わりはなく、利用者に対して「新しいストレージサーバーが導入されたので引っ越しましょう」というアナウンスを繰り返すことになります。

システムの運用コストはサーバーの数ではなくサーバーの種類に比例するとはいいますが、筆者の個人的な経験によれば、管理方法がまちまちなステートフルなサーバーについてはこれは成立しません。上記のような経緯で導入されたストレージサーバーは導入世代によって少しづつ仕様や利用形態が異なり、それが一台増える毎に運用コストは増えていくという状態でした。

このような状態を打破するために、以下のような要件で、容量だけでもスケールするようなストレージシステムを模索しました。

ハードウェアを追加するだけで今後数年分の容量を拡張できること
ハードウェアを追加してもI/O性能が同様にスケールすること: 大規模深層学習のIO負荷に耐えられること
POSIXと同等か、それ以上のファイルの権限管理ができること
運用が簡単で、専任の担当が必要ないこと（それまでもストレージサーバーの運用に専任はいなかった
故障に対してもある程度ロバストで手間がかからないこと
プロプライエタリなライセンスでないこと

特に4番目の運用性の要件が厳しく、なかなか我々の用途に合ったシステムを見つけることはできませんでした。いくつかシステムを検討しましたが、最終的には Cloudera Manager という管理ツールが提供されているHadoop (具体的には、CDHというディストリビューション）を選択しました。このツールは無償で提供されており、運用上の我々の課題を大きく解決してくれました。複雑なシステム構築スクリプトや、ログ管理機構を書き上げることなくすぐに構築を終わらせ、運用にまで入ることができました。
Hadoopシステム内のプロパティ変更やログ検索、正常性の確認、アラートの表示、性能に関するメトリクスの表示、ノードの再起動やシステムのアップデートなど、ありとあらゆるシステム管理操作がCloudera Managerひとつで解決します。これがなければ、例え小さくてもHadoopクラスタを片手間で運用はとてもできなかったでしょう。

他にもいくつか候補を検討しました。たとえば、当時からCephはコミュニティが成長しており、我々の要件を期待以上に満たしていましたが、多くの部分がカーネル空間で動作することは我々のスキルと労力の範囲を超えており難しいと判断しました。
AWSやGCPのクラウドストレージは、上記の要件をほぼ完全に満たしますが、ひとつだけどうしても許容できない点がありました。それは、PFNのスーパーコンピュータとの接続性の問題でした。我々の計算資源は会社の競争力そのものであり、MNシリーズを自社で構築することで優位性を維持しています[2]。計算機にデータをロードしたり、計算結果を書き込むときのスループットやレイテンシはある程度の性能とコストパフォーマンスが必要です。

利用例1: 大規模な分散深層学習

学習用のデータセットは、公開されているものだけでも非常に多種多様なものがあります。深層モデルの学習はほどんどがSGDベースのものであり、ストレージからみると学習時のアクセスパターンは基本的にはランダムリードになります[1]。学習データを全て単一のストレージサーバーに置いて利用すると、どんな学習でも必ずストレージサーバーにアクセスすることになり、ストレージサーバーでの seek に大きな負荷がかかることになります。特に、分散深層学習ではひとつのデータセットを複数のノードが一斉に読み出すことになり、単位時間あたりのデータ読み出し量はより多くしなければなりません。
そこで、HDFSのようにスピンドルの数を増やせる構成にしておけば、多少のseekがあっても、IO帯域に余裕がうまれます。また、HDFSではメタデータが全て始めからNameNodeのメモリ上に保存されていますから、ディスク上に保存されている inode に対してランダムアクセスするよりも効率がよいです。学習時のデータのキャッシュ方法など、実際の利用例については [1] のスライドで解説しています。

利用例2: 学習用のデータセット管理

PFNでは、実際のオブジェクトをさまざまな角度から撮影し、それをもとに学習用のデータセットを作成しています。多様なタスクに対してデータセットをこのデータベースからオブジェクトを選んで構築して、実際のモデルに学習させて利用します。

ひとつのオブジェクトにつき複数枚の画像と、さまざまな付属データがあります。これはフラットなファイルとしてHDFS上に保存するには小さすぎます。そのため、これを HBase に保存しています。画像の保存にはMOBを利用しています[3]。実際のデータは数TB程度で、データ量としては多くはないですが、アノテーション済みのデータは重要で、今後も増やしていく必要があります。そのため、拡張性の高いバックエンドを利用しました。

このシステムで課題だったのは、 Python から HBase へのアクセスの難しさです。HBase へのアクセスにはさまざまなアダプタが用意されていますが、ここでは Thrift Server を経由する方法を選びました。 HappyBase を使った中継サーバーをKubernetes上に立ち上げ、それに対してPythonからアクセスしています。このサーバーはさまざまなプロトコルに対応しており、 fluentd で画像を保存することもできるようになっています。クライアントから HBase に実際に保存するまでは多くのコンポーネントを中継しており無駄が多いようにもみえますが、いずれも基本的にはステートレスなシステムであり、必要に応じて増やすことができます。

利用例3: 学習中のスナップショット保存

社内のある研究開発プロジェクトで、高解像度の動画を crop することなく学習するものがありました。もともとの解像度が大きいため、画像認識のモデルもパラメータが増えてサイズが大きなものになり、ディスクに保存したサイズでだいたい数百MB程度になりました。一方で学習自体も難しく精度の収束が見えにくかったため、 epoch が終わる度にモデルのスナップショットを保存する必要がありました。モデルがいくら保存されても大丈夫なように、HDFSに直接モデルを保存するプログラムにしました。これによって、ストレージサーバーの残り容量を気にすることなく実験をいくらでも並列で実行することができます。

ピーク時には多くの学習タスクを並列で動かしていたため、一日あたり数千から数万個のモデルのスナップショットが保存されていました。現在、この機能は PFN の PyTorch向けライブラリ pytorch-pfn-extras, pfio で利用することができます [5, 6, 7] 。

運用

最初のHadoopクラスタがPFNで稼働し始めたのは 2018年のことでした。それから2年近くになりますが、いくらか障害や計画停止はあったものの、データロスなどの大きなインシデントはまだ起きていません。クラスタが大きく拡張されたわけでも、Hadoop でなければできないことがあったわけでもありませんが、筆者は個人的にはこれは成功だと考えています。 Cloudera Manager が使いやすいことも大きいですが、構築や運用に携わっているメンバーが優秀で経験豊富なので、事前に運用を簡単にする自動化をかなりできていたこともあります。ディスクエラーがあると通知がきたり（False alert も少なくはありませんが）、事前に監視すべきメトリクスを決めて可視化しておいたり、TLSで使う証明書の自動更新をセットしたりしてあります。
それでもシステムを運用していると、問題は起きます。ここでは、運用中に起きた問題をいくつか紹介したいと思います。

運用事例1: なぜかDataNodeが起動後に再起動する

ユーザー管理は Kerberos を使っていましたが、グループ管理は当初、別途 LDAP group mapping を利用していました。これを sssd を利用した shell-based な group mapping へ移行したのですが、sssd に切り替えた直後、数分〜数十分ほど経過するとノードが不安定になってOSハングするという現象に悩まされました。ハングするとSSHで入るどころかpingにも応答しなくなっていました。
最初に部分適用した1台だけだったのでユーザーへの影響はありませんでしたが、当初は原因が分からずOSを再起動したりしていました。しばらく調べたところ、古いCPU microcodeでUbuntu 16.04がハングする報告を見つけて、カーネルアップデート時にCPUの microcode もアップデートされることを利用して修正したところ、ハングする現象がおさまりました。

その後他のノードにも変更を適用していきましたが、原因が不明なまま運用中のOSがハングするのは本当に怖いです。原因不明の問題が起きたままでもユーザーにほとんど影響がないのは、分散ストレージのよいところであり、HDFSで本当に助かりました。

運用事例2: CDH5からCDH6へのインプレースアップデート（Hadoop編）

昨年10月にCVE-2018-11768が報告され[8, 9]、社内で運用しているクラスタのひとつでデータロスのリスクがあることが判明しました。初期のクラスタはCloudera Manager も CDH もいずれも5系のまま運用していましたが、この問題に対処しつつソフトウェアを最新にアップデートするために、5系から6系へまるごとアップデートすることにしました。アップデートするにあたって、Cloudera Managerのアップデート自体は難しいことも少なく、また、このアップデートに失敗しても5系へ切り戻すことはDBのスナップショットが保存されていることもあり難しくはありませんでした。
しかしながら、HDFSとHBaseにはユーザーが利用中のデータがあるため、移行には慎重な作業が必要でした。Cloudera Managerによって作業がかなり簡単になっていることと、予めリハーサルをしておいたこともあり、アップグレードの手順は順調に完了しつつありました。しかしメジャーバージョンを跨ぐときの最大の山場ともいえるHDFSのメタデータ変換作業が終わらないという状況に陥りました。当初それなりにデータが格納されているので処理に時間が掛かっているのだろうと高を括って眺めていましたが数時間待ったところでログを見てみると良くない兆候が記録されていました。NameNodeがアップグレード後のブロックプールを構成するためにDataNodeからのブロックレポートを待っている一方で、DataNodeでは大量のブロックを一度にアップグレードしようとしてGCで止まったりOOMで落ちて再起動して再度大量のブロックを一度にアップグレードしようとしてまた落ちて、を繰り返していました。NameNode側ではそのDataNodeにおいて障害が発生していると見なしてブロックレポートの受け取りを拒否して切り離されていたためいつまで経ってもブロックプールが完成しないという状態になっていました。DataNodeのヒープサイズを8GBに設定していたためメモリが足りなかったことがわかり、一時的に12GBに増やしてDataNodeを起動すると無事にブロックレポートが送信されました。Hadoopの運用時の推奨値としてヒープサイズ8GBは適正値で、あまり増やしすぎるとCloudera Managerがきちんと警告してくれます。

12GBまで増やさないといけなかった理由は、PFNのDataNodeの物理的なスペックにあります。このクラスタのDataNodeは 8TB のHDDを36本備えており、物理容量は 288TB もあり、いわゆる High-density server といわれるレベルのスペックです。Hadoopを構築するにあたって、一般的には HDDのIO帯域を考えるとHDDはもっと少なくてよいです。PFNではMapReduceやSparkといったCPUインテンシブなユースケースは少ないので、CPUを相対的に少なくしてもよいだろうと考えてこの構成にしましたが、その考えが甘かったことが後から分かりました。実際にはHDFSのさまざまなパラメータが推奨構成に従って設定されており、推奨ヒープサイズもノード当たりのディスク容量が相対的に少ない構成を想定していたのでした。

運用事例3: CDH5からCDH6へのインプレースアップデート（HBase編）

全プロセスの再起動が完了し、Cloudera Managerを見る限りオールグリーン。いくつか hdfs コマンドを実行してみても特に異常なし。無事にアナウンスも完了して、さてめでたしめでたし…とはなりませんでした。 HBase のユーザーから報告がきます。いざHBaseから画像を取得してみると

^@6~Ø19692d3b721333bb59d9edab20059d1f20200326e0ce530fe2f646148f30ba218ca7507a

という文字列が返ってくるとのことでした。 hex でみると 0xA や 0xD というのが、もとの画像に対して追加されているようです。テーブル情報を確認すると、

hbase(main):003:0> desc 'namespace:image'
Table namespace:image is ENABLED
namespace:image
COLUMN FAMILIES DESCRIPTION
{NAME => 'blob', VERSIONS => '1', EVICT_BLOCKS_ON_CLOSE => 'false', NEW_VERSION_BEHAVIOR => 'false', KEEP_DELETED_CELLS => 'FALSE', CACHE_DATA_ON_WRITE => 'false', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', MIN_VERSIONS => '0', REPLICATION_SCOPE => '0', BLOOMFILTER =>
'ROW', CACHE_INDEX_ON_WRITE => 'false', IN_MEMORY => 'false', CACHE_BLOOMS_ON_WRITE => 'false', PREFETCH_BLOCKS_ON_OPEN => 'false', IS_MOB => '\xFF', COMPRESSION => 'NONE', BLOCKCACHE => 'true', BLOCKSIZE => '65536'}
{NAME => 'meta', VERSIONS => '1', EVICT_BLOCKS_ON_CLOSE => 'false', NEW_VERSION_BEHAVIOR => 'false', KEEP_DELETED_CELLS => 'FALSE', CACHE_DATA_ON_WRITE => 'false', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', MIN_VERSIONS => '0', REPLICATION_SCOPE => '0', BLOOMFILTER =>
'ROW', CACHE_INDEX_ON_WRITE => 'false', IN_MEMORY => 'false', CACHE_BLOOMS_ON_WRITE => 'false', PREFETCH_BLOCKS_ON_OPEN => 'false', COMPRESSION => 'NONE', BLOCKCACHE => 'true', BLOCKSIZE => '65536'}

MOBも無事にオンになっています。特におかしいことはない…と思っていたら、 IS_MOB => '\xFF' という不穏な表現にメンバーが気づきました。本来これは boolean なので、 true か false が入っていなければなりません。これは画像のカラムなので true になっているはずです。みんな脳内に???が連呼されていて、 hbase check や Cloudera Managerも特におかしなことは報告してきません。ということで、半信半疑で

> alter 'namespace:image', {NAME=>'blob', IS_MOB=>'true'}

を実行します。そうすると不思議なことに、画像がきちんと画像として読み出されます。調べてみると、HBASE-12085で IS_MOB のフラグの扱いがかわっており[10]、そもそも boolean などなかった（ \xFF という値でMOBかどうかを判断していた）のが、 boolean 型が導入されてしまい、MOBのカラムがMOBとして処理されなくなっていました。CDH5系から6にアップデートしたことで、HBaseも 2.0 になり、この問題が起きるようになったということでした。

おわりに

ここまで、利用事例や実際の運用の様子をお伝えしてきました。この他にもいくつか課題があり、それぞれに解決を図っています。特に、コミュニティの重心がOzone [11]へ移りつつあることには注目していて、今後は検証を進めていきたいと思っています。Ozoneに移行することにより、これまでHadoopに最適化されたクライアント側のライブラリが共通のRESTプロトコルへ移行して処理系のポータビリティが向上することや、小ファイル問題 [12] が解決することが期待されます。

謝辞

PFNでHadoopシステムを構築および運用するにあたって、Hadoopコミュニティの友人の皆さんに多くのアドバイスをいただきました。この場で感謝申し上げます。また内輪ではありますがこの場を借りて、インタビューに協力してくれたPFNメンバー、Hadoop運用（ボランティア）チーム、クラスタ運用チームの皆さんに感謝申し上げます。特に筆者は、こと運用事例においてはほとんど手を動かしておらず、横から口を出していただけであることを告白いたします。