23:50)river-wikipediaの前々回の記事で書きましたが、bulk_sizeに関連して登録件数がやけにきりが良いのが気になると書いていました。Riverの仕組みを勉强がてら、elasticsearch-river-wikipediaのソース(1.2.0)を読んでみました。Riverの作りRiverはorg.elasticsearch.river.Riverというinterfaceを実装することで作らています。Riverがinterfaceとなっていますが、o.e.river.AbstractRiverComponentというクラスを継承して作られています。AbstractRiverComponentにはRiverの名前や設定などが用意されています。ま、ここはそれほど重要じゃないので、軽く流してと。Riverの設定関連は実装したRiverクラス(ここでは、WikipediaRiverクラス)のコンストラクタで、設定値の読み取りなどの記述を記載します。このコンストラクタが、_river/hogehoge/_metaをPUTした時のJSONを元にElasticSearchから呼ばれて、Riverのインスタンスが作成されます。(たぶん、このへんがその処理だと思う。このあたりはまた今度)実際のRiverの処理はWikipediaRiverクラスのstart()メソッド内部に記述されています。}内部ではインデックスの作成バルクアップデート用クライアントの設定WikiXMLのパーサの初期化ページごとにキックされるコールバック処理の登録デーモンスレッドの起動と起動といった処理の流れになっています。このスレッドの起動後は、4.で用意したparser.parse()処理がグルグル回ります。1ページがパースされるたびに、WikipediaRiver.PageCallbackクラスのproess()メソッドが呼ばれます。このメソッドの最後で、processBulkIfNeeded()メソッドが呼ばれています。ここで、実際にパースしたページをインデックスに登録する処理が実行されます。このメソッドの1行目が鍵でした。bulkSize以上の件数がバルクのリクエストに貯まった時だけ、実際にインデックスに登録する処理が実行されます。このため、スレッドが回っている間は、bulkSize以上のデータが貯まらないと、インデックスへの登録は行われないわけです。次に、このスレッドを止めるには、前々回書いたように、_riverにPUTした、Riverの設定をDELETEするしかありません。(あとは、ElasticSearchを停止するとかでしょうか。)で、DELETEが実行される呼ばれるのが、WikipediaRiverクラスのclose()メソッドです。}見ていただくと分かりますが、スレッド止めて終了です。問題点は?ということで、WikipediaのXMLを読み込んでもRiverは停止しないRiverの停止を行ってもスレッドが止められるだけ。bulkSize以下の件数がcurrentRequestに残っているけど、破棄されるとまぁ、こんな流れになっているので、最後の端数のドキュメントがインデックスに登録されないようです。(まだ、ちゃんと確認してないんですが、備忘録のため先に書いちゃいました。)じゃあ、全部うまく登録するにはどうしたもんかなぁと。いまのところ思いついたのはこんな感じです。他にいい案があったら教えて下さい。案1:close()処理の中で、スレッド停止後に、currentRequestに貯まっているデータをインデックスに登録しちゃう案2:bulkSize以外に、定期的(指定された時間)で登録処理を実行してしまう。簡単なのでとりあえず、案1を実装してみるかなぁと。(さっさとコード書けよって話ですね。その前にMLで質問ですかねぇ、英語で。WikipediaのRiverをざっと眺めてみた感じですが、わかりやすい作りだなぁと。他のRiverがどうなってるかをちゃんと見てませんが、他にもbulkSize指定をするRiverの場合は、このように件数がbulkSizeに満たない状態ではデータが登録されないといったことがあるかもしれません。ElasticSearchのソースを読み始める取っ掛かりとしては面白いかと思いますので、興味ある方は読んで作ってみるといいかもしれません。21:00)MLで質問してみました。pages他のRiverでは対応してるしバグだね、Issue上げてとのことで、あげときました。ついでにプルリクも出せばいいんでしょうが、プルリクまだやったことないヘタレです。あと、案2についても同じトピックで質問してます。BulkProcessorにその機能があるよと。flushintervalというプロパティがありそうです。調査してブログorLTかな。udpにはその値を設定できそうなのがあるんだよなぁ。23:50)さっそく修正版がコミット(コミットログ)されてました。BulkProcessorにflush_intervalの設定をすれば、よしなにやってくれる仕組みがすでに実装されているようです。bulkSizeについても同様に、BulkProcessorに設定すれば良いようです。Riverの仕組みが結構スッキリしています。実装されていた、bulkSizeごとの処理も消されています。確かに、BulkProcessorの仕組みとして実装されている方がしっくりきますね。いうことで、考える暇もなくコミットされてしまいました。やって質問しつつ、少しずつソースを読んでいこうかなと思ってるとこです。