たごもりすメモ

コードとかその他の話とか。

トップ > Hadoop

Hoop(httpfs)とwebhdfsの違い

Hadoop

Hadoop 1.0.0がリリースされた。まあ中身のほとんどはただの 0.20.x 安定板リリースなので特別に言うことはないんだけど。詳しくは以下のblogを読むのがよろしい。

hadoopのバージョン表記について - 科学と非科学の迷宮

ただしひとつだけびっくりしたのは、webhdfsなる機能が入ってきたこと。(このblogでよく話題にしている)Hoopと並んでそんなようなものがあること自体は知ってたけどあんまり興味なかったのだが、Apache Hadoopのパッケージに(Hoopより先に)入ったとなるとちょっと注目せざるをえない。
が、httpfs(Hoop)とwebhdfsじゃ名前も似てて超まぎらわしい。いったい何がなんなの。

なお自分はWebHDFSはAPIリファレンスを読んだだけで、実際にはカケラも触っていない。その状況での理解による内容なので、注意して読んでください。

先に結論だけ書く

どちらもクライアントからはHTTP REST APIで叩く。可能なオペレーションの幅も同じ。認証もユーザ名ベースの仮認証(pseudo auth)かKerberosのどちらか。ただしサーバ側をどのノードが担当するかという点、およびHTTPの通信パターンが異なる。
ぶっちゃけて言うと、これまでJava実装のHDFSClientがやっていた通信をHTTP REST APIで置き換え可能にする、というものがWebHDFSだと言っていい。

このため、機能的にはほぼ同じだが、性能特性はいくらか異なることが考えられる。どのようなアクセスパターンでどちらが良いかという確定的なことはまだ言えない。ぶっちゃけだいぶ実装による。ので、実際に試すしかないと思う。クライアントの実装がたぶん重要。
あと安定性にも違いが出るかもしれない。webhdfsのHTTPサーバ部分の実装がhdfsコアと同じ完成度なのかどうか、という点が重要。ここの完成度が高いならwebhdfsも安定するだろう。Hoop(httpfs)は今のところ安定性の面で問題を見付けたことはない。

Hoop(httpfs)の構造

Hoop Serverを立てる場合の構造および通信パターンとプロトコルをざっくり図示すると、こんな感じ。

クライアントは全ての操作について、Hoop Serverに対してHTTP REST APIでリクエストを投げる。Hoop Serverはリクエストを受け取ったら、それに対応するHDFSアクセス操作をJava実装のHDFSClient経由でHadoopクラスタに対して行い、結果をクライアントに戻す。

構造は単純だ。Hoop ServerはただのプロキシサーバでHDFS側から見ると普通のクライアントアプリケーションなので、ぶっちゃけ我々でも書こうと思えば書ける。

WebHDFSの構造

対してWebHDFSの場合、ノード間の関係および通信パターンはこんな感じ。

つまり、全てのNameNodeとDataNodeがHTTPをしゃべる。いっぽうプロキシサーバ的なものは存在せず、HDFSClientが各ノードと行っていたような通信を、そのかわりにHTTPを使って行うこととなる。HDFSClientの実装に依存するしかなかったクラスタとの直接通信をHTTPで代替できるようになったということだ。

いっぽう面倒くさい部分がある。周知の通りHDFSはディレクトリ構造およびファイルのメタデータをNameNodeが、ファイル実体をDataNodeが管理している。このため、ディレクトリ構造やファイルのメタデータに関する操作ならNameNodeにリクエストを投げてレスポンスを見ればいいのだが、ファイルの内容に関する操作(ファイル作成・追記、ファイルの内容取得)については、以下のような手順を踏む必要がある。

  1. NameNodeのリクエストを送信
  2. NameNodeのレスポンスによりDataNodeのどれかにリダイレクトされる
  3. DataNodeにリクエストを送信
  4. DataNodeから結果が返ってくる

上記のように一回のHTTPリクエストで処理が完結しない。これはクライアントの実装を行う上ですこし面倒だし、性能面でも影響を与える可能性がある。*1
いっぽう間にプロキシを挟まないためDataNodeと直接通信をする部分においてはスループットがかなり出ることも予想され、その面ではパフォーマンスに関する期待は持てる。つまりは結局アクセスパターン次第ですね、ということになるかもしれない。

なおHTTPリクエストのパターンが面倒なのは、将来的には解決される可能性がある。本来ならレスポンスとして 'Expect: 100 Continue' を使って直接DataNodeに処理を引き継ぐようにしたいらしい。現状でNameNodeのレスポンスがDataNodeへのリダイレクトになっているのはJettyのバグのせいであるとREST APIに明言されている。
これが解決されたら、クライアント側で2回のHTTPリクエストをハンドリングする必要がなくなるかもしれない*2。ただし、性能特性としては変わらないままだろう。

雑感

NameNodeおよびDataNodeへの直接の通信は、これまでJava実装によるHDFSClientに頼るしかない状況だった。これがHTTP REST APIという平易なもので置き換え可能になったという点で、WebHDFSの存在意義は大きいと思う。つまり誰でも各言語ネイティブのHDFSクライアントライブラリが書けるようになったということだ。
WebHDFSのREST APIリファレンスはだいぶしっかりした作り*3になっていて、仕様的にちゃんと考えられているんだなあ、という気もする。レスポンスJSONのスキーマなんかもちゃんと書かれているのでだいぶ実装にあたっての心理的障壁は低い*4。このAPIの互換性がしっかりと守られるなら、もしかしてJava実装のHDFSClientベースでクライアントライブラリを使うよりもWebHDFSベースでクライアントライブラリを書いてしまった方がバージョン互換性が保たれるということにもなりそうだ。

とはいえ、全てのクライアントアプリケーションがNameNodeおよびDataNodeのすべてと相互に通信するというのはなかなか面倒だし、トラブル時に何をどう考えればいいのかも複雑になる。なので、通信パターンの単純化、ということだけ考えてもhttpfs(Hoop)の存在意義はある。実際の運用を考えるとこの面はあまり軽視したものでもない。

ということで最後にはいつもの通りだが、用途とアクセスパターン、運用状況にあわせて適宜選択するのがよろしいでしょう。ああ、誰かHadoop 1.0.0のWebHDFS実装でベンチマークをとってくれないか・・・！

おまけ

ふと考えたんだが、HTTP REST APIでクライアントのリクエストを受け取ってWebHDFSでHadoopクラスタに処理を投げるプロキシサーバがあると、実は一番良いんじゃなかろうか。本当かな。

*1:前半のNameNodeからのレスポンスをキャッシュできるのかどうか、できるとしたらどの条件下でキャッシュできるのか、が明らかになれば少し違うかもしれない。

*2:HTTPクライアントライブラリが処理を裏側でやってくれるなら。

*3:ただしPDFのみ。なぜ……。

*4:この点Hoopはだいぶダメダメ。。。

fluent-plugin-hoop v0.1.0 released

fluent Hadoop

みんな大好き fluentd は素敵だと思うんだけどHDFSへの書き出しをするプラグインが無い！ なんで！ という全世界100万人のエンジニアの怨嗟の声が聞こえてきそうだったので、とりあえずどうにかするべく書いた fluent-plugin-hoop がだいたい動くようになったのでリリースした！

rubygems.org経由でリリースしてあるので gem install fluent-plugin-hoop で入ります。あとはなんだっけ、システム全体の環境に影響を与えたくなければ fluent-gem install fluent-plugin-hoop かな。

Hoop(HDFS http-fs)はHDFSへのオペレーションをover HTTPで提供するサーバソフトウェア。詳しくはぐぐれ。ぐぐってください。*1
書き込みパフォーマンスが十分に出そうだったので、これを経由してHDFSに書き込むということにすればRubyでさくっとfluentdのプラグインが書けそうだったので書いてみた。さくっと書けたという割にはあれこれ試行錯誤があったけどいちおうできて、そこそこのデータ流量にしてみても問題ない程度にはパフォーマンスが出ているので公開してる。自分の手元の用途に対して本能に大丈夫かどうかはまだ最終確認できてない……。

なお、現状の動作確認はすべて Cloudera版の Hoop 0.1.0-SNAPSHOT で行っています。0.23系クラスタを手元で作れる誰かに是非試してみてほしい。

特徴

ログに書かれている時間を見て書き込み先の(HDFS上の)パスを確定するようになっている*2。このため、普通のログローテーションにありがちな「23:59:59のログが翌日分のファイルに泣き別れ」などの事態を避けられる。
またTAB区切りのプレーンテキストファイルで各値を出力できるようにしており、これはHiveが直接読める形式なので個人的にはとても嬉しい。0分0秒*3にはLOAD一発走らせることででHiveクエリが発行可能になる。グレイト。

残念なのが、書き込み先のHDFSパス指定でタグを使えない(パス指定でタグを展開するような設定ができない)こと。これは fluentd TimeSlicedOutput の仕様によるので、ちょっと簡単にはどうにもできない。このためタグごとに出力パスを分けたい場合は設定ファイルにおいて分けたいタグの数だけmatch節を並べる必要がある。
正直に言ってこれはだいぶ残念なので、時間ができたらどうにかできないか考えてみたい。fluentdにTimeSlicedOutputより更に高機能なOutputプラグインベースクラス*4を作ることになるのかなあ。

ベンチマークをとっていて明らかになったことだが、HoopServerは同一パスへの追記が複数のHTTPセッションから混在して連続するとどうも怪しい挙動を示すことがある。バグ報告したいし確認用のコードも書いたけどまだしてない*5。これを回避するため割とヒューリスティックなwaitつきretryの処理を入れてあって、トラフィックを見てるとちょっと不審な挙動を示すのが残念なところ。

ベンチマーク

とりあえずベンチマーク。

サーバは前のfluentdベンチマークのときと同機種のものを使った。構成としてはベンチマークスクリプトが scribe プロトコルでデータを送り、fluentdは in_scribe でそれを受信、out_hoop で HoopServer に送信する。HoopServerはHoopベンチマーク時のNameNodeに立てたのをそのまま使っている。
fluentdはサーバ上に4プロセス起動、ベンチマークツールも(別サーバ上に)4プロセス起動し、ベンチマークツールとfluentdは1対1に対応している。fluentd 4プロセスはそれぞれ同時にHoopServerと通信し、出力はHDFS上のひとつのファイルに4プロセス分混ざって出力される。

fluentdの設定ファイルは以下の通り。gistにも置いといた。ただしin_scribeのポート番号のみ 14631, 14632, 14633, 14634 とひとつずつずらしてある。

<source>
    type scribe
    # 14631, 14632, 14633, 14634
    port 14631
    add_prefix scribe
</source>

<match scribe.*>
    type hoop
    hoop_server hoop-server.local:14000
    path /hoop/%Y%m%d/scribe-%Y%m%d-%H.log
    username edge-dev
    time_slice_wait 30s
    flush_interval 5s
    output_include_time false
    output_include_tag  true
    output_data_type attr:message
    add_newline false
    remove_prefix scribe
    default_tag   unknown
</match>

出力先ファイルは1時間ごとに切り替え、最大30秒は遅延ログの到着を待つ、という設定。scribeプロトコルで受け取ったログのmessageデータは末尾に改行を含むので out_hoop 側での改行付与はオフにしてある。
また後々他のログデータのルーティング設定と併存させたときに混ざらないよう in_scribe 側でタグのprefixとして 'scribe' をつける*6。in_scribe は受け取ったデータのタグは category を見るため、例えば category が 'foo' だった場合、fluentd内部でのルーティング時にはタグは scribe.foo となる。out_hoopの設定では remove_prefix scribe があるので、fleuntd内部のルーティング用につけた文字列はここで除去されてファイルに書かれる。

ベンチマークスクリプトは github.com/tagomoris/fluentd-tester/ に置いてあるコレ。保存ログから欠損を見付けるためにタグ(scribeにおけるカテゴリ)に通番を埋められるようにしたりしてある*7。こんな感じで起動する。

 $ ruby scribe_stream.rb hoop-server.local 14631 first 300 250 40 720 &
 $ ruby scribe_stream.rb hoop-server.local 14632 second 300 250 40 720 &
 $ ruby scribe_stream.rb hoop-server.local 14633 third 300 250 40 720 &
 $ ruby scribe_stream.rb hoop-server.local 14634 forth 300 250 40 720 &

メッセージ長300bytesは現実のApacheのログ長の平均がこのくらいかな、というのに合わせた*8。いちおうベンチマークの目標としてはプロセスあたり 10,000 msg/sec を4並列の 40,000 msg/sec となる。実際には7/8の 35,000 msg/sec をいくらか下回る程度流れたようだ。
12時間で 1,493,340,000 行(約14億9千万行)。HDFS上のファイルをカウントしたところ1行の欠損もなく保存されていた。

以下グラフ。左の方の山は無視していただいて、問題はいちばん右の12時間分あるやつです。
まずトラフィックで、Hoopサーバ(役のNameNode)、fluentdサーバ、の順。緑がinputで青線がoutput。

fluentdサーバのinputは安定していて、これはベンチマークツールの出力を受けたもの。だいたい100Mbps。狙ってたわけじゃないけど偶然ですな。HoopサーバのoutputはHDFS DataNodeへの書き込みで、これも約100Mbpsで安定してる。
問題はfluentd(out_hoop)とHoopサーバの間の通信で、だいぶ上下してる。ちょっとアレな感じ。

  • Hoopへの通信は根本的に HTTP のオーバーヘッドを伴うので、実データに較べてちょっと増える
    • これは1プロセス時に確認したもので、多分バッファリングのサイズをもっと大きくすれば目立たなくなる
  • データのAppend時にHoop Serverが 500 (Internal Server Error) を返すことがあり、こうなると0.5秒くらいはエラーを返し続ける症状を示す
    • このため out_hoop で怪しげな500エラーの場合0.3秒のwaitを入れてからリトライしている

後者のリトライが重なったりするんだろうなーという気がする。Hoop Server の500応答は2並列30秒走行でも明確に発生が確認できる程度の頻度で起きるので、4並列だとさぞ頻発してるんだろうと思われる。確認してないけど。
とはいえ、まあ4並列でこのくらい速度が出るならいちおう及第点。ほんとはこの倍くらい出ることを確認しときたいんだけど、後で。

お次はCPU使用率とロードアベレージ。同じく左側がHoopサーバで右側がfluentdサーバです。基本的には問題ないでしょう。複数プロセス使えばCPUがそこそこ使えるようになりますねってだけですね。メモリは完全に全く問題なかったのでパス。

まあ、だいたいの用途には問題ない出来になっているのではないでしょうか！

まとめ

out_hoop を書いたことでfluentdからHDFSへの投入の道が開けました。みんなこれでログをどんどんHDFSに投入してガンガンHadoopタスクを回すと良いんじゃないでしょうか。
タグをパスに埋められるようにはどうにかしたい！ match節いっぱい書くとか嫌だ！

*1:Hoopだけでぐぐったらなんか全然違うのが出てきたけど hoop hdfs でぐぐったらこのblogがトップに！ 俺つおい！

*2:正確にはログに書かれている時間は input plugin 側でparseしてやる必要があって in_scribe には現在その機能はない、のであんまり正確な説明ではない。各言語向けのloggerなどはログを出力した時刻をとるようになってると思うのであんまり気にする必要はないだろう。Apache等のログ時刻のparseをどうするかは自分の場合ちょっと違うアプローチをとるので in_scribe への機能追加は後回し……。たぶんそのうちやる。

*3:の数十秒後

*4:時間とタグの2値でバッファのキーにする、それにあわせてAPIを少し変える？

*5:そもそもCloudera版Hoopのコードで動作確認してるんだけど現状の開発はApache Hadoopのツリー上でHDFS httpfsとして行われてるからどうしていいのやら。

*6:この変更は自分でパッチ書いた、けどまだリリースされてないかな？ githubからcloneしてfluentd起動時に -p オプションでプラグインを個別指定する。

*7:実は通番桁数を8桁にしちゃったため、12時間で億の桁にいってしまってあんまり役に立たなくなってしまったけど。

*8:最近おまえらUser-AgentとかRefererとか長過ぎだろ！！！

HDFSのファイルオペレーション各種 #hadoopAC11jp

Hadoop

この記事は hadoopアドベントカレンダー の14日目の記事です。

みなさんHDFS使ってますか。使ってますよね。最近はgluster fsなどの選択肢も出てきていますが、まだ第一の選択肢はHDFSという人がほとんどだと思います。
で、HDFSのファイル操作をどうしようか悩みますよね。めんどくさいです。いくつかあるので比較してみましょう。

hadoop fsコマンド

みんな大好きhadoop fs。日に100回くらいは叩きますね。基本的にはみんなこれを使うでしょうか。ただし以下いくつかの「たるいなー」という点があります。

Hadoopがインストール・設定されていないと動かない

あたりまえなんですけど意外にめんどくさいですね。
Hadoopの処理対象となるファイルはもちろんHadoopクラスタ「以外」のサーバから出てくるわけで、そこからも直接ファイルをHDFSに突っ込みたいなーというのは割と自然な欲求のような気がします。自分の扱ってる環境だと基本的にサーバにJVMも入ってないんで、各サーバにHadoopの環境を用意するのはだいぶハードルが高いです。

JVMの起動時間を待つ必要がある

普通のファイルシステム上での操作と同じ感覚でコマンドを叩いていると、これが意外にストレスですね。
hadoop fsコマンドはJava実装のHDFSClientをつかってあらゆる操作を行うため、何をやるにしてもJVMの起動を待つ必要があります。普通のファイルシステムであれば mv などは一瞬で操作が終わるもので、もちろんそれはHDFS上でもほぼ同じ(renameのコストはかなり低い)ですが、実際にはJVMの起動・停止の待ちがあってかなり待ちます。

FUSE hdfs

Linux(やその他のOS)には FUSE(filesystem in userspace) という機構があり、それを介してHDFSに接続することでLinux上でHDFSをmountする(したように見せる)ことができます。Hadoop Wikiにも解説(英語)があります。また自分も以前にblogエントリを書きました。
fuseでhdfsをmount - tagomorisのメモ置き場

利点はLinux上での通常のファイル操作がそのまま行えること。じつに大きい利点だと思います。だいぶいいですね。
しかしこれはこれで、どうにも信用のならない点がいくつかあります。

libhdfsのバージョン問題

FUSE hdfsはHadoopのパッケージに添付の libhdfs を使用します。簡単に解説するとlibhdfsとはHDFS関連のオペレーションを行うためのC実装のライブラリなのですが、その実態は JNI でJava実装のコード(HDFSClient)を呼び出しているだけです。
このため、Hadoopのバージョンに強く依存します。Hadoop自体のバージョンを上げた場合にはクライアント側でも追随しなくてはなりません。またクライアント側のバージョンアップ時にはHDFSのumount、libhdfs の更新、fuse_dfs の再コンパイルとインストール、そしてHDFSのmountという手順を踏むことになります。だいぶ面倒な上にmountの切断時間が長くて悲しいですね。これを関連するクライアント全台で行うと考えるのはだいぶげんなりします。

パフォーマンスの問題

FUSE hdfsはバックエンドとしてひとつのプロセスがHadoopクラスタと通信を行います。mountしてあるHDFS領域に関するすべてのオペレーションをこのプロセスが担当するので、通信量などによってはパフォーマンスが出ないことが予想されます。せっかくファイルシステム側が分散システムなのに、手元の1プロセスに制約されるのは、仕方がないとはいえ残念ですね。
特に最近のCPUは多コア化して各コアの処理能力はそこまで求めない場合が多く、このような環境だと1プロセスに処理が集中するモデルは不利となります。またバースト的に負荷がかかったときに、FUSE hdfs自体の安定性があまり良くない気もしています。*1

Hoop

という現状に問題意識があったのか、最近 Hoop というツールが出てきました。Clouderaから公開されたものですが、Hadoop本体へのマージが進められたようで、0.23.1(および0.24)のツリーにおいてマージされたようです。

[HDFS-2178] HttpFS - a read/write Hadoop file system proxy - ASF JIRA

HoopはWeb(REST) API経由でHDFS上のファイルオペレーションを行うもので、Hoop Serverをどこかに立てて使います。クライアントはHTTPリクエストを送れさえすれば何でもいいということで、これまでのHadoopに密結合された環境を要求しないのが最大の利点と言えると思います。
なお、現状(？)の0.20系においても、Clouderaからリリースされているバージョンを使えば既に使用できます。これについてもエントリを過去に書きました。

Hoop (HDFS over HTTP) を試してみた - tagomorisのメモ置き場
Hoopの性能を確認してみたらもうlibhdfsとかオワコンでHoop使えって結果になった - tagomorisのメモ置き場

とはいえHoopに問題が無いかといえば、無いわけでもないです。

Hoop ServerがSPOF

前述した通り全クライアントがHoop Server経由で通信を行うため、これが障害に遭うと全クライアントが通信できなくなります。ちょっと問題です。
ただ、複数立てられるので、怖い人は複数立てておけばよいでしょう。だいたい現状だとHDFSのNameNodeがSPOFな時点で……って感じですね！

またHoop Serverのパフォーマンスを気にされる人がいるかもしれませんが、これについては前述のエントリでとったベンチマーク結果の通り、基本的に問題なさそうという認識でいます。

HTTPでリクエストを投げる必要がある

これが一番の障壁に思う人が多いかも。
普通のファイルシステム操作とだいぶ異なるオペレーションを行わないといけなくなるため、扱う側にとってあれこれ考えを変える必要があります。REST APIのリファレンスを見ればわかる通り curl コマンドを叩けばどの操作もできるんですが、ちょっと敷居が高いのは確かですね。
普段のオペレーションに使いたい場合はREST APIを叩く専用のラッパーコマンドなりを自作するのが良いでしょう。

まとめ

HDFS上のファイルのオペレーションを行う方法をみっつ挙げましたが、どれも一長一短な側面はあります。個人的にはHoopはいいなと思いますが、みなさんの環境や運用状況にあわせて適切に選ぶとよいのではないでしょうか！

*1:あまり明確な証拠が出せるものがないのですが！

Hoopの性能を確認してみたらもうlibhdfsとかオワコンでHoop使えって結果になった

Hadoop

前に書いた エントリ の通りHoopが有望な感じだったんだけどどのくらいの性能が出るのか見てみないことには本番投入して性能出ませんでした乙、ということになりかねない。ので見てみた。

なお検証に関係する環境としては以下の通り。ちなみに前はCDH3u1で試してたけど、今回はCDH3u2 (JDK6u29) on CentOS5。メモリが問題になることは全くないので全て省略。

  • ベンチ用サーバ
  • データ中継サーバ (deliver)
  • Hadoop NameNode (+JobTracker) (namenode)
  • Hadoop DataNode + TaskTracker x9

Hoopを起動したのはデータ中継サーバとNameNodeの2ヶ所で、それぞれに対して試してみた。上記サーバはすべて別のラックにありGbEで接続されている。Hadoopクラスタの各ノードはNameNodeと同一および隣のラックに配置。

なお データ中継サーバ(deliver)はこのテスト中も scribed が走っていて、やってくるログをhdfs(scribed with libhdfs)に書き込んでいる。が、scribedによるCPU負荷やネットワーク使用率が問題になることはあるまいということで、そのまま実施した。(実際に、多分問題なかった。)

書き込み／読み込みのスループット

とりあえずナイーブな書き込みおよび読み込みについて、hadoop fsコマンドと較べてどのくらいの数字差が出るか。以下の条件でやってみた。

  • 2GBのファイルのPUTを以下の順序で(最初にそのまま実行、次に time コマンドをかませて計測)
    1. hadoop fs -put による書き込み
    2. curl でNameNode上のHoop REST APIを使って書き込み(op=create)
    3. curl で中継サーバ上のHoop REST APIを使って書き込み(op=create)
  • 2GBのファイルのGETを同じように実行
    1. hadoop fs -get による読み出し
    2. curl でNameNode上のHoop REST APIを使って読み込み
    3. curl で中継サーバ上のHoop REST APIを使って読み込み

結果は以下のとおり。まずPUT。

hadoop fs -put 2gdata.txt /tmp/put1.txt ;\
time hadoop fs -put 2gdata.txt /tmp/put2.txt ;\
curl -X POST 'http://namenode:14000/tmp/put3.txt?op=create&user.name=hoge' --data-binary @2gdata.txt --header "content-type: application/octet-stream" ;\
time curl -X POST 'http://namenode:14000/tmp/put4.txt?op=create&user.name=hoge' --data-binary @2gdata.txt --header "content-type: application/octet-stream" ;\
curl -X POST 'http://deliver:14000/tmp/put5.txt?op=create&user.name=hoge' --data-binary @2gdata.txt --header "content-type: application/octet-stream" ;\
time curl -X POST 'http://deliver:14000/tmp/put6.txt?op=create&user.name=hoge' --data-binary @2gdata.txt --header "content-type: application/octet-stream"

# hadoop fs -put
real	1m8.564s
user	0m8.736s
sys	0m1.635s

# curl - Hoop(namenode)
real	1m2.889s
user	0m0.830s
sys	0m1.676s
    
# curl - Hoop(deliver)
real	1m2.837s
user	0m0.811s
sys	0m1.532s

どれも対して変わらない、っていうか hadoop fs コマンドが一番遅いって……。あとから考えたがJVMの起動コスト分ですかねえ。user 8.736s の大部分がそれかな。2GB/60secだとビットレートにして266Mbps。わお。
で、つぎはGET。

hadoop fs -get /tmp/put1.txt - > /dev/null ;\
time hadoop fs -get /tmp/put1.txt - > /dev/null ;\
curl -X GET -s 'http://namenode:14000/tmp/put3.txt?user.name=hoge' > /dev/null ;\
time curl -X GET -s 'http://namenode:14000/tmp/put3.txt?user.name=hoge' > /dev/null ;\
curl -X GET -s 'http://deliver:14000/tmp/put5.txt?user.name=hoge' > /dev/null ;\
time curl -X GET -s 'http://deliver:14000/tmp/put5.txt?user.name=hoge' > /dev/null

# hadoop fs -get
real	0m31.224s
user	0m12.519s
sys	0m4.926s

# curl - Hoop(namenode)
real	0m49.797s
user	0m2.741s
sys	0m4.727s

# curl - Hoop(deliver)
real	0m30.389s
user	0m1.354s
sys	0m2.404s

これもdeliver経由だとhadoop fs -getより速くて、やっぱりJVM起動コストですかね。2GB/30secだと532Mbps？ わーお！ namenodeはCPU弱いサーバなので、たぶんそこで差が出ちゃったかな。本当は同一スペックのサーバで試せればどこにHoopを置くか決まって良かったかもしれないけど、でもまあ49秒で終わる程度にパフォーマンス出るなら実質充分。

継続的な追記のスループット

とはいえ単発のGET/PUTの性能はあんまり関係なくて、継続的に大量のログを書くときの中継点としてHoopが使えるのか知りたい。ので、特定のパスにひたすら追記(append)を繰り返す処理を試してみた。

ファイルへのappendは Hoop REST API 経由で、ベンチマークのクライアント側はRubyでコードを書いた。これは一定時間、指定されたデータソースを指定されたHoopサーバにappendしつづける。データソースは最初にメモリに読み込むのでクライアント側のdisk I/O負荷は考えなくていい。
continuous write over hoop

これを一度起動すると、指定したサーバに対して指定した秒数だけappendを繰り返し*1、最終的に何度のappendが成功したか、何度失敗したか*2、概算の平均スループットはどうだったか、およびtimeコマンドの出力を以下のように出す。

written chunk:13421, failed:0
rate: 149 Mbps

real    120m0.573s
user    0m7.917s
sys     0m37.009s

これをシェルスクリプトで並列起動させ、以下のような処理になるようにした。

  • 最初の60分
    • 1つ起動、10MBのファイルをメモリに読み込んでひたすらappend
  • 次の60分
    • 2つめも起動(2並列)、おなじくappendを別ファイルへ
  • 次の60分
    • 3つめも起動(3並列)、おなじくappendを別ファイルへ

結果がこちら。まずnamenode上のHoop、次にdeliver上のHoopに実行している。コメントは説明のため加えた。

starting namenode ....
Mon Oct 31 20:24:39 JST 2011

# 1st process (namenode)
written chunk:17644, failed:0
rate: 130 Mbps

real    180m0.816s
user    0m10.253s
sys     0m49.879s

# 2nd process (namenode)
written chunk:10285, failed:0
rate: 114 Mbps

real    120m0.642s
user    0m5.701s
sys     0m28.964s

# 3rd process (namenode)
written chunk:4360, failed:0
rate: 96 Mbps

real    60m0.704s
user    0m8.196s
sys     0m41.168s

ended master101.analysis
Mon Oct 31 23:24:40 JST 2011

starting deliver
Tue Nov  1 00:24:40 JST 2011

# 1st process (deliver)
written chunk:20787, failed:0
rate: 153 Mbps

real    180m0.165s
user    0m12.570s
sys     0m57.139s

# 2nd process (deliver)
written chunk:13421, failed:0
rate: 149 Mbps

real    120m0.573s
user    0m7.917s
sys     0m37.009s

# 3rd process (deliver)
written chunk:6088, failed:0
rate: 135 Mbps

real    60m0.326s
user    0m16.180s
sys     1m14.040s

ended deliver101.att.scribe.admin
Tue Nov  1 03:24:42 JST 2011

これだとわかりにくいよねー、ってことで、ずばり上記の試験期間中のグラフはこんな。

ベンチマークサーバ(ネットワークトラフィック＆CPU使用率)


namenode(ネットワークトラフィック＆CPU使用率)


deliver(ネットワークトラフィック＆CPU使用率)


まとめ

性能を見る限り deliver でのスコアは3並列で430Mbpsくらい出ていて*3、そのいっぽう該当の時間帯のdeliverサーバのCPU使用率を見ても1コアを使い切った数値(12.5%)にまったく届いていない。Hoop自体のパフォーマンスとしては430Mbps出て、まだ余裕あり、というところだろう。
2並列から3並列への数値の伸びを見るに1.5倍にはなってないので、このあたりでネットワークなりなんなりの要素が絡んできてるかなーという気はする。これ以上やろうと思うとネットワークまわりの環境をもっといじったりしないと難しいかも。

とはいえ、ぶっちゃけこんだけ出てれば何も問題ない。400Mbps書き込めることがわかってればなんでもできるでしょ。というかlibhdfsとか要らないってこれ。
libhdfsみたいにJVMに依存しJNIに依存しHadoopのバージョンであれこれあり、というものに較べてクライアントは軽量なHTTP REST APIを叩くだけでこんだけ性能出るんだもん。もう全部これでいいんじゃないですかね。

Hoop REST APIの罠

今回やってて気付いたこと。REST API経由でのはなし。

  • Append時に対象のパスが存在しなかった場合はエラー(FileNotFound)になる
  • Create時に対象パスのディレクトリが存在しなかった場合は勝手に作られる
  • Createのデフォルトオプションは overwrite=true
    • つまりファイルが存在した場合は無かったことになって、新たにCreate時のデータで新規作成される

これはどうしたもんかなーという挙動。本当は逆で、以下みたいなのがいいんだけど。

  • Append時に対象のパスが存在しなかった場合は新規作成されて書き込まれる
  • Create時に対象パスのディレクトリが存在しなかった場合はエラー

とはいえこうなってないものは仕方無い。ので、実用上は以下のようにするしか無いと思う。

  • 書き込みはすべてAppendでやる
    1. エラーになったらリトライとしてCreateを overwrite=false でやる
    2. Createもエラーになったら*4Appendを再度試みる

これで誰かと競合しても大丈夫かなあ。ということでHoopを経由して書き込むときは競合でデータが失われないよう、みなさん気をつけましょう。

*1:最初の一度だけはcreateだけど

*2:REST APIのレスポンスコード依存

*3:deliver側のトラフィックを見ると約480Mbpsになってるけど、これは通常のログ書き込みが50Mbpsくらいあるため

*4:誰かが作ったということなので

CDH3u2とCDH3b2がHDFSプロトコルにおいて非互換らしい

Hadoop scribe

いま使っている環境(scribe + CDH3b2クラスタ)からHadoopクラスタを別のノード群で構成しなおすことになったのでCDH3u2で組んでみたんだけど、いざ相互接続をやるべー、と思ったらなんかscribedからHDFSに書き込みができない。
最初はよくあるネットワークや設定の問題かと思ったんだけど、あれこれ試してみたらタイトルのような状況らしいことがわかってきた。

以下、CDH3b2のノードを nodeX.cluster CDH3b2クラスタのNameNodeを nodeY.cluster CDH3u2クラスタのNameNodeを nodeZ.cluster と書くことにする。
また大前提として nodeX はCDH3b2で構成されたクラスタに普通に hadoop fs -ls できるし nodeZ はCDH3u2で構成されたクラスタに普通に hadoop fs -ls できる。

CDH3b2のノードからCDH3u2のクラスタへのアクセス

失敗する。

$ hadoop fs -fs hdfs://nodeZ.cluster:50071/ -ls /
Bad connection to FS. command aborted.
CDH3u2のノードからCDH3b2のクラスタへのアクセス

失敗する。

$ hadoop fs -fs hdfs://nodeY.cluster:50071/ -ls /
Bad connection to FS. command aborted. exception: Call to nodeY.cluster/10.0.xxx.xxx:50071 failed on local exception: java.io.EOFException
nodeXに仮のCDH3u2環境をつくってCDH3u2クラスタへアクセス

成功しちゃった。手順はざっと。~/configs/hadoop にCDH3u2クラスタ用の設定一式があると思いねえ。

$ cd /usr/local; tar xzf hadoop-0.20.2-cdh3u2.tar.gz
$ export HADOOP_HOME=/usr/local/hadoop-0.20.2-cdh3u2
$ export HADOOP_CONF_DIR=$HOME/configs/hadoop
$ export PATH=$PATH:/usr/local/hadoop-0.20.2-cdh3u2
$ hadoop fs -fs hdfs://nodeZ.cluster:50071/ -ls /
log4j:ERROR setFile(null,true) call failed.
java.io.FileNotFoundException: /var/log/hadoop/SecurityAuth.audit (No such file or directory)
	at java.io.FileOutputStream.openAppend(Native Method)
	at java.io.FileOutputStream.<init>(FileOutputStream.java:177)
	at java.io.FileOutputStream.<init>(FileOutputStream.java:102)
	at org.apache.log4j.FileAppender.setFile(FileAppender.java:290)
	at org.apache.log4j.FileAppender.activateOptions(FileAppender.java:164)
	at org.apache.log4j.DailyRollingFileAppender.activateOptions(DailyRollingFileAppender.java:216)
	at org.apache.log4j.config.PropertySetter.activate(PropertySetter.java:257)
	at org.apache.log4j.config.PropertySetter.setProperties(PropertySetter.java:133)
	at org.apache.log4j.config.PropertySetter.setProperties(PropertySetter.java:97)
	at org.apache.log4j.PropertyConfigurator.parseAppender(PropertyConfigurator.java:689)
	at org.apache.log4j.PropertyConfigurator.parseCategory(PropertyConfigurator.java:647)
	at org.apache.log4j.PropertyConfigurator.parseCatsAndRenderers(PropertyConfigurator.java:568)
	at org.apache.log4j.PropertyConfigurator.doConfigure(PropertyConfigurator.java:442)
	at org.apache.log4j.PropertyConfigurator.doConfigure(PropertyConfigurator.java:476)
	at org.apache.log4j.helpers.OptionConverter.selectAndConfigure(OptionConverter.java:471)
	at org.apache.log4j.LogManager.<clinit>(LogManager.java:125)
	at org.apache.log4j.Logger.getLogger(Logger.java:105)
	at org.apache.commons.logging.impl.Log4JLogger.getLogger(Log4JLogger.java:229)
	at org.apache.commons.logging.impl.Log4JLogger.<init>(Log4JLogger.java:65)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:39)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:27)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:513)
	at org.apache.commons.logging.impl.LogFactoryImpl.newInstance(LogFactoryImpl.java:529)
	at org.apache.commons.logging.impl.LogFactoryImpl.getInstance(LogFactoryImpl.java:235)
	at org.apache.commons.logging.impl.LogFactoryImpl.getInstance(LogFactoryImpl.java:209)
	at org.apache.commons.logging.LogFactory.getLog(LogFactory.java:351)
	at org.apache.hadoop.fs.FsShell.<clinit>(FsShell.java:53)
log4j:ERROR Either File or DatePattern options are not set for appender [DRFAS].
Found 2 items
drwxr-xr-x   - dareka supergroup          0 2011-10-25 17:17 /scribe
drwxr-xr-x   - dareka supergroup          0 2011-10-24 18:58 /var

なんかlog4jのエラーが出てるがどっかに /usr/local/hadoop (こっちはCDH3b2)を使う設定が残ってたんだろう。それはまあいいとして、HDFSの中身が見えてる。ぐええ。

結論

CDH3u2とCDH3b2の間ではHDFSのプロトコルに互換性がない。beta3, update0, update1, update2 のどこで失われたかは知らんし調べる気力もない。
しかしへこむなー。scribedがlibhdfs使ってるので、CDH3u2と組合せて再コンパイルしないと。めんどくさい。

Hive Client Webアプリケーション shib をつくった

Hadoop Hive node

(2013/04/02追記 see: http://d.hatena.ne.jp/tagomoris/20130402/1364898063 )

まだ完成度がいまいちだからなーと思ってエントリ書いてなかったんだけどLTでしゃべっちゃったので、ちゃんと書いておく。

Hiveにクエリを発行して結果を確認するためのWebアプリケーションを社内用途で作ってるんだけど、普通に他でも使えると思うので公開してあります。

tagomoris/shib · GitHub

シブ と読みます。
セットアップ方法はドキュメントを参照のこと。起動してブラウザでアクセスするとこんな画面が出てくる。

使いかたは見ればわかる、と思う。たぶん。クエリは参照専用(SELECTのみ)。 __KEY__ とか __KEY1__ とかがプレースホルダですよってくらいかな。エディタ内でプレースホルダを書くとプレースホルダを置換する値のボックスが出現する。PRIMARY KEYWORD は特別扱いされて、履歴リストに出現する。など。ただし複数のキーワードを使うのはちょっとUIの動作が不安定な気がする。

そういえばChromeでしか動作を確認してない気もする。SafariやFirefoxでも多分動くんじゃないかなー。IEは死ね。

できること

  • クエリの実行
    • ただしSELECTクエリに限る
    • 一度にひとつのクエリのみで、セミコロンが入ってたらエラー
      • サブクエリは通る(はず)
  • 実行されたクエリの結果はいつでも取り出せる
    • 取り出しはTSVもしくはCSVで
    • あんまりデカい結果の場合はCSVだと重いorエラーになる、かも
  • 過去履歴からのクエリ参照＆実行
    • 既に実行したクエリのリラン可能
      • ただし取り出せる結果は最新のもののみ(過去のも取り出せるようにしてもいいけど、めんどい上に用途がない)
    • 過去のクエリをコピーして新しいクエリとして編集できる
      • コピペベースのクエリ再利用

基本的にはお手軽カンタンにHiveクエリを実行できる、というのが最優先の作り。人を迷わせるような機能類はすべて排除。あとHTML5なブラウザなら「そのブラウザで実行したクエリ」の履歴が出る。

できないこと

  • 認証およびアクセスコントロール
  • 結果のグラフやチャート表示
    • これはそのうちできるようにする予定
  • hive 0.7.0以降の新機能の対応
    • データベース対応や認証など
    • 新しい関数なんかは問題なく使えるはず
  • クエリの進行状況や対象データのサイズ表示
  • 賢いクエリエディタ

そのうちやりたいけどできないのがこのあたり。不要不急なので放置してあるだけで、手をつけられるようになったらやるはず。

なんでつくったの

MapReduceってなんですか？ という人がHiveクエリを叩けるようにするには既存のツールだといろいろ問題あるなーと思ったから。既存ツールというのは Beeswax のことだけど。
前にも書いたとおり、お手軽カンタンというのが最優先。MapReduceって何ですか、というディレクターにBeeswaxの画面を見せれば拒否反応しか起こさないと思う。どうしても使う必要があるならやらせれば覚えるかもしれないけど、今のところウチではそこまでこのオペレーションは重要じゃない。ある人がある日にふとある数字が見たいと思ったとき、できるだけ拒否感なく使ってもらえるように、と思ってる。コピペベースのクエリ再利用はそのため。

破壊的操作はできないようになってるんだと不便すぎないか、と言われたら、そうかも。でもそれで不便なようだったらBeeswax使えるように立てればいいし、特に問題ない。

クエリエディタの高機能化はちょっとしたい気がするけど、やりたいことに対して自分のUIエンジニアリング力が足りなさすぎる、というかjavascript力が足りなさすぎる。正直誰かに作ってほしい。おれ無理。UDFの補完くらいならそこまで苦労せずにやれるかなあ。

実装は？

アプリケーションはNode.jsで書いて、データ保存はKyoto Tycoonでやってる。RDBいらなかったし、クエリの結果が数MB以上になる可能性が充分にあったので統一してぶちこめるストレージを選んだ。Kyoto Tycoon(というかKyoto Cabinet)のファイルハッシュデータベースはこういう用途に非常に便利だと思う。Node.jsであることの利点はこのエントリに前に書いた。
Node.jsなWebアプリでJobQueueなしにラクラク巨大処理を実行 - tagomorisのメモ置き場

UIはHTMLとしては1枚で、あとは全部Ajaxで叩いて表示を更新するような形になっている。だからHTTPリクエストの詳細を見て外部から叩くようにすれば、バッチ用スクリプトからshibを介してクエリを実行することもできる。結果の保存をやってくれるからそういう使いかたも便利かもね。

shib(シブ) ってなに？

渋峠 - Wikipedia

まとめ

用途がハマる人にとっては便利なんじゃないかなーと思います。お好きにどうぞ。

Hadoop Conference Japan 2011 Fall に行ってきた＆しゃべってきた

Hadoop

なんかすごいイベントだった。User Group主催のイベントなのに2ホールを1日貸し切り(というか施設まるごと貸切)でキャパシティ1400人のイベントとかどういうこと。しかも無料参加なのにランチボックスとか飲み物とか出てた。意味がわからん。カネのあるところは違うということかー？
ともあれとりあえず主催者の方々はお疲れさまでした！

内容のサマリが読みたい人はこんなエントリを読んでないで、他の人がまとめてるものがあるのでそっちに行きましょう。

しゃべってきた

こっちを先に片付ける。ライトニングトークで時間もらったので、ライブドアでHadoopをこんなことに使ってるよ、そのためにこんなツールが欲しかったから作って使ってるよ、という話をしてきた。

Hadoop and subsystems in livedoor #Hcj11f
View more presentations from tagomoris
動画の様子はあとでUstream等で公開されるんだと思う。多分。されないと自分が見てなかったトラックとか見たくて悶死するし。実はあれが自分がプレゼンしたときの聴衆数最大だった気がする。ちょっとビビり気味でしゃべってたので他の人からどう見えてたか不安……。

10分にちょっと詰めすぎた気もするけど、現状の端的な紹介ができたと思う。内容について細かく知りたいという人は個別に声をかけるなりしていただければいくらでも出てくると思います。scribelineについてはこのblogに書いてるけどshibはまだ書いてなかったと思うので、続けてエントリを起こす予定。Beeswaxとの違いとかリクルートさんのWebHiveとの違いとかあるんじゃないかなー、目的特化のツールなんで用途にハマれば便利なんじゃないかなー。

行ってきた

10時のオープニングで参加登録時のアンケートの内容が一部紹介されていたけど、なんとHadoop使ったことない人が半数。うわー、という気分にちょっとなった。まあ冷静に考えてマシン複数台と分散処理対象のデータがないと実際に試せもしないフレームワーク*1なんだから、1000人以上とか集まる方がおかしいという気もする。そこから考えると半数が使ったことないってのもおかしくはないのかな。事例を知りたいんでしょうねきっと。スーツ率高かったし。*2

午前中は Cloudera, HortonWorks, MapR の3社のKeynoteっぽいお話。たいへんビジネス臭のする世界で良いですね。技術的な話はあんまりなくて正直ちょっと退屈。特にClouderaとHortonWorksがパッケージングについて綱引きしてたけど、個人的には本家OSSプロジェクトからrpm/debが出る方が嬉しいなーと思ってる。まあ最終的には、バージョン不整合の問題のないパッケージがどっかから出てくるんならそれでいいやという気分だけど。どうせ頻繁にアップデートとかしないし。MapRは惹かれる話が満載だけど、ミドルウェアにおカネかけてられないからパス。技術的には興味があるけどなあ。
で、ビジネス上の話で綱引きはしつつ、Hadoop関連コンポーネントではClouderaもHortonWorksもそれぞれにコードを提供して進めていっているし、なんかいい関係ですね。フルスタックが完成するまで黙ってて、いきなりドカっと出して「これでいこうぜ！」とか言いだす企業よりはるかに好感が持てる。

午後は正直自分のプレゼンが終わるまでは他人事どころではないので上の空に近かった。が、自分のが終わったらそれなりに落ち着いてふらふら。
で、結局問題になるのはHadoopそのものというよりも、データの収集と変換であったりとか、クエリをどう発行するかとか、そういうことなのかなーとは思った。データがきちんと整備された状態でHDFSに入ってれば、それを処理するロジック自体は生MapReduceでもPigでもHiveでもよくて*3、その処理をどうkickするか、データの整備はどうやってやるのか、に注力せざるをえない。Hadoop上で動くロジックの効率だのなんだのはノード増やせば済むしね。うーむ。
HortonWorksのOwenのプレゼンは来たるべきHadoop 0.23について。次期stableで、2011Q4に出るらしい。アーキテクチャの変更とか解説があったが、いろいろと良い改良が行われるように見える。楽しみ。HadoopがMapReduce専用基盤ではなくて汎用の分散処理基盤のようになる、という理解でいいと思う。ストリーム処理用のアプリケーションとかもやってみれば展開できそうに見えたけど、どうかなあ。

懇親会、二次会

いろんな人といろんな話をしてた。楽しかった。なんかもういろいろあるけど省略。
とりあえず stream processing framework ほしい。という話をけっこうしてた。ClouderaのToddに「なんでFlumeじゃダメなの？」とか言われて、いやあれは too heavy だと思うんだよね、とか言ってもあんまし理解してもらえなかった。OwenがKAFKAはScalaで良さそうだったとか言ってたのでそのうち調べる。あとはJVM上で動くってのは諦めて受け入れるとして、JRubyで書くとかかなあ。flumeかfluentで。

全体的に超楽しかったです。おつかれさまでした！

*1:もちろん動かすだけなら1台でもできるけど、普通に考えてなーんにも嬉しくないよねそれ。

*2:でもHortonWorksのOwenが懇親会二次会で「Tシャツが多かった」と言ってたのを考えるとアメリカではHadoopの集まりはより強烈なスーツの集団なんだろうなきっと。

*3:そういえばPig使ってるって話は自分が聞いたセッションではひとつも出てこなかったなーという気がする。