Hadoopクラスタの構築・実行
Hadoopの起動
では、実際に起動してみましょう。「id_rsa-gsg-keypair」を配置したディレクトリで、下記のコマンドを実行します。
$ ./hadoop-0.17.1/src/contrib/ec2/bin/hadoop-ec2 launch-cluster test-cluster 2
このコマンドの第2引数はクラスタの名前（自分で自由に決めて問題ありません）、最後の数字がマシンの台数です。
これだけです。しばらく待つとHadoopクラスタがEC2上に構築されます。
Hadoopでジョブを実行
Hadoopでジョブを実行するには、マスターにログインする必要があります。loginコマンドを使用します。
$ ./hadoop-0.17.1/src/contrib/ec2/bin/hadoop-ec2 login test-cluster
ログインしたマスター上でHadoopのサンプルプログラムを実行します。
# cd /usr/local/hadoop-* # bin/hadoop jar hadoop-*-examples.jar pi 10 10000000
自分で作成したHadoopプログラムも、マスターに転送して同じように実行できます。
また、クラスタのステータスは、
$ ./hadoop-0.17.1/src/contrib/ec2/bin/hadoop-ec2 proxy test-cluster
として表示されるアドレスで確認することができます。
EC2/S3でHadoopを利用する際のコスト
blogeyeではEC2上に構成したHadoopクラスタを通常時は4台、著者属性推定ジョブや実験のためのジョブを投げるときは最大100台近くまで拡大して運用しています。
ただし100台起動した場合1日で240ドルかかってしまうので、注意が必要です。S3のストレージは非常に安価で提供されており、またEC2からの読み出しは無料で行えるため、EC2で処理する大量データを保管する先としては最適と思われます。
blogeyeにおけるHadoopクラスタの利用
さて、クラスタの構成法を紹介したところで、blogeyeでどのようにこのHadoopクラスタを利用しているかを紹介したいと思います。
Hadoopを利用するための独自のラッパークラス「SimpleMapReducer」
blogeyeでは、Hadoopを利用したプログラムを簡単に記述できるように、独自のラッパークラスを用意しています。
Hadoopを利用するプログラムを書く上で最も手間がかかるのは、データの入出力にIntWritable、TextといったHadoop専用クラスを利用する必要があることです。
blogeyeでは多少の速度を犠牲にして、キーはすべてString、値はすべてシリアライズ可能なObject型で入出力することとし、出力時はObject型をラッパーでバイト列にシリアライズしてBytesWritable型としてHDFSに記録、入力時はSequenceFileからBytesWritable型のデータを読み出し、バイト列をObjectInputStream経由でObjectとして読み込んでいます。
これらの処理を行うSimpleMapReducerクラスの詳細については、記事添付のサンプルコードを参照してください。
ラッパークラスによるコーディングの簡略化
SimpleMapReducerクラスを用いることで、コードをすっきりさせることができます。例えば、時間ごとのブログ投稿記事の分布を調べるプログラムは、次のように、簡潔に書くことができます。
public class ArticleCounter extends SimpleMapReducer { public static class Mapper extends SimpleMapper { public void Map(String key, Object value) throws IOException { BlogArticle article = (BlogArticle) value; int d= article.timestamp.getHours(); Output(""+d, 1); } } public static class Reducer extends SimpleReducer { public void Reduce(String key, Iterator values) throws IOException { int count = 0; while(values.hasNext()){ count += (Integer)values.next(); } Output(count); } } public static void main(String[] argv) throws Exception { JobConf jobConf = Init(ArticleCounter.class, ArticleCounter.Mapper.class, ArticleCounter.Reducer.class); jobConf.setJobName("article_counter"); jobConf.setNumMapTasks(1000); jobConf.setNumReduceTasks(1); jobConf.setCombinerClass(ArticleCounter.Reducer.class); jobConf.setInputPath(new Path(argv[0])); jobConf.setOutputPath(new Path(argv[1])); jobConf.setSpeculativeExecution(true); RunJob(jobConf); } }
さて、準備が整ったところで、blogeyeで実際にHadoopをどのように利用しているのかを紹介したいと思います。
ブログのクロール処理
まずブログのクロールですが、まずpingサーバーに問い合わせを行い、更新されたブログのリストをダウンロードします。これを一旦クロール待ちURLキューに書き出します。このキューにはHDFSではなくMySQLを利用しています。せいぜい数万件のエントリを入れるテーブルであるのと、blogeyeのクローラは最大でも10台程度のマシンからしか読み出さないため、特殊なデータベースを利用しなくてもMySQLで十分です。
実際の記事のクロールは、Hadoopを利用して並列化したブログ記事クローラで行っています。このクローラは、まずマスタ上でMySQLのクロール待ちキューからURLのリストを取得してHDFSに書き出し、それをMapReduceの入力として指定します。クロールしたブログ記事はまたMySQL上のテーブルに書き出します。
HDFSとRDBMSの使い分け
HDFSはランダムアクセスや少量データの書き込みを苦手とするため、少しずつデータを追記したり、重複記事検出をする必要がある、直近にクロールしたブログデータを書き込んだりするのに向いていません。
一方、MySQLは大量データの保管や、数百台からの並列読み出しには向かないため、1日に1回、1日分のブログデータをまとめてSequenceFileの形式でHDFSに書き出し、MySQLからは削除しています。
ここでMySQLとHDFSを比較すると、下表のようになります。一般的に、頻繁に利用しない大量データはHDFSに、その他のデータはMySQLに置くのが良いと言えるでしょう。
| MySQL | HDFS（分散ファイルシステムを使用した場合、S3を使用した場合共に） | |
| ランダムアクセス | 速い | 遅い |
| レスポンスタイム | 早い | 遅い（特にS3を利用する場合は遅い） |
| データの保管場所 | マスタ | S3（スレーブにキャッシュ） |
| 読み書きのスケーラビリティ | 低い | 高い |
著者属性の推定処理
さて、最後にblogeyeで行っている処理の中で最もHadoopの恩恵を受けている、著者属性推定処理について紹介します。これは、過去にクロールしたすべてのブログ記事を使い、各ブログサイトの著者の「性別」「年齢」「居住都道府県」を推定するという処理です。
1TB近いブログの記事データを100台から並列に読み出すという操作も、HDFSからならば問題なく可能です。また、投稿日ごとに管理している2億のブログ記事をサイトごとにまとめ直すという操作はMapReduceが得意とする処理です。
具体的には、著者属性推定プログラムは下記のような構成になっています。
- Mapper
SPAMフィルタOutput（ブログサイトURL、記事データ）
- Reducer
各ブログ記事のテキストを推定処理推定結果のデータベースへの挿入
package trendsio.prop; import java.io.IOException; import java.util.HashMap; import java.util.Hashtable; import java.util.Iterator; import ohkura.lib.mapred.SimpleMapReducer; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.SequenceFile.CompressionType; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.JobPriority; import trendsio.comp.BlogArticle; import trendsio.comp.MasterDB; public class BlogPropertyEstimator extends SimpleMapReducer { public static class Mapper extends SimpleMapper { public void Map(String key, Object value) throws IOException { BlogArticle article = (BlogArticle) value; // spam filter if (isSPAM(article.description)) return; Output(article.feed, article); } } public static class Reducer extends SimpleReducer { public void Reduce(String key, Iterator values) throws IOException { Report("START: " + key); MasterDB db = MasterDB.getDB(); try { BlogPropertyManager prop_manager = BlogPropertyManager.getInstance(); Report("ESTIMATING: " + key); HashMap<String, double[]> props = prop_manager.estimate((Iterator<BlogArticle>) values); Hashtable<String, Object> dbvalues = new Hashtable<String, Object>(); dbvalues.put("feed", key); dbvalues.put("count", 0); dbvalues.put("props", MasterDB.toByteArray(props)); Report("TO INSERT: " + key); db.replace("blogprops", dbvalues); } catch (Exception e) { e.printStackTrace(); } } } public static void main(String[] argv) throws Exception { JobConf jobConf = Init(BlogPropertyEstimator.class, BlogPropertyEstimator.Mapper.class, BlogPropertyEstimator.Reducer.class); jobConf.setJobName("BlogPropEstimator"); jobConf.setNumMapTasks(100); jobConf.setNumReduceTasks(100); for (int i = 0; i < argv.length; i++) { if (i == 0) jobConf.setInputPath(new Path(argv[i])); else if (i == argv.length - 1) jobConf.setOutputPath(new Path(argv[i])); else jobConf.addInputPath(new Path(argv[i])); } jobConf.setJobPriority(JobPriority.VERY_LOW); jobConf.setSpeculativeExecution(true); jobConf.setMapOutputCompressionType(CompressionType.BLOCK); jobConf.setMaxMapTaskFailuresPercent(1); SubmitJob(jobConf); } }
通常、1TB近いデータを100台のマシンで、しかも相互にデータを交換しながら処理するとなれば、たくさんのコードを記述しなければならず、またデバッグも簡単ではありません。しかし、blogeyeではHadoopを用いることにより、これらの処理を極めて簡単に実現することができました。
まとめ
今回は、Hadoopを実際に大量データ処理に用いた例としてblogeyeでの利用法について紹介しました。また、AmazonのEC2／S3を利用する利点、Hadoop開発を容易にする方法なども紹介しました。
HadoopとEC2によって、これまで個人ではなかなか手が出せなかった大規模処理を手軽に実現できるようになっています。この記事をきっかけに、是非実際に試して、この便利さを実感してもらえれば幸いです。
次回はバトンを太田さんにお返しして、分散データベースBigTableのオープンソースクローン「hBase」のインストール方法や使い方を紹介していただく予定です。
