#hcj13w 午後6.halook

Hadoop Conference Japan 2013 Winterの午後6（会場：7階）『トラブルシューティングのために欲しかった、Hadoopがまるっと分かる可視化ツール』のメモです。
講演者はAcroquest Technologyの落合 雄介さん。（→資料）

halook（ハルック）の紹介。随所でデモを見せてくれて分かり易かった。
落合さんはhalookのロゴをイメージした被り物を装着して登場ｗ（New York Hadoop User group Meetupでも被っていたらしい）

halookは、Hadoop・HBaseの可視化ツール。
HDFS使用量やHBaseリージョンの状態を照会できる。

Acroquest TechnologyはJava Troubleshooting Service（JaTS（ジャッツ））という、Javaに関するトラブルシューティングサービスを行っている。期限内に解決できなかったら半額返し！
現在のところ解決率100％らしい。
で、その為にJavaシステムを動的に診断して「見える化」するツール（ENdoSnipe（エンドスナイプ））を開発している。
なので、同様にHadoopのトラブルシュートツールとしてhalookを作った。
フロントエンドのウェブアプリケーションと同時に統合的な監視が可能。
また、halook公開を機にENdoSnipeもOSS化した。

○HDFS使用量
　データ量やデータの偏りを表示
○MapReduceジョブ
　ジョブの実行時間をガントチャートで表示
　タスクを時系列/ホスト毎に表示
　使用スロット数のグラフ化
○HBaseのリージョン数の推移
　splitのタイミングとか
　リージョンの分布をサーバー毎/テーブル毎に表示（偏りが分かる）

例えば、データノード間のデータ量の偏りを直す「start-balancer.sh」を実行しても すぐにデータ移動するわけではない、というのがツールを見ていると分かる。（十数時間かけて移動していく）

Asakusa Frameworkを動かした際のデモもあった。タスク数とか実行時間を見て、きちんと分散されていることが確認できた。

インストールとしては、「ENdoSnipe Javelin（ジャベリン）」をマスターノード（NameNode・JobTracker・HBaseMaster）に入れる。そこからデータ収集される。バイトコードインスツルメンテーションによってJMX以上のデータを取得できる。
（現在はスレーブは対象外だが、開発中とのこと）
収集したデータはPostgreSQLに格納する。（DataCollector）
クライアントはhalookサーバーにアクセスする。過去データの表示および現在データのリアルタイム表示が出来る。
（画面表示にはWeb Graphical Platform（WGP）というものを使っている。これもOSS化している）

Q&A1
ノード数が多いと表示が遅くなりそう
→2000タスクくらいなら表示できるが、それ以上はこれからの課題。（データ収集は大丈夫だが、見せ方はこれから）

Q&A2
他にも可視化ツールはあるが、halookの売りは？
→他のツールでもマスターのデータは取れる。halookはスレーブも取れるようにする。

Q&A3
可視化に力を入れているが、解析支援は？
→ENdoSnipeはまさに解析支援ツール（メモリリーク・SQLの大量発行・フルスキャン等の検出）だった。
halookはまだそこまで到達していないが、元々のENdoSnipeがそういうものなので、いずれやりたい。 
（これは自分も思った疑問。見えるのはいいが、それが意味するものを読み解くのは大変そう。むしろ人間系のトラブルシューティングサービスで解析支援をやるのが良いような気がした）

#hcj13w 午後5.Asakusa Framework

Hadoop Conference Japan 2013 Winterの午後5（会場：1B）『AsakusaによるHadoopでの業務系バッチ処理の課題と解決』のメモです。
講演者はノーチラス・テクノロジーズの神林 飛志さん。（→資料は非公開だそうです）
なんだけど、資料の内容は社内でダメ出しをくらったらしく、ノーチラス・テクノロジーズの見解でなく神林さん個人の意見ということでの発表だった＾＾；
Hadoop Conferenceはお祭りだからということで、ポジショントークも無しで、神林節全開という感じｗ
のっけから、自己紹介しませんとか、Asakusa Frameworkを知っている前提で話しますとか！（第三者として聞いている分には楽しいんだけど(汗)）
AsakusaFWは裏トラックとか言ってるし。（お金が…とか言っていたのは、会場1AのTwitterのリアルタイム解析の話らしい。Twitter社からツイートを買うのにすごくお金がかかっているらしい）

えー、気を取り直して＾＾；

Asakusa Frameworkの事例で新しいものは出せないけれど、水面下で進んでいるものはあるがNDA（秘密保持契約）があるので表には出せない。夏か秋には出てくるかも。
小規模なものも大規模なものもある。実証はだいたい完了し、本番系への歩みが始まっている。

---

そして問題も顕在化している。

１．スモールデータとショートバッチの問題
実際の大規模な業務バッチのフローの一部のスライドが出ていた。あれでも一部だそうだが。
そういうフローが1500段くらいあるバッチ（Hiveでは無理でしょう）が、AsakusaFWでは書き切れる。その為にこの問題が顕在化した。すなわち、データ量が少なくて実行時間が短いジョブが多くなってしまう。
挙げられていた表は、左の方が実行時間が短いジョブ（一番左は1分以下）で、右の方が長いジョブ（10分くらい）。各列の棒グラフはジョブ数を表している。
右の方はHadoopでなければもっと時間がかかるものなので、それ以上は短くならない。それにジョブ数も少ない。
問題は一番左のジョブで、1つは1分以内だけど7000個もある。これらを合算すると、トータルとしてバッチ全体の実行時間があまり短くならない、ということになってしまう。
短いジョブの個数が少ないなら目をつぶることも考えられるが、往々にして99:1にもなってしまう。 

理想としては、短いジョブにはHadoopを使わないようにすればよい。小さいデータならRDBMSを使う従来のバッチの方が速い。
が、開発中にどの部分が小さいかを予測するのは困難（意識する人は少ない）だし、後から変わっていくこともある。
また、プロダクトを2種類扱うのも敷居が上がってしまう。（AsakusaFWのみで開発したい） 

２．データ転送
データ転送のボトルネックの話。
BIや集計系のバッチでは出力データが減るが、業務バッチでは減らない（むしろ増える）ことがある。
1500段もあると、失敗したときに最初からやり直せばいいという単純なことにはならない。

３．運用ツール
リラン等のやりやすさ。

---

●解決策１．（スモールデータとショートバッチの問題）
小さなジョブはRDBMSで処理すればよい。それを透過的に扱って（自動的に判断して）欲しい。
という訳で、次世代AsakusaFWでは新しくAsakusa実行エンジンというものが出来る。（コードネームはFireworks（花火））
RDBを使って実行するRDB Controllerと、今まで通りHadoopで実行するMapReduce Controllerができ、mediator（調停者）が振り分ける。
（作っている人の名前から、ueshional engine（うえしょなるエンジン）と呼ばれていたこともあるｗ（ちなみに@ueshinさんと言えばHBase。つまりRDB以外にHBaseも対応される可能性が…？！ｗ））
（まぁHBaseはともかく、）他にも速い基盤が出てくれば、それに対応することは考えられる。

これはつまり、実行時最適化を行うということになる。
・トランザクション処理の方式が変更になる
・インポートデータの再利用
・データの先読みも可能

また、Asakusa DSLの中間表現が導入される。
DSLと実行計画が分離されるので、自分用のDSLが作れるようになる（中間表現を出力すればよい）。

Asakusa DSLそのものにも拡張が入る。
・レンジ結合（BETWEENで結合）
・定数表の導入（マスターデータのような感じの固定データをフローの入力とせずに使用できる？）
・任意の演算子で結合処理が可能（今は結合用の演算子が準備されている）

これにより、Hadoopをかなり隠蔽することになる。

●解決策２．（転送ボトルネック）
RDBMS→Hadoopについては、fluentdとか使えばいい。（同時間帯に7階で発表されていたｗ）
問題はHadoop→RDBMSで、RDBがボトルネックになる。Sqoopを使うとかバルクロードとかインメモリーDBを使うとか、総力戦。
（Sqoopは、PostgreSQL用のdirectモードのパッチをNTTデータの方が作っていましたし）
ただ、これらはAsakusaFWは関係ない。

AsakusaFWとしては、データ転送順序の変更で対応する。（解決策１の実行計画に関連）
空いているI/Oがあればそこを使う。（ジョブが始まる前でも（トランザクション等に問題が無ければ）先読みするとか）

また、EAIツールとの連携も考えられる。
・データスパイダー
・アステリア

その他に、共有ストレージを使う。（東芝のは良いらしい。他のは自粛ということで、ここでは書きません＾＾；）
CDHよりMapRの方が速いとか。

●解決策３．（運用ツール）
ジョブ管理基盤を作成中。コードネーム：Nakamise
「なんとか変更履歴のブログを書いている人」とか言って誰にも通じてなくて滑ってたけど、さもありなんｗ（ueshional engineの人とは知名度が違うのでｗ）
ちなみに「UIが得意な人」は、JavaFX関連でTwitterを見ていると見つかるかも？ｗ
で、Nakamiseは、Asakusaアプリを実行・停止したり再開させたりするもの。（YAESSは実行のみ）
JP1/A-AUTO/Senjuと連携する。

---

後は与太話とのことで、AsakusaFWとは関係ない話。
だけど、どこまで書いていいのかなぁ＾＾；　えーと、無難そうなところだけ書きます。。

RDBMSの逆襲が始まる（ただしそのままでは来ない）。
・transaction制御
・logとrecoveryの応用
・query optimization

Impalaは「リアルタイム」ではない。Hiveと比べると速いがRDBMSと比べると遅い。「リアルタイム」という言葉を不用意に使うと色めき立つ人達がいる＾＾；
エンドユーザーの3秒ルールというものがある。（ユーザーは3分以上だと壊れたと思う、30分経つと席を立つ）

AWS
ほぼ一年間使ったが、致命的な問題は無し。キックでこけることがあって、それは自分で何とかする必要がある。

#hcj13w 午後4.Mahout

Hadoop Conference Japan 2013 Winterの午後4（会場：1B）『Hadoopで拓く新しい分析基盤の未来』のメモです。
講演者は新日鉄住金ソリューションズ（NSSOL）の東 英樹さん。

Mahout（マハート）の話。
Mahoutの名前は知っているけれども、機械学習ということ以外は何も知らなかったので、どういうものなのかを窺い知ることが出来た。
Mahoutでは機械学習のアルゴリズムが何種類か実装されており、利用者がどれを使うかを指定する。

例として分類処理というものを扱っていた。
トレーニングデータ（学習用データ）をMahout（分類アルゴリズム）に与えてモデルを作る。
そしてテストデータ（分類対象データ）をモデルに適用して分類する。

また、異なるアルゴリズムを合成すると精度が良くなるという話もあったが、そんなことは考えたことも無かったので新鮮だった(笑)

#hcj13w 午後3.LT

Hadoop Conference Japan 2013 Winterの午後3、会場1BはLTでした。

---

１．『MapReduce optimization by node/rack-level aggregation』
講演者はNTTの小沢健史さん。

Shuffle処理はボトルネックになる可能性が高い。（Mapperのノードから読み込む為にファイルI/Oが発生、転送する為にネットワークI/Oが発生、Reducerのノードに書き込む為にファイルI/Oが発生、さらにそこから読み込む）
このため、Combinerでデータ量を減らしてI/Oサイズを減らすという仕組みがある。

しかしCombinerは1タスクで1個しか動かない（ファイルが1つしか作られない）。
今のマシンはコア数が多いのに使われないということになり、もったいない。
というわけで、Combinerを改良するパッチを作った（提案中）、という話！

方法としては、遅いネットワークで転送する前に（高速なネットワーク経由で）中間集計用ノードにデータを一旦集めて集計を行ってから遅いネットワークで転送する、というもの。
コーディングとしては設定を2～3行追加するだけで使用できるようにする。
ついでにCombinerのオーバーヘッドを減らす実装も入れている。

これでベンチマークをとったらおよそ1.3倍（最大1.5倍）速くなったそうだ。（どんなデータ量でも程度の差はあれ速くなっている）

未完了事項（TODO）としては、テスト作成中。分散環境なので難しい。
また、Hive/Pigで使えるようにする。（グループ処理が速くなると期待）

これが導入されたら、素直にCombinerを使っているアプリケーションは速くなるので、期待大です。

---

２．『High Availability for the HDFS NameNode』
講演者はClouderaの小林大輔さん。 

NameNode HAの紹介。
Hadoop2.0（CDH4）で導入された。（従来のHadoopではNameNodeがSPoFだった）

クォーラムベースジャーナリングという仕組みになっている。
外部の（特別なHA用の）ハードウェアは使用しない。
JournalNodeという、editsログを書き込むノードを複数用意する。（今まではNameNodeのローカルに書いていた）
JournalNodeは独立したマシンである必要は無く、NameNode等の（比較的信頼性が高い）マシンに同居させることが出来る。
NameNodeにはJournalNodeと通信する為の「Quorum JournalManager」を稼動させる。
各JournalNodeへ書き込み、過半数（quorum）から成功が返れば書き込み全体が成功したものとする。

NameNodeはSPoFでないようにする為に2台用意する（アクティブ・スタンバイ）。
2台から同時に書き込まれると不整合になってしまうので、editsを書き込めるのは1台だけになるような仕組みが必要。これがフェンシングと呼ばれる仕組み。
（Paxosと同様に）NameNodeにエポック番号を割り振る。NNがアクティブになる度にエポック番号を増加させていくので、複数のNNが同じ番号になることはない。
JournalNodeでは最新のエポック番号を保持するので、古い番号のNNから書き込みが来てもエラーに出来る。 

NameNode HAというものが導入されたのは知っていたけれど、どういう仕組みかは勉強していなかったので、ためになりました。

---

 ３．『メインフレームとHadoopで大量のPDF帳票を出力してみた』
講演者はJSOLの三木大知さん。『パターンでわかるHadoop MapReduce ビッグデータのデータ処理入門』の著者だそうです。

メインフレーム帳票（紙）を出力していたが、PDF用データベースを導入して画面で見られるようにした。
で、あるとき、DVDに焼いてくれと頼まれた。画面表示は1ページ0.1秒、400万ページあったので、単純計算でも110時間はかかる。また、DBには特別な形式で保存していたので、Hadoopで変換することにした。

PDF用データベースはDB2だったので、DB2のエクスポート機能を使ってファイルを出力してHDFS上に乗せた。ファイルは2種類あって片方はインデックスのようなものなので、HadoopのReaderを独自実装して上手く読み込めるようにした。

Hadoopで失敗したときの為に用意した変換プログラムでは60時間かかったが、Hadoopだと20分（データノードを減らしても1時間）で出来たとのこと。

自分も昔メインフレーム（汎用機）で帳票を出力するプログラムを作っていたので、よく理解できる話でした＾＾；