概要

今回はJobTrackerTaskTrackerの導入を行ないます。ClouderaのCDH3を利用。

本項を実行するにはNameNodeとDataNodeが稼動している必要があります。また、最低でも2台(できれば3台以上)のパソコンが必要になります。

NameNodeとDataNodeのインストールについては下記を参照してください

完全分散モードのHadoop NameNode導入(Ubuntu10.04、CDH3)

@Date 2011/11/7
@Versions Hadoop0.20, Ubuntu10.04, CDH3

今回やること

NameNode導入の項で、NameNode(HDFS親)とDataNode(HDFS子)が動いている状態ができました。

導入済みの構成
親 : NameNode
子 : DataNode

今回はNameNodeが入っているマシンでJobTracker(MapReduce親)を、DataNodeが入っているマシンでTaskTracker(MapReduce子)を動かします。

今回の構成
親 : NameNode + JobTracker
子 : DataNode + TaskTracker

通常、DataNodeとTaskTrackerは同じマシンに導入されます。

それぞれ違うマシンに導入することもできますが、そうするとMapReduceの処理を行った際に、処理対象のデータをDataNodeからTaskTrackerに向けて転送しなければいけなくなります。大規模データ処理でそれをやるとあっという間にネットワークが枯渇してしまいます。

NameNodeとJobTrackerについては同じマシンに入れる必要はありませんが、自宅で構築する際は2つを一緒に入れておいた方が台数が節約できて良いです。

NameNode導入の項で行なった作業によって、NameNodeと同じマシンにJobTrackerが、DataNodeと同じマシンにTaskTrackerがインストールされた状態になっているので、今回は設定ファイルを編集して複数台のマシンでMapReduceが実行できるような環境を構築します。

mapred-site.xmlの編集

MapReduce周りの設定はmapred-site.xmlに記述します。

主な設定項目は、JobTrackerのサーバ名とポートを指定するmapred.job.tracker、実行されるMapperやReducerを最大でいくつ立ち上げるかを指定するmapred.tasktracker.{map|reduce}.tasks.maximum、一時ファイルの出力先のmapred.local.dirなどです。

細かい設定は後回しにして、まずはmapred.job.trackerだけ指定します。この設定はすべてのJobTrackerとTaskTrackerが動作するマシンに対して行う必要があります。

<configuration>
  <property>
    <name>mapred.job.tracker</name>
    <value>tora:54311</value>
  </property>
</configuration>

toraは我が家の親機のマシン名です。サーバ名:ポート番号で指定します。

mapred.tasktracker.{map|reduce}.tasks.maximumはデフォルトでは2になります。この辺りの適正値はCPUのコア数とかによって変わってきます。

mapred.local.dirはデフォルトでは${hadoop.tmp.dir}/mapred/localになります。

ディレクトリの作成

JobTrackerやTaskTrackerはmapredユーザによって実行されます。そのためmapred.local.dirで指定されたディレクトリは、mapredユーザによって操作できるようにしておく必要があります。

我が家の設定ではhadoop.tmp.dirが/mnt/hdfs1/hadoop.tmpというディレクトリに指定されているので、mapred.local.dirは/mnt/hdfs1/hadoop.tmp/mapred/localになります。

$ sudo mkdir /mnt/hdfs1/hadoop.tmp/mapred
$ sudo chown mapred:mapred /mnt/hdfs1/hadoop.tmp/mapred

上記のようにmapredディレクトリを作ってそこの権限を与えておけば、あとは勝手にディレクトリ作って動いてくれます。

次にmapred.system.dirも作成します。こちらはHDFS上に作られるディレクトリで、デフォルトは${hadoo.tmp.dir}/mapred/systemになります。

$ sudo -u hdfs hadoop fs -mkdir /mnt/hdfs1/hadoop.tmp/mapred
$ sudo -u hdfs hadoop fs -chown mapred:mapred /mnt/hdfs1/hadoop.tmp/mapred

// ディレクトリができてることを確認
$ hadoop fs -ls /mnt/hdfs1/hadoop.tmp/
Found 1 items
drwxr-xr-x   - mapred mapred          0 2011-11-07 03:04 /mnt/hdfs1/hadoop.tmp/mapred

これで必要なディレクトリは揃いました。

JobTrackerの起動

設定が終わったら、親機でJobTrackerを立ち上げます。

$ sudo service hadoop-0.20-jobtracker start

立ちあげると、JobTrackerを立ち上げたサーバの50030ポートでJobTrackerの状態を確認できるようになります。

http://tora:50030/

まだTaskTrackerを立ち上げていないので、Nodesの項は0になっていると思います。

うまく立ち上がらない場合はログを見ます。

$ vi /usr/lib/hadoop-0.20/logs/hadoop-hadoop-jobtracker-*.log

TaskTracerの起動

次はTaskTrackerです。とりあえず子機の中から1台適当なマシンを選んで、startします。

$ sudo service hadoop-0.20-tasktracker start

正常に立ち上がると、50030ポートでそれまでは0件と表示されていたNodesの項が1になります。反映されるまでには数十秒かかる場合があります。

うまく立ち上がらない場合はやっぱりログを見ます。

$ vi /usr/lib/hadoop-0.20/logs/hadoop-hadoop-tasktracker-*.log

あとは残りのマシンも順次立ちあげていきます。立ち上げた数分だけNodesが認識されたら起動成功です。

MapReduceのサンプルを実行する

では、さっそくMapReduceで何か処理をしてみましょう。CDH3の場合は以下のパスにサンプル用のjarファイルが用意されています。

/usr/lib/hadoop-0.20/hadoop-examples.jar

hadoop jar jar_fileのような形式で実行することができます。実行する際のユーザは、HDFSに書き込み権限があるhdfsユーザを利用します。

$ sudo -u hdfs hadoop jar /usr/lib/hadoop-0.20/hadoop-examples.jar

上記のように実行すると、実行できる処理の種類が表示されます。代表的なところでは以下のような処理があります。

wordcount指定ファイルを読み込んで出現する単語の数を数える。英語向けなので日本語の単語数は数えられない。
pi円周率を計算する。並列処理はこの手の計算は得意ではないので、けっこういい加減な結果が返ってくる。
grep指定文字が存在する行を抽出する。
terasort大きなファイルをソートする
teragenterasortをする素になるサンプルファイルを生成する
sudoku数独の問題を解く

teragentとterasortあたりは性能評価なんかで使えます。

とりあえず何の準備もなく使える円周率あたりを触ってみましょう。

$ sudo -u hdfs hadoop jar /usr/lib/hadoop-0.20/hadoop-examples.jar pi 10 100000

Estimated value of Pi is 3.14155200000000000000

上記の命令は10000個のサンプルを10回計算しろという意味になります。見ての通り、結果は割と適当なものが返ってきます。

Reducer実行時にToo many fetch-failuresというエラーが出た場合は、hostsの指定が怪しいです。うちで出た時は、1台のSlaveが他の1台のSlaveをhostsに書き忘れていたことが原因でした。

Jobの途中停止

上のpiの計算でサンプル数を高く設定し過ぎると、いつまで待っても計算が終わらなくなったりします。途中で停止する場合は、hadoop job -killを使います。

// 止めるJobのIDが必要なのでlistを表示
$ hadoop job -list
1 jobs currently running
JobId	State	StartTime	UserName	Priority	SchedulingInfo
job_201111070304_0004	4	1320683401404	hdfs	NORMAL	NA

// 出てきたJobIdを使ってkill
$ hadoop job -kill job_201111070304_0004

JobIDはhttp://tora:50030/を見ても分かります。

Mapperの数の指定

Mapperの数はデフォルトだと1つのTaskTrackerにつき2つになっています。なので4台のTaskTrackerを動かした場合、動くMapperの数は8つになります。

4コアや6コアのマシンを使っている場合は、2のままだと一部のコアが遊んでしまいます。

試しに4コアのマシン1台を使って、Mapperの数を変えながらhadoop-examples.jarのpiを実行し、実行時間の変化を観察してみましょう。

実行されているMapperの数は処理を実行している最中にhttp://tora:50030/を見ると分かります。

Mapperの最大数はmapred.tasktracker.map.tasks.maximumで設定できます。

$ sudo vi /etc/hadoop-0.20/conf.my/mapred-site.xml

  <property>
    <name>mapred.tasktracker.map.tasks.maximum</name>
    <value>1</value>
  </property>

// 変更したらTaskTrackerを再起動
$ sudo service hadoop-0.20-tasktracker restart

上記のように1を設定すると、そのTaskTrackerで利用されるMapperの数が1になります。これを利用してMapperの数を変えながら実行時間を計測してみます。

// 実行したコマンド
$ sudo -u hdfs hadoop jar /usr/lib/hadoop-0.20/hadoop-examples.jar pi 100 100
Mapperの数所要時間Mapperの数所要時間
1621秒5195秒
2327秒6160秒
3248秒7153秒
4218秒8176秒

計測時間はmap処理が0%から100%になるまでの時間です。

4コアなのにMapperを6個並列で動かすまでは普通に性能が向上しています。適切なMapperの数は処理内容によっても変わるので、6は少し大きすぎる気もします。コア数と同じくらいの数(Reducerを動かす個数によってはさらに少なめ)を設定しておくのが普通かと思われます。

まとめ

他にもMapReduceに関する設定項目はいろいろあります。Reducerの数を調節したり、MapperやReducerが使用するメモリの量を指定したり、結果を圧縮することで出力にかかる時間を短くするなどなど。

適切な設定をしてあげると速度が平気で倍になったりすることがあるので、いろいろ設定をいじってみてどういった処理はどう設定すれば速度が出るのかいろいろと遊んでみたいところです。