AsakusaFW0.2.6の見どころ

Asakusa Frameworkの0.2.6がリリースされたので、個人的に興味があるところを見てみる。

まず、基本となる部分では、対象のHadoopのバージョンがCDH3u4になった。最近バージョンアップされた最新版ですね。 

---

 コーディングに関係するところでは、Direct I/OでSequenceFileが扱えるようになった。

と言っても、扱いたいWritableクラスとAsakusaFWのModelクラスとの各フィールドの移送は自分で書かないといけないけど。CSVと違って色々な型や形式が採れるので、プログラマーが指定せざるを得ない。
でもこのおかげで、例えばSqoopで生成するSeqFile・Writable等を自由に扱える。

あと、Direct I/Oで出力ファイル名を指定する際にランダムな値が指定できるようになった。
これで、（プロパティー（データ値）に依存せずに）複数のファイルに分割できる。
特に1ファイルにまとめる必要が無い場合は、複数ファイルに分けておく方が効率が良いことがあるからねぇ。

もうひとつ、コア演算子を扱うための便利クラスが追加になった。
今まではcore = new CoreOperatorFactory();とした上でcore.stop(in);の様に記述していたが、 CoreOperatorsというクラスが追加になり、各演算子はそのクラスのstaticメソッドとして用意されている。つまりCoreOperators.stop(in);という様に書ける。

---

内部実装でも効率改善の機能として、入力スプリットの結合機能が加わった。これはHadoopのCombineFileInputFormatに相当する機能。

通常だと、入力対象ディレクトリー内に複数のファイルがあると、Mapタスクはそのファイル数（およびブロック数）分作られる。しかしそれらのファイルが小さくて沢山あったとしたら、Mapタスク起動のオーバーヘッドの影響の方が大きくなってしまい、効率が悪い。
そういう時に、1つのMapタスクで複数のファイルを読み込むと効率が良くなる。
（そして結合してある程度大きいファイルになれば、後続処理でも効率が良くなる） 

---

目立たないかもしれないけど重要な実行時の機能として、YAESS（AsakusaFWアプリを実行するツール）が出力するログのログコードが規定された。
「YS-CORE-I01999」でバッチ、「YS-CORE-I02999」でジョブフロー、「YS-CORE-I03999」でフェーズ、「YS-CORE-I04999」でジョブ（ステージ）の終了メッセージが出る。elapsedでその処理にかかった実行時間も出るので、メトリクス収集に便利。（数値がカンマ区切りなのがちょっとだけ面倒だけど＾＾；）

ちなみに、Asakusa DSLがBatch DSL＞Flow（FlowPart）DSL＞Operator DSLという階層構造になっているのと似た感じで、実行時の階層がバッチ＞ジョブフロー＞フェーズ＞ジョブ（ステージ）となっている。ジョブ（ステージ）がMapReduceジョブに相当する。
フェーズは各ジョブフロー内の初期処理・主処理・終了処理に相当する。（mainフェーズで複数のジョブ（stage0001～）を実行する）
（→フェーズ名の一覧） 

ちなみに、ジョブフロー単位で実行できるyaess-flow.shが加わっている。

---

実行計画に関係するものとしては、フローの図が出力される場所が変わった。
今まではコンパイルしたtargetディレクトリーのvisualize内にdotファイルが作られていたが、$ASAKUSA_HOME/batchapps/バッチID/opt/dsl-analysisの下に配置されるようになった。
つまり、YAESSで実行するために$ASAKUSA_HOME/batchappsにアプリのアーカイブを展開するが、その中に入っていることになる。

また、dotファイルの他にtxtファイルも生成されるようになった。ここには、フロー内に含まれるフェーズやステージのクラス名、さらにステージ内に含まれる演算子のクラス名・メソッド名が書かれている。
どのMapReduceでどの演算子が実行されるのかは、ここを見れば分かるわけだ。 

targetの下だけでなくバッチのアーカイブにこれらのファイルが含まれている理由は、target内だとコンパイルする度に変わってしまうが、batchappsの下であれば、リリース時点の状態の図がそのまま残るから。したがって、障害発生時や後から性能を調査する時などに使える。

---

さて、一番最後に、一番の目玉。ジョブ実行クラスタの振り分け（マルチディスパッチ）！
Hadoopクラスター自体を複数用意しておき、ジョブフロー毎に実行クラスターを指定できるというもの。
そんな環境あるんかいなー？と思うところではあるが＾＾；、主な想定は、通常のHadoopクラスター（分散環境）＋スタンドアローンモードHadoopという事らしい。

AsakusaFWで既存システムのバッチ全体を記述すると、少量データを扱うMapReduceがけっこう出来てしまうらしい。こうなると、MapReduceジョブを起動するオーバーヘッドの方が大きいということになる。
そこで、そういう小さいジョブをスタンドアローンHadoopで実行しよう（データはHadoopクラスター上のHDFSを使用する）、スタンドアローンモードだとオーバーヘッドが無くて（小さなデータなら）実行が速いから。ということらしい。
（確かに、自分が昔試しに測定したものも、スタンドアローンの方が速かったなぁ…。1ジョブで10秒の差があったとしたら、30ジョブあれば5分は変わってくるもんなぁ） 

なお、スタンドアローンHadoopを使うに当たっては、スタンドアローンHadoopが動くサーバー上にHadoopをキックする仕組みが必要ということで、用意されたのがJobQueue（地名じゃない！？）。
JobQueueクライアントをYAESS側に置き、JobQueueサーバーをスタンドアローンHadoopに置く。
JobQueueサーバーはサーブレットコンテナ（Tomcatとか）上で動いていて、HTTP（HTTPS）でジョブを受け取る仕組み。
ちなみに、JobQueueサーバーのアプリはPlay Framework（Scala）で作られているらしいぞ！ｗ（warファイルの中にScalaやPlayのjarファイルが含まれている）
（JobQueueに関しては、Scalaが使われている事に一番興味があったりして(爆)）