みんな大好き fluentd は素敵だと思うんだけどHDFSへの書き出しをするプラグインが無い！ なんで！ という全世界100万人のエンジニアの怨嗟の声が聞こえてきそうだったので、とりあえずどうにかするべく書いた fluent-plugin-hoop がだいたい動くようになったのでリリースした！
- fluent-plugin-hoop | RubyGems.org | your community gem host
tagomoris/fluent-plugin-hoop · GitHub→ 公式リポジトリの仲間入りしました fluent/fluent-plugin-hoop · GitHub
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時間で億の桁にいってしまってあんまり役に立たなくなってしまったけど。
