Hadoop Streaming（あるいはC言語 vs Java）

Hadoopアドベントカレンダー2011の26日目です。

自分はJavaをそこそこ使ってきたのでHadoopを扱うにもJava APIを使うのが自然な選択でしたが、世間ではHadoop Streamingを使ってJava以外の言語で書くことも多いようです。
という訳で、ちょっとHadoop Streamingを試してみました。

---

まずはストリーミングってどうやるんだろう？というのをざっと調べる為に「Hadoop Streaming」でググってみたら、さすが色々ありますね～(笑)
Perl・PHP・Python・Ruby、JavaScriptなんてのも！

どうやらhadoop-streaming.jarというJavaプログラムに“各言語で書いたMapper・Reducerプログラム”を引数として渡すようですね。ふむふむ。

と思ってサンプル（おなじみWordCount）を見ていたら、2点ばかり気になることがありました。

---

1つはReducerでの集計方法で、連想配列（ディクショナリーやマップと呼ばれることもある）を使って単語毎に件数をカウント（保持）していることです。

それって、単語数が多かったらメモリー不足で落ちたりするんじゃないですかね(汗)
（Hadoopは大量データを処理するものですから、そういう所は気にしましょう。メモリー内に入りきるようなデータ量だったら、そもそもHadoopを使う必要ありません）

せっかくキーでソートされて入ってくるんだから、キーブレイク処理にしましょうよ。

---

もう1つは、カウント時に（Reducerの入力値を使わず）直接1を加算している例が多いことです。
これじゃ何の為にMapperで1を出力しているんだか分かりませんよ＾＾；

件数カウントをSQLで例えると、通常は「select key, count(*) from t group by key」という風に書くと思いますが、別の方法として「select key, sum(1) from t group by key」と書くことも出来ます。
Mapperで出力している1は、後者の「sum(1)」の1に当たります。

それに、Reducerで入力レコードをカウントする方法でも結果は正しく出ますが、その方法だとCombinerを使ったときにおかしくなりますよね。
と思って実行方法の方を見たら、案の定、Combinerは指定していませんでした。

…というか、どうやら古いバージョンのHadoopでは、CombinerにはJavaのクラスしか指定できなかったようです。それじゃ、ストリーミングのサンプルとしては当然記述しないですよね～＾＾；
象本によると、Hadoop0.21でCombinerにJavaクラス以外を指定できるようになったようです。また、CDH3も（Hadoop0.20ベースですが）大丈夫でした。
なので、それらのバージョンを使っている人は、ぜひCombinerを指定しましょう。2倍くらい速度が違いますよ！

---

という訳で、ストリーミングを使ってWordCountを作ってみました。
対象言語は、実行速度が一番速いという噂のC言語(笑)（→ソース）

まぁ自分は仮想分散環境しか持っていないので速度を比較するには不適切なんですが、それを踏まえて、結果は、Javaで作ったWordCountとほぼ同じ速度でした。
160MBのファイルをデータノード3台で処理して、Javaが23秒、C言語が25秒という感じです。ちなみにawkだと34秒くらい。

結論。
ストリーミングは標準入出力を使うのでその分は遅いと思いますが、ネイティブな実行ファイルを作るような言語なら、Java APIを使うのとほとんど同じ速度が出るようです。
すごいですね！（どっちが？ｗ）

P.S.
しかし、ストリーミングを使うということは素のMapReduceを書いているのと結局同じなので、そういう意味ではPigやHiveを使う方が保守性は良いような？(爆)

AsakusaFW0.2.4 WindGate

当ブログはHadoopアドベントカレンダーの21日目として書いています。（本当は12/21に書きたかったけど、忘年会で酔っ払ってて…＾＾；）
Asakusaフレームワークの0.2.4が12/19にリリースされましたので、自分が注目した点を3点ばかり書きたいと思います。

---

まず、ドキュメントが増えました！

自分が使い始めた0.2.1の頃は、むしろドキュメントがある事に気付かず、一所懸命ソースを追っていて、後からドキュメントを知って「ちゃんと載ってるじゃん！」と愕然としたものです(爆)

今回はページも増えて、各章の見出しの大小も分かりやすくなりました。
まぁ、読む量が増えて大変になったと言えなくもないですが＾＾；

---

そして、WindGateがCSVファイルに対応しました！！

今まではAsakusaFWのファイルはSequenceFileがベースだったので、テキストファイルを扱おうと思ったら自分でImporterやInputFormatをコーディングする必要があり、ステップ数がかさんでいました。
今回、その辺りも自動生成されるようになったので、自分で記述する必要がある部分はごくわずかになりました。

しかもこのCSVパーサーが意外と高機能で、ダブルクォーテーションで囲まれた項目にも対応していますし、UTF-8以外の文字コードにすら対応しているようです。（Hadoopは（PigもHiveも）基本的にUTF-8が前提）
ファイル名をデータとして取得することも出来ます。ファイル名に店舗コードとか日付とかを付けて識別する業務で使えるかも？

---

最後に、アーキタイプWindGateで作られるサンプルアプリがCSVファイルベースのものになりました！
（メソッド名・変数名やJUnitの構成も実用的なコーディングの参考になるよう変更されたみたいです）

0.2.1のアーキタイプbatchappはMySQLベースだったので、MySQLの設定をするか、MySQLを使わないように設定変更したりダミーのDMDLを作ったりする必要がありました。
が、今回のWindGate版はそのようなことをせず、そのまま（テストまで）実行できます。
また、0.2.3のWindGateはPostgreSQLベースだった模様です。

これで、AsakusaFWもかなり試しやすくなったのではないかと思います。まだ触ったことの無い人は、これを機会に見てみるとよいのではないでしょうか(笑)

第1回EMR勉強会のメモ

Amazon Elastic MapReduceの勉強会『第1回EMR勉強会（Hadoop on AWS）』に参加しましたので、そのメモです。（Twitterのタグ→#emrstudy_jp、他の人のメモ→くろのさん）
（会場の最寄り駅はテレコムセンター駅。ゆりかもめは国際展示場正門以外で初めて降りたかもｗ）

---

最初はクリエーションライン株式会社の李さん。

まずAmazon Web Service（AWS）の簡単な紹介として、AWSはWeb系・業務系・Highパフォーマンス（並列）・BigData等、色々な分野で使われている。
EMRはクラウド型Hadoopサービス。

プログラムや入出力ファイル・ログはS3に格納する。（HiveのメタデータはRDSに置く）
プログラムはStreaming・Hive・Pig・Custom JAR（自作Map/Reduce）・Cascadingが使える。
◆●初めてEMRのコマンドを見たけど、自分のローカルPC上にEMRのコマンドをインストールしておくのかな？
「elastic-mapreduce --create ～」を実行するとHadoopクラスターが構成される。その引数に入出力ファイルのパスとかプログラムのファイル名を指定する。
実行するプログラムや入出力ファイル名をJSON形式のファイルに書いておき、それを指定することも出来る。
◆●ファイルは、「s3://～」で直接S3の場所が指定できるんだねぇ。

運用面は、ジョブ管理にAWSコンソール・Ruby CLI（コマンドライン）・APIが使える。
Gangliaで統計情報も見られる。

EC2インスタンスタイプによりインスタンス当たりのMapper数・Reducer数が決まっている。
BootStrapにより、Hadoopの個別の設定をすることも出来る（起動前に設定が読み込まれる）。

---

次に、Amazon Data Services Japanの大谷（shot6）さん。EMR利用事例。

• Foursquare（スマートフォン）
  • 機械学習・データ分析・トレンド分析等に使用
  • 平均40ノード（増減する）
  • RubyでStreaming
  • ログ収集はApache Flume、ログ保存はS3、ログ解析はEMR、結果を見るときはHive
• Razorfish（広告のSI）
  • 1日35億レコード、170万広告
  • 100ノード
  • 処理時間：2日かかっていたのが8時間に
  • HBaseを使ってるみたい
• Sonet
  • 広告配信ログの分析
  • 1日平均10GB、年3.65TB
• Etsy（巨大小売業）
  • 434GB
  • Ruby（Sinatra）
• Yelp（地理データ）
  • 1日400GB
  • 全部EMR。1週間毎にノード数を変更している
• 名前は明かせないけど、金融系
  • 100％AWS
  • 60年分のデータ（100万ロケーション）
  • 1200～1800インスタンス
• Hatena
  • Perl

Streaming・Hiveが多く使われていて、MapReduce直接は少ない。
Cascadingもけっこう使われている（日本では少ないが）。

Hadoopクラスターを起動したら、最初にS3からHDFSへデータをコピーする必要がある。
（EMRを起動しないと何も存在しない状態だから）
EMRを起動させっぱなしであれば、HDFS上にずっとデータを置いておくことも可。

---

次はクックパッドの佐々木（sasata299）さん。→資料（からあげ！ｗ）

“たべみる”で1年分の検索データを分析。
2009/09：MySQLのGROUP BYを使って処理しようとしたら、7000時間（約1年）かかるという見積もりｗ
2009/10：Hadoop（CDH1）のStreaming（Ruby）をEC2上で実行、30時間で出来た。
2010/07：CDH1のバグに遭遇（大きいデータを扱うとSocketTimeoutExceptionが頻発）。解決策はCDH2かEMRを使うこと。比較した結果、コストはEMRの方が少し高いが安定性やバージョンアップが自動で行われる点でEMRを選択。
「環境構築をしたいのではなく、データ分析をしたい」
2010/08：EMR使用

2010/4にブログで「EMRを使わない3つの理由」を書いたが、「使う理由」に訂正したいとのこと(笑)

MySQLで出来ることはMySQLでやり、出来ないことをEMRでやる。

クックパッドさんの中では、各エンジニア（全エンジニアは40人で、その中の10人くらい）がRubyでそれぞれEMRを使っている。（クックパッドのエンジニアは皆Rubyを使える。EMRの使い方ドキュメントは用意してある）
また、エンジニア以外でも使えるようにI/Fを用意している。

---

次はヴェルク株式会社の津久井（quarterkota）さん（インフラ設計）・石田（o918）さん（アプリ設計）。→資料

ログ解析で、アクセスログをS3に転送し、EMRで処理して集計結果をS3に格納する。
集計管理サーバー（ELB/EC2/EBS/RDS）でEMRを監視している。
利用時のみ起動しているので、運用コストが安い。1時間8.8円なので、8台2時間で140円程度。

アプリとしては、Hiveを使用。
EMRを常時起動させてはいないので、CREATE TABLE・INSERTでデータを保持しておく方法は使えない。
S3上にHiveが認識できる形でディレクトリーを作り、データを格納しておく。

• バケット/ACCESS_LOG ←テーブル用ディレクトリー
  • ACCESS_YM=201111 ←パーティション用ディレクトリー
  • ACCESS_YM=201112
    • ログ1 ←パーティション項目は入っていない
    • ログ2

パーティションは、「パーティション項目名=値」というディレクトリー名にする。

そして、elastic-mapreduceコマンドでHadoopクラスターを作成する。同時にHiveのスクリプトを指定して実行する。
まず「CREATE EXTERNAL TABLE ～ LOCATION 's3://～'」でS3上のディレクトリーを指定したテーブルを作成する。
ただしそれだけだとパーティションが認識されないので、「ALTER TABLE テーブル名 RECOVER PARTITION」でパーティションを認識させる。
そしてINSERT DIRECTORY ～ SELECTでHQLを実行する。

◆●Hiveのパーティションは使ったことが無かったけど、RECOVERなんて命令があるのか。

---

最後は株式会社gumiの本間（CkReal）さん。ソーシャルゲームのEMR活用事例。→資料

今までのgumiの課題として、

1. ユーザーが（GREE経由で）カスタマーサポート（CS）に問い合わせをする。
2. CSはエンジニアに調査を依頼する。
3. エンジニアはNFSサーバーにあるログを調査する。
   しかし、NFSサーバー上のログは毎日ギガバイト単位（最大18GB、圧縮して2.4GB）で発生するので、grepするのも大変。

そこで、MongoDBにJSON形式で格納することとし、EMRで変換している。

gumiは全てAWSを使っているので、EMRにするかEC2にするかという問題もあったが、EC2で構築するのは大変（あと、たまにEC2が再起動されたりする）なので、EMRに。

使用している言語はPython。（gumiのアプリは全てPythonで動いているし）
（Pigは習得コストがかかる。HiveはSELECTする為にある程度ログが整形されている必要があり、最終的にJSONに変換するという目的に合わない）

◆●Hiveを選択しなかった理由が「目的に合致しているかどうか」なのはいいね。
◆●Pigは自分も覚えるの面倒だと思ってたけど、やってみたらそんなに難しくは無かったけど…。

システム構成は、

1. NFSサーバー（複数）からバッチサーバーへログを転送
2. バッチサーバーでgzip圧縮し、S3に溜めておく（2千万件）
3. EMRで処理（2時間くらい）し、S3に集計ログ（30万件）を保存
   （EMR起動時にBootStrapでPython2.7をインストールしている）
4. バッチサーバーでS3からMongoDBへ格納

EMRの感想としては

• S3上のファイルをいつでも利用できる（EC2⇔S3は20MB/sで転送できる）
• Hadoopクラスターを管理する必要が無い
• 変化する要件に対応しやすい（データはS3上にあるので、そのまま使える）
• たまにジョブが失敗する（Reduceが終わらない？集計ログを回収しきれない？）
• CPU使用率：リニアにスケールさせるのは難しい
  ※チューニング方法模索中（Hadoopにあまり詳しくない）

◆●Reduceが終わらないというのは自分も経験したことあるけど、仮想環境の設定がおかしかった為だからなぁ。EMRがそのレベルで設定が足りないとは思えないけど。

---

BigDataの各自の定義
大谷さん：持っているのが苦しくなってきたらBigData
津久井さん：MySQLが動かなくなる数千万レコード（少ないならMySQLを使う方が楽）

---

EMRの説明だけでなく、具体的な（細かい）使い方を知ることが出来たので面白かった。

ただ、Hadoopはちょっとかじっているので少しは分かっているつもりだけど、AWS用語は不勉強で分からないものがあった(汗) EC2やS3は有名なので知っていたが、RDSって何だろう？けっこう頻繁に出てきてた感じがするが…。
「elastic-mapreduce」コマンドは初めて見たが、分かりやすかった(笑)（自分は実際に見てみないと理解できないんだな(苦笑)）

最後に、VELCの銘が入ったチロルチョコおいしかったです(笑)
ありがとうございました。

DOM廃止

プログラミング言語比較のページでは、表に各プログラミング言語のキーワードを並べている。ただ、言語が多くなると横に長くなってブラウザーからはみ出して見づらくなる為、指定した言語（列）だけ表示する機能を入れていた。

しかしそれを実現するのにMicrosoftのXML DOMを使い、表のデータをXMLファイルから読み込んで特定の列だけ抽出するようにしていた為、IE（のバージョン5以降）でしか使えなかった。
自分がIEしか使ってなかった頃はそれで良かったのだが(爆)、最近ではIEを使っていないので、何とかしたいと思っていた。

で、素直にJavaScriptを使ってタグの表示・非表示を切り替えるように修正した。
XMLファイルやXSLファイルが不要になったのでファイル数が減ったｗ

XMLファイルを使っていた頃は htmlファイルを読み込んでXMLファイルを生成するプログラムを用意していたのだが、これは結構やっつけ仕事プログラムで(苦笑)、このファイル限定のタグ解析をしていた。
今回もテーブルの各セルにidを付けるプログラムを作ったが、これは自作のHTMLパーサーを使ったから、まだましかなー。（このパーサーは、今回の様に属性値を変えたり追加したりする為に作ったので）
ただ、久しぶりにこのパーサーを使ってみたら、メソッドが不便だったんでちょっと追加してしまった。最低限しかAPIを用意していないJDK1.5から便利メソッドが追加されたJDK1.6になったような気分ｗ

AsakusaFWのフレームワークAPI

Asakusaフレームワークには、「フレームワークAPI」という便利なAPIが存在する。
…教えてもらうまで気付かなかったけど、ちゃんとドキュメントに載ってるし…！ よく読んでなくてすみませんorz
（Importerで引数を取得する手段が無いかと思って親クラスは色々調べたんだけど、全然関係ないクラスのstaticメソッドは、さすがに見つけられなかった＾＾；）

今のところ、フレームワークAPIではレポートAPI（ログ出力）とコンテキストAPI（バッチ引数）が提供されている。

レポートAPIは、ログを出力するもの。異常データを見つけたときのエラーメッセージ出力とかに使える。
また、いわゆるprintfデバッグが出来るｗ ジョブフローの途中でデータをログ出力することも出来るので、いざとなったら役に立つかも。

コンテキストAPIは、バッチの引数を設定・取得するもの。
例えばデータをソートして先頭n件抽出する、という時のnを外部から指定できる。
他にも判定に使う条件値を渡して色々出来そう。

また、WindGateではファイル名部分も外部から指定できるらしい（まだ試してない）。
例えば障害が起きたときに、緊急避難的なデータを入力にする為にディレクトリーやファイル名を一時的に（リコンパイル抜きで）変えられたりするわけだ。
（この辺り、汎用機のバッチがJCL（シェルみたいなもの）でプログラムとそれに渡すファイル名を指定する仕組みになっていて、いざとなれば臨時作業用のJCLを作って実行するのと同じようなイメージらしい）

往年の技術も応用が利くものは取り込んでいる訳ですね～。