FANCOMI Ad-Tech Blog

株式会社ファンコミュニケーションズ nend・新規事業のエンジニア・技術ブログ

とあるオープンソースの分散処理基盤 -Hadoop（ハドゥープ）-

Hadoop nend k_oomori

Introduction - Hadoopとは?

　こんにちは、k_oomoriです。今回は、オープンソースミドルウェアのApache Hadoopを取り上げたいと思います。

　コンピュータの性能は日々進化しているとはいえ、1台のマシンの性能にはおのずと限界があります。また、その時々で常に最高性能のマシンを維持していこうとすると莫大なコストがかかってしまいます。そこで登場するのが分散処理という考え方です。1台1台は特に高性能ではないマシンでも多数並べて処理を同時並行で行うことにより、全体の処理性能を上げようというアプローチです。この方法では性能が足りなくなった場合にはマシンを追加することで容易に性能向上ができる反面、分散処理の実装そのものが難しいという問題がありました。例えば、処理の振り分けアルゴリズムやサーバの死活監視、結果の集約など、考えなければならないことがいくつもあります。この分散処理にまつわる厄介ごとを一手に引き受けてくれるのがHadoopです。一度Hadoopクラスタを構成し、MapReduce※1という作法に則ってプログラミングをすれば、プログラマはそれ以上分散処理について意識する必要はありません。自分が本来やりたい処理に集中することができるのです。

　またHadoopによる分散の利点は計算の処理速度向上だけではありません。HadoopはHDFS（Hadoop Distributed File System）というHadoop専用の分散ファイルシステムを使用します。これによって得られる恩恵としては

  • 複数のディスクから同時に読み書きすることができるため、ディスクI/Oのスループットを向上させることができます。
  • ファイルシステム全体の記憶容量は各スレーブサーバのディスク容量の和になるため、サーバを追加しさえすればどれだけ大容量のデータであっても格納することが可能です。
  • データのレプリケーションにより信頼性（データ損失に対する耐性）を確保できます。

　こう見てくるといいことずくめのようですが、もちろんなんでもかんでも分散して高速化ができるわけではありません。では、どういった処理がHadoopに向いているのでしょうか。

Hadoopの使い所

　まず、そもそも分散処理できない（してはいけない）例として、個々の入力データの間の関連性が無視できない場合が挙げられます。例えば、ログの出力順序が重要で、必ず時系列に処理しなければならない場合には、１箇所で集中管理する必要があります。
　またHadoopは分散処理のためのオーバーヘッドが大きいため、たとえごくわずかなデータを処理する場合であっても10～20秒くらいかかってしまいます。従ってWebサービスのような低レイテンシが求められる処理には向きません。※2
　またデータ量がそれほど多くなく、RDBMSでも充分に扱えてしまうような場合、Hadoopでもやってできないことはないのですが、素直にRDBMSや既存のプロダクトを使っておいた方が柔軟性もあり処理も容易である場合が多いと思います。
　以上のことから、Hadoopに向いている典型的な処理としては
　　大量データを扱うバッチ処理で、入力データを独立に処理してよいもの
ということになるでしょう。

　では、アドネットワークサービスnendにおけるHadoop向きな処理とは何でしょうか。
　まず真っ先に思いつくのはインプレッション（広告露出）ログの集計処理です。nendでは（そしておそらくはほとんどどの広告配信サービスでも）特定の時間間隔ごとにバッチ処理でインプレッションの集計を行っています。
　またインプレッションログはどのレコード間にも関連性はありません。集計単位に従って独立に数え上げるだけです。
　1度に扱うデータ量としては、2013年1月現在で約2～3千万レコード、容量にして数GB程度となっています。これはまだHadoopが本気を出すレベルではないのですが、既存のシングルプロセスによる処理では結構時間がかかるようになってきており、またこれまでのインプレッション数の推移（図1）を見るに今後さらなる増加が見込まれます。つまり、大容量データとして対処すべき段階に入りつつあると言えるでしょう。
　以上のことから、インプレッションログの集計はHadoopに向いた処理の条件を満たしていると言えます。このような経緯で、Hadoopの動作検証を行うことになりました。導入後は、データ解析等いろいろな処理にHadoopを活用していきたいと考えています。

f:id:fan_k_oomori:20160502121023p:plain 図1. nendインプレッションの推移

Hadoopのインストール、設定

　それではHadoopのインストールに進みましょう。ここでは管理のしやすさや周辺プロダクトとの相性を考慮し、コミュニティーバージョンではなくHadoopディストリビューションの一つ、CDH（Cloudera's Distribution including Apache Hadoop）を採用することにします。2013年3月現在、利用可能なバージョンはCDH3 Update 5 (CDH3u5)とCDH Version 4.2.0 (CDH4.2.0)になりますが、CDH4に含まれるHadoopのバージョンが2.0.0(α版)であるなどかなり先進的な作りになっていることから、今回は安定性を重視しCDH3u5を選択します。また、今回使用するマシンのOSはCentOS 6.3です。
　まずは推奨に従ってJDK 1.6.0_26をインストールします。

# /usr/java/jdk-6u26-linux-x64-rpm.bin
# java -version
java version "1.6.0_26"
Java(TM) SE Runtime Environment (build 1.6.0_26-b03)
Java HotSpot(TM) 64-Bit Server VM (build 20.1-b02, mixed mode)

　yumを用いてCDH3をインストールするため、yumリポジトリーにClouderaのサイトを追加し、yum自身をアップデートします。

# cd /etc/yum.repos.d/
# wget http://archive.cloudera.com/redhat/cdh/cloudera-cdh3.repo
# yum update yum

　いよいよHadoopのインストールです。gzip圧縮を使う際にはネイティブライブラリ（hadoop-0.20-native）が必要となるのでこれもあらかじめ入れておきます。

# yum install hadoop-0.20 hadoop-0.20-native

　この段階でHadoop用のユーザ・グループが追加されているはずなので念のため確認しておきます。

# tail -2 /etc/passwd
mapred:x:101:105:Hadoop MapReduce:/usr/lib/hadoop-0.20:/bin/bash
hdfs:x:102:104:Hadoop HDFS:/usr/lib/hadoop-0.20:/bin/bash

# tail -3 /etc/group
hadoop:x:103:mapred,hdfs
hdfs:x:104:
mapred:x:105:

　擬似分散環境※3のインストールを行います。

# yum install hadoop-0.20-conf-pseudo

　設定ファイル群 /etc/hadoop-0.20/conf.pseudo/ とともに、パッケージ間の依存関係のため以下のデーモンプログラムも一緒にインストールされます。
・hadoop-0.20-namenode
・hadoop-0.20-secondarynamenode
・hadoop-0.20-datanode
・hadoop-0.20-jobtracker
・hadoop-0.20-tasktracker

　デフォルトではランレベル3, 5でデーモンが自動起動するようになっていますが、どうせ設定を書き直したりファイルシステムをフォーマットしたりしないといけないので、私は自動起動しないようにしています。（お好みでどうぞ）

# chkconfig hadoop-0.20-namenode off
# chkconfig hadoop-0.20-datanode off
# chkconfig hadoop-0.20-jobtracker off
# chkconfig hadoop-0.20-tasktracker off
# chkconfig hadoop-0.20-secondarynamenode off

　初めてHadoopを入れた場合はこの段階で擬似分散環境のテストを行うことになると思いますが、本記事では一気に完全分散環境の設定に進みます。完全分散モードでは複数台のサーバ（マスター×1、スレーブ×任意(≧1)）を用いるので、全サーバにHadoopをインストールしておきます。
　完全分散環境用設定ファイルの雛形として、擬似分散環境のものを用います。

# cd /etc/hadoop-0.20/
# cp -r conf.pseudo conf.fully
# cd conf.fully/

　必要な設定をしていきます。

# vi core-site.xml
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
  <property>
    <name>fs.default.name</name>
    <value>hdfs://namenode.hadoop.nend.net:8020</value>
  </property>
  <property>
     <name>hadoop.tmp.dir</name>
     <value>/tmp/hadoop</value>
  </property>
</configuration>

　fs.default.nameにNameNode※4のURIを、hadoop.tmp.dirに一時ファイル領域の設定をします。ディレクトリの作成は後ほど行います。
　ここで一つ注意しなくてはいけないのは、Hadoopでサーバを指定する場合はホスト名で指定しなくてはいけないということです。IPアドレスで指定しても、そのままでは通信を行うことができません。どうしてもIPアドレスを使いたい場合は、hdfs-site.xml, mapred-site.xmlに次のような設定を行う必要があります。

<property>
   <name>slave.host.name</name>
   <value>192.168.xxx.xxx</value>
 </property>

# vi hdfs-site.xml
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
  <property>
     <name>dfs.permissions</name>
     <value>false</value>
  </property>
  <property>
    <name>dfs.safemode.extension</name>
    <value>0</value>
  </property>
  <property>
     <name>dfs.safemode.min.datanodes</name>
     <value>1</value>
  </property>
  <property>
     <name>dfs.name.dir</name>
     <value>${hadoop.tmp.dir}/dfs/name</value>
  </property>
  <property>
     <name>dfs.data.dir</name>
     <value>${hadoop.tmp.dir}/dfs/data</value>
  </property>
  <property>
     <name>dfs.block.size</name>
     <value>67108864</value>
  </property>
</configuration>

　dfs.replicationにはHDFS上のデータの多重度を指定します。つまり、2以上を指定するとその数だけデータを重複して保持するので、DataNodeに障害があった場合でもデータが失われる可能性を低くすることができます。しかしもちろんその分データ量は多くなり、また書き込みにも時間がかかるようになるので、性能と可用性のトレードオフになります。今回の使用目的では、解析期間中のみデータを保持できればそれでよいので、ここでは1（つまりバックアップを用意しない）と指定しています。
　dfs.name.dir, dfs.data.dirにはそれぞれNameNode, DataNodeで使用するファイルを格納するディレクトリを指定します。
　(HDFSに限らず)ファイルシステムでは、ファイルを書き込む際ブロックという単位に分割してHDDに保存しています。dfs.block.sizeには、HDFSのブロックサイズを指定します。WindowsやLinuxなどで使われている通常のファイルシステムではブロックサイズは4kBといった比較的小さな値ですが、HDFSではディスクのシーク時間をなるべく短くするため、デフォルトで67,108,864B=64MBという大きな値が使われます。後ほど、この値がパフォーマンスにどのような影響を及ぼすか見ていきたいと思います。

# vi mapred-site.xml
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
  <property>
    <name>mapred.job.tracker</name>
    <value>jobtracker.hadoop.nend.net:8021</value>
  </property>
  <property>
    <name>mapred.local.dir</name>
    <value>${hadoop.tmp.dir}/mapred/local</value>
  </property>
  <property>
    <name>mapred.system.dir</name>
    <value>${hadoop.tmp.dir}/mapred/system</value>
  </property>
  <property>
    <name>mapred.compress.map.output</name>
    <value>false</value>
  </property>
  <property>
    <name>mapred.reduce.tasks</name>
    <value>8</value>
  </property>
  <property>
    <name>mapred.tasktracker.map.tasks.maximum</name>
    <value>4</value>
  </property>
  <property>
    <name>mapred.tasktracker.reduce.tasks.maximum</name>
    <value>4</value>
  </property>
</configuration>

　mapred.job.trackerにはJobTrackerのアドレスを指定します。
　mapred.compress.map.outputは、Map処理の出力ファイル（つまり、MapReduce処理における中間生成ファイル）を圧縮するかどうかを指定します。これをtrueにした場合、圧縮形式をmapred.map.output.compression.codecで指定することになります。
　mapred.reduce.tasks, mapred.tasktracker.map.tasks.maximum, mapred.tasktracker.reduce.tasks.maximumはそれぞれReduce処理の全タスク数、各TaskTracker上でのMap/Reduce処理のタスク数の上限を表します。これらはTaskTrackerの台数やコア数などに応じてチューニングされるべき値です。
　なお、mapred.map.tasksという設定項目もあるのですが、これは単なるヒントにすぎず、実際のMap数は 処理対象ファイルサイズ÷ブロックサイズ で決まるようです。

# vi slaves
slave1.hadoop.nend.net
slave2.hadoop.nend.net
...

　slavesにはスレーブサーバのホスト名(FQDN)を1行に1つずつ記述します。

　以上の設定が済んだら、完全分散環境の設定を有効にします。

# alternatives --install /etc/hadoop-0.20/conf hadoop-0.20-conf /etc/hadoop-0.20/conf.fully 25
# alternatives --set hadoop-0.20-conf /etc/hadoop-0.20/conf.fully

　alternatives --installコマンドでhadoop-0.20-confという名前のalternativesに/etc/hadoop-0.20/conf.fullyを優先度25で追加します。なお、hadoop-0.20-confには/etc/hadoop-0.20/conf.emptyと/etc/hadoop-0.20/conf.pseudoがそれぞれ優先度10, 30ですでに登録されています。ステータスが自動の場合は、優先度が最も大きな設定が有効になるのですが、Hadoopの動作環境を自動で変えるメリットがあまり思いつかないので、ここではalternatives --setコマンドにより手動で完全分散モード（/etc/hadoop-0.20/conf.fully設定）に向けています。
　alternatives --displayコマンドで、conf.fullyが有効になっていることを確認しましょう。

# alternatives --display hadoop-0.20-conf
hadoop-0.20-conf - status is manual.
 link currently points to /etc/hadoop-0.20/conf.fully
/etc/hadoop-0.20/conf.empty - priority 10
/etc/hadoop-0.20/conf.pseudo - priority 30
/etc/hadoop-0.20/conf.fully - priority 25
Current `best' version is /etc/hadoop-0.20/conf.pseudo.

　なお、この操作の結果、具体的には /etc/hadoop-0.20/conf -> /etc/alternatives/hadoop-0.20-conf -> /etc/hadoop-0.20/conf.fully というシンボリックリンクができています。

　そろそろ起動・・・といきたいところですが、まだ準備が残っています。がんばりましょう。。。
　まずはディレクトリの作成とパーミッションの設定です。先ほどの設定内容に合わせて

# mkdir /tmp/hadoop
# chown hdfs:hadoop /tmp/hadoop
# chmod 777 /tmp/hadoop

# mkdir -p /tmp/hadoop/dfs/name 
# mkdir -p /tmp/hadoop/dfs/data
# chown -R hdfs:hadoop /tmp/hadoop/dfs
# chmod -R 700 /tmp/hadoop/dfs/name/
# chmod -R 755 /tmp/hadoop/dfs/data/

# mkdir -p /tmp/hadoop/mapred/local
# chown -R mapred:hadoop /tmp/hadoop/mapred/

　サーバをホスト名で指定するようにしているので、/etc/hostsファイルに記載して名前解決できるようにしておきます。

# vi /etc/hosts
192.168.xxx.xxx slave1.hadoop.nend.net
...

　マスター⇔スレーブ間はもちろん、Shuffle処理中にはスレーブ⇔スレーブ間でも通信が発生するので、SSH公開鍵を設置してパスワードなしでログインできるようにしておきます。ついでにknown_hostsファイルも作成しておきます。

# cd /usr/lib/hadoop-0.20/
# mkdir .ssh
# chmod 02775 .ssh/
# chgrp hadoop .ssh/
# ssh-keygen -t rsa -P '' -f /usr/lib/hadoop-0.20/.ssh/id_rsa
# chmod g+rw .ssh/id_rsa*
# touch .ssh/known_hosts
# chmod 664 .ssh/known_hosts
# chmod 0755 .ssh/

　作成した公開鍵(id_rsa.pub)の内容を各サーバの.ssh/authorized_keysに登録し、hdfs, mapredユーザでパスワード入力せずにsshログインできることを確認します。

　最後に、hdfsユーザでNameNodeのフォーマットを行います。

[root@namenode.hadoop.nend.net]# su - hdfs
[hdfs@namenode.hadoop.nend.net]$ hadoop namenode -format

　さて、ついに準備が整いました！いよいよHadoopのノードを起動しましょう。まずはHDFSです。マスターノード上で次のコマンドを実行します。

[hdfs@namenode.hadoop.nend.net]$ /usr/lib/hadoop-0.20/bin/start-dfs.sh

　これによりマスターノード上でNameNodeが、スレーブノード上でDataNodeが起動します。プロセスを確認しましょう。

[hdfs@namenode.hadoop.nend.net]$ jps
21486 SecondaryNameNode
10884 Jps
21295 NameNode
[hdfs@slave1.hadoop.nend.net]$ jps
26981 DataNode
29747 Jps

　動作確認のため、書き込み・読み出し・削除をしてみます。

[hdfs@namenode.hadoop.nend.net]$ echo 'hoge' | hadoop fs -put - /user/hdfs/test.txt
[hdfs@namenode.hadoop.nend.net]$ hadoop fs -ls /user/hdfs/test.txt
Found 1 items
-rw-r--r--   1 hdfs supergroup          5 2013-03-08 20:25 /user/hdfs/test.txt
[hdfs@namenode.hadoop.nend.net]$ hadoop fs -cat /user/hdfs/test.txt
hoge
[hdfs@namenode.hadoop.nend.net]$ hadoop fs -rm -skipTrash /user/hdfs/test.txt
Deleted hdfs://namenode.hadoop.nend.net:8020/user/hdfs/test.txt
[hdfs@namenode.hadoop.nend.net]$ hadoop fs -lsr /user/
drwxr-xr-x   - hdfs supergroup          0 2013-03-08 20:26 /user/hdfs

　次にMapReduceです。mapredユーザにスイッチし、マスターノード上で起動コマンドを実行します。

[root@jobtracker.hadoop.nend.net]# su - mapred
[mapred@jobtracker.hadoop.nend.net]$ /usr/lib/hadoop-0.20/bin/start-mapred.sh

　マスターノード上でJobTrackerが、スレーブノード上でTaskTrackerが起動しています。

[mapred@jobtracker.hadoop.nend.net]$ jps
24688 JobTracker
11056 Jps
[mapred@slave1.hadoop.nend.net]$ jps
29786 Jps
4633 TaskTracker

　動作確認のため、円周率計算のサンプルプログラムを実行してみましょう。

[mapred@jobtracker.hadoop.nend.net]$ hadoop jar /usr/lib/hadoop-0.20/hadoop-examples.jar pi 4 10
～略～
Estimated value of Pi is 3.40000000000000000000

　Hadoopは以下のURLでWeb管理画面を提供しています。
HDFS: http://namenode.hadoop.nend.net:50070/
MapReduce: http://jobtracker.hadoop.nend.net:50030/
これによってMapReduce処理の進行状況などを確認することができます。

　これでようやく、Hadoopを完全分散モードで動かすことができるようになりました。

Summary

　In this article I introduced Hadoop as an open-source software framework for distributed data processing. I explained what kind of process was suited for Hadoop and that Hadoop would be useful for counting impression logs in nend.
　I then described how to install and configure Hadoop to CentOS 6.3 using CDH3. Hadoop daemon processes were started up in fully distributed mode.

Appendix: Some Tips

　スレーブサーバの数を増やしたり、設定値を変えたりしてHadoopのノードを立ち上げなおすと、うまくプロセスが立ち上がらなかったり動作しなかったりする場合があります。いくつかのケースに対し、対処法を示しておきます。

1. namenodeが起動しない
/usr/lib/hadoop-0.20/bin/start-dfs.shを実行しても、NameNodeデーモンプロセスが立ち上がらない場合があります。（標準エラー出力にもログファイルにも特に何も出ません…）この場合、namenodeをフォーマットし直すとうまくいくかもしれません。

[hdfs@namenode.hadoop.nend.net]$ /usr/lib/hadoop-0.20/bin/stop-dfs.sh
[hdfs@namenode.hadoop.nend.net]$ hadoop namenode -format
[hdfs@namenode.hadoop.nend.net]$ /usr/lib/hadoop-0.20/bin/start-dfs.sh

2. Incompatible namespaceIDsでdatanodeが起動しない
前ケースとは逆に、DataNodeが立ち上がっていない場合もあります。ログファイル（デフォルトでは/usr/lib/hadoop-0.20/logs/hadoop-hdfs-datanode-{NameNodeのホスト名}.log）に以下のようなIncompatible namespaceIDsというエラーメッセージが出力されている場合は、

2013-01-11 16:24:51,650 ERROR org.apache.hadoop.hdfs.server.datanode.DataNode: java.io.IOException: Incompatible namespaceIDs in /var/lib/hadoop-0.20/cache/hdfs/dfs/data: namenode namespaceID = 1519183789; datanode namespaceID = 1780715549

DataNodeの/tmp/hadoop/dfs/data/current/VERSION中のnamespaceIDをNameNodeの/tmp/hadoop/dfs/name/current/VERSIONのnamespaceIDに合わせてやるか、またはデータを全削除してフォーマットし直します。

[root@namenode.hadoop.nend.net]# rm -rf /tmp/hadoop/dfs/name/*
[root@datanode.hadoop.nend.net]# rm -rf /tmp/hadoop/dfs/data/*
[root@namenode.hadoop.nend.net]# su - hdfs
[hdfs@namenode.hadoop.nend.net]$ hadoop namenode -format

3. HDFSに書き込もうとするとSafeModeExceptionが発生する

# echo hoge | hadoop fs -put - hoge
put: org.apache.hadoop.hdfs.server.namenode.SafeModeException: Cannot create file/user/root/hoge. Name node is in safe mode.

起動直後、HDFSはセーフモード（データが規定量までレプリケートされるのを待っている状態）に入っていて書き込みを行うことができません。

# hadoop dfsadmin -safemode get
Safe mode is ON

この場合、次のコマンドでセーフモードを強制的に解除することができます。

# hadoop dfsadmin -safemode leave
Safe mode is OFF

4. Mapステージが完了した段階で処理がストップする
本文中でも書きましたが、Shuffle処理時にスレーブ間での通信が発生します。サーバ間でmapredユーザがパスワードなしでログインできること、ノードはホスト名で指定しているか、/etc/hostsに記載があるか等を確認しましょう。

5. ジョブを殺したいとき
一度ジョブを流してしまうと、hadoopコマンドをCtrl+Cで終わらせても処理は止まりません。これを削除するためには、まず現在実行中のジョブを一覧表示します。

$ hadoop job -list
1 jobs currently running
JobId   State   StartTime       UserName        Priority        SchedulingInfo
job_201303151735_0002   1       1363336652206   root    NORMAL  NA

殺したいジョブが特定できたら、そのJobIdを引数に与えてhadoop job -killコマンドを実行します。

$ hadoop job -kill job_201303151735_0002

ジョブの特定はWeb管理画面から行っても構いません。

Reference（参考にさせていただいた本、サイト）

※1. ^ ここではMapReduce処理（Shuffleも含む）の詳細については割愛させていただきます。こちらなどをご参照ください。

※2. ^ Hadoop本体の場合の話です。周辺プロダクトも含めて考えるとHBaseやImpalaなどリアルタイム処理可能なものが存在します。

※3. ^ 分散ソフトウェアとはいえ、開発時には1台でお手軽にできるよう、Hadoopには以下の3つの動作モードが用意されています。

動作モードサーバ台数ファイルシステム用途
ローカル（またはスタンドアロン）モード1台通常のローカルFS (ext3など)MapReduce動作確認
擬似分散(pseudo distributed)モード1台HDFSHDFS上でのMapReduce動作確認
完全分散(fully distributed)モード複数台HDFS実運用

※4. ^ Hadoopのサーバは役割によって次のような名前がついています。
NameNode：HDFSにとってのマスター。どのデータがどのDataNode上にあるかなどメタデータを管理します。
DataNode：HDFSにとってのスレーブ。実際のデータが配置されます。
JobTracker：MapReduceにとってのマスター。クライアントから投入されたジョブをいくつかのタスクに分割し、TaskTrackerに割り振ります。
TaskTracker：MapReduceにとってのスレーブ。割り当てに従って実際のMapReduce処理を行います。

f:id:fan_k_oomori:20160502121228p:plain この画像はこちらで作らせていただきました。（文字数制限が･･･orz）