logstash を使って手元にあるファイルを elasticsearch に取り込むを読んで，確かに標準入力から読み込めるとちょっと便利だなぁと思ったので標準入力からログを受け取るstdinプラグインを書いた．

repeatedly/fluent-plugin-stdin

fluentdにはin_tailがあるので，複数ファイルを一気に転送とかだとそっちの方が楽だけど，1ファイルだけこまめに転送したいとか，テストでファイルを使いたい，みたいな場合に毎回設定ファイルを変更するのも面倒なので，そういう時にはこっちの方が便利．

インストール

$ fluent-gem install fluent-plugin-stdin --no-document

使い方

上記の記事のLTSVの読み込みと同じようなことをする設定は以下(useragentフィルターが何やっているか分からなかったのでそこはスキップ)．

# stdin.conf
<source>
  type stdin
  format ltsv
  types status:integer,size:integer # 任意のフィールドを数値に変換．ここではstatusとsize
</source>

<match stdin.**>
  type stdout
</match>

これに対して，以下のように実行してLTSVを流し込むと，標準出力に結果が出力される．

% cat ltsv.log | fluentd -c stdin.conf -q --no-supervisor
2015-01-01 12:00:00 +0900 stdin.events: {"host":"192.168.0.1","ident":"-","user":"-","req":"GET /index.html HTTP/1.1","status":200,"size":342,"refer":"http://www.example.com/","ua":"Mozilla/4.08 [en] (Win98; I ;Nav)"}

EOFに到達するとバッファをフラッシュしてストップするという少しトリッキーなプラグインになっている(stop_at_finishedをfalseにすれば何もしない)．
stdoutをelasticsearchにすれば，Elasticsearchにローカルのファイルをどんどん転送出来て，バルクロードっぽいことが簡単に実現できる．

コアにあってもいいかな？と思ったけど，まぁそんなにユースケースはないので，とりあえず3rd partyのプラグインとして公開した．

バルクロードとFluentd

Fluentd(に限らずscribe, flume, logstashなどのログ収集ミドルウェア)は，ストリーミング的にデータを流すことを目標にしているので，バルクロードには向きません．もちろんそれっぽいことは出来ますが，パフォーマンスやトランザクションを含めた堅牢性などを考慮すると，相性が悪いです．実際stdinプラグインを書きましたが，バルクロードするのにこれを使うのは個人的にはオススメしません．
ということでそこを埋めるために，バルクロードに特化したツールEmbulkをOSSで近々公開します．1月27日にあるイベントで発表・公開するので，もう少しお待ちください．

Fluentd/Embulkのペアで，データ収集・転送の敷居をもっともっと下げて行きたいと思っているので．是非フィードバックを頂ければと思います :)