Hadoopを使ってみよう [2]

CDHとは

Hadoopをインストールするためには多数のパッケージが必要なため、手順が大変複雑です。そこで、米Cloudera社が提供しているCDH(Cloudera's Distribution Including Apache Hadoop)を利用するとよいでしょう。

CDHはさまざまな形式で配布されています。コンポーネントごとの個別配布、各種クラウドプラットフォーム上にクラスタを構築するためのライブラリであるApache Whirrを用いた形式、さらにデモ用としてVMware、KVM、VirtualBox用のイメージとしても配布されています。

今回はKVM用のイメージを用いてCDHのデモを行います。

KVMによるCDHv3のデモ

まず、KVM用のイメージを取得します。およそ1GBのサイズがあり、ダウンロードには数時間かかります。ダウンロードが完了したら、md5sum の確認を行います。

% ls -l cloudera-demo-vm-cdh3u4-kvm.tar.gz 
-rw-r--r-- 1 cf cf 914523647  5月 17 09:23 cloudera-demo-vm-cdh3u4-kvm.tar.gz
% cat cloudera-demo-vm-cdh3u4-kvm.md5 
b182919964474b19e12838046d9b6d11  cloudera-demo-vm-cdh3u4-kvm.tar.gz
% md5sum -c cloudera-demo-vm-cdh3u4-kvm.md5 
cloudera-demo-vm-cdh3u4-kvm.tar.gz: 完了
%

ファイルを展開します。

% tar xfz cloudera-demo-vm-cdh3u4-kvm.tar.gz
% ls -la cloudera-demo-vm-cdh3u4-kvm
合計 5352572
drwxr-sr-x 2 cf cf       4096  5月 23 21:42 .
drwxr-sr-x 3 cf cf       4096  5月 23 21:42 ..
-rwxr-xr-x 1 cf cf 5475663872  5月 17 03:23 cloudera-demo-vm-sda.raw
%

CentOS 5.8 (x86_64) をベースとして、CDHv3のコンポーネントが一式含まれています。つまり、Hadoopのマスターサーバとスレーブサーバが1台のマシン上で動作する形式になっています。
このイメージに2GBのメモリを割り当てて仮想マシンを起動します。起動したら、ユーザ名 cloudera パスワード cloudera でログインすることができます。

Hadoop TutorialのWordCount.javaを例として使います。これは入力された英文テキスト中に出現する英単語の数を集計するプログラムです。

[cloudera@localhost ~]$ cat > WordCount.java
package org.myorg;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class WordCount {

public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
}
}
}

public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}

public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(WordCount.class);
conf.setJobName("wordcount");

conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);

conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));

JobClient.runJob(conf);
}
}
[cloudera@localhost ~]$

Hadoopの利用するための環境変数を設定します。

[cloudera@localhost ~]$ export HADOOP_HOME=/usr/lib/hadoop-0.20
[cloudera@localhost ~]$ export HADOOP_VERSION=0.20.2-cdh3u4
[cloudera@localhost ~]$

WordCount.javaをコンパイルします。

[cloudera@localhost ~]$ mkdir wordcount_classes
[cloudera@localhost ~]$ javac -classpath ${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-core.jar -d wordcount_classes WordCount.java
[cloudera@localhost ~]$

生成されたクラスファイルをアーカイブします。

[cloudera@localhost ~]$ jar -cvf /home/cloudera/wordcount.jar -C wordcount_classes/ .
マニフェストが追加されました。
org/ を追加中です。(入 = 0) (出 = 0)(0% 格納されました)
org/myorg/ を追加中です。(入 = 0) (出 = 0)(0% 格納されました)
org/myorg/WordCount$Reduce.class を追加中です。(入 = 1611) (出 = 649)(59% 収縮されました)
org/myorg/WordCount.class を追加中です。(入 = 1546) (出 = 749)(51% 収縮されました)
org/myorg/WordCount$Map.class を追加中です。(入 = 1938) (出 = 798)(58% 収縮されました)
[cloudera@localhost ~]$

hadoopコマンドを用いて、HDFSのルートディレクトリを表示します。

[cloudera@localhost ~]$ hadoop dfs -ls /
Found 3 items
drwxrwxrwx - hue supergroup 0 2012-05-23 12:52 /tmp
drwxr-xr-x - hue supergroup 0 2012-05-23 12:52 /user
drwxr-xr-x - mapred supergroup 0 2012-05-23 12:52 /var
[cloudera@localhost ~]$

なお、ここではhadoopコマンドにdfsという値を与えていますが、fsで解説される場合もあります。hadoopコマンド内を確認すると、どちらも同じクラスを呼び出しており、違いはないようです。

[cloudera@localhost ~]$ less /usr/lib/hadoop-0.20/bin/hadoop
:
elif [ "$COMMAND" = "fs" ] ; then
CLASS=org.apache.hadoop.fs.FsShell
HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
elif [ "$COMMAND" = "dfs" ] ; then
CLASS=org.apache.hadoop.fs.FsShell
HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
:

引き続き、HDFS上にディレクトリを作成します。

[cloudera@localhost ~]$ hadoop dfs -mkdir /usr/joe/wordcount/input
[cloudera@localhost ~]$

[cloudera@localhost ~]$ hadoop dfs -ls /usr/joe/wordcount
Found 1 items
drwxr-xr-x - cloudera supergroup 0 2012-05-23 13:34 /usr/joe/wordcount/input
[cloudera@localhost ~]$

ローカルに、Hadoopに入力するためのファイルを作成します。

[cloudera@localhost ~]$ cat > file01
Hello World Bye World
[cloudera@localhost ~]$ cat > file02
Hello Hadoop Goodbye Hadoop
[cloudera@localhost ~]$

ローカルに作成したファイルをHDFS上に転送します。

[cloudera@localhost ~]$ hadoop dfs -put file01 file02 /usr/joe/wordcount/input
[cloudera@localhost ~]$

正常に転送できたことを確認します。

[cloudera@localhost ~]$ hadoop dfs -ls /usr/joe/wordcount/input
Found 2 items
-rw-r--r-- 1 cloudera supergroup 22 2012-05-23 13:34 /usr/joe/wordcount/input/file01
-rw-r--r-- 1 cloudera supergroup 28 2012-05-23 13:34 /usr/joe/wordcount/input/file02
[cloudera@localhost ~]$

[cloudera@localhost ~]$ hadoop dfs -cat /usr/joe/wordcount/input/file01
Hello World Bye World
[cloudera@localhost ~]$ hadoop dfs -cat /usr/joe/wordcount/input/file02
Hello Hadoop Goodbye Hadoop
[cloudera@localhost ~]$

wordcountアプリケーションを実行します。

[cloudera@localhost ~]$ hadoop jar /home/cloudera/wordcount.jar org.myorg.WordCount /usr/joe/wordcount/input /usr/joe/wordcount/output
12/05/23 13:39:57 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
12/05/23 13:39:58 WARN snappy.LoadSnappy: Snappy native library is available
12/05/23 13:39:58 INFO util.NativeCodeLoader: Loaded the native-hadoop library
12/05/23 13:39:58 INFO snappy.LoadSnappy: Snappy native library loaded
12/05/23 13:39:58 INFO mapred.FileInputFormat: Total input paths to process : 2
12/05/23 13:39:58 INFO mapred.JobClient: Running job: job_201205231252_0003
12/05/23 13:39:59 INFO mapred.JobClient: map 0% reduce 0%
12/05/23 13:40:06 INFO mapred.JobClient: map 66% reduce 0%
12/05/23 13:40:09 INFO mapred.JobClient: map 100% reduce 0%
12/05/23 13:40:16 INFO mapred.JobClient: map 100% reduce 100%
12/05/23 13:40:17 INFO mapred.JobClient: Job complete: job_201205231252_0003
12/05/23 13:40:17 INFO mapred.JobClient: Counters: 27
12/05/23 13:40:17 INFO mapred.JobClient: Job Counters
12/05/23 13:40:17 INFO mapred.JobClient: Launched reduce tasks=1
12/05/23 13:40:17 INFO mapred.JobClient: SLOTS_MILLIS_MAPS=13845
12/05/23 13:40:17 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0
12/05/23 13:40:17 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0
12/05/23 13:40:17 INFO mapred.JobClient: Launched map tasks=3
12/05/23 13:40:17 INFO mapred.JobClient: Data-local map tasks=3
12/05/23 13:40:17 INFO mapred.JobClient: SLOTS_MILLIS_REDUCES=10048
12/05/23 13:40:17 INFO mapred.JobClient: FileSystemCounters
12/05/23 13:40:17 INFO mapred.JobClient: FILE_BYTES_READ=79
12/05/23 13:40:17 INFO mapred.JobClient: HDFS_BYTES_READ=348
12/05/23 13:40:17 INFO mapred.JobClient: FILE_BYTES_WRITTEN=227780
12/05/23 13:40:17 INFO mapred.JobClient: HDFS_BYTES_WRITTEN=41
12/05/23 13:40:17 INFO mapred.JobClient: Map-Reduce Framework
12/05/23 13:40:17 INFO mapred.JobClient: Map input records=2
12/05/23 13:40:17 INFO mapred.JobClient: Reduce shuffle bytes=91
12/05/23 13:40:17 INFO mapred.JobClient: Spilled Records=12
12/05/23 13:40:17 INFO mapred.JobClient: Map output bytes=82
12/05/23 13:40:17 INFO mapred.JobClient: CPU time spent (ms)=1450
12/05/23 13:40:17 INFO mapred.JobClient: Total committed heap usage (bytes)=426913792
12/05/23 13:40:17 INFO mapred.JobClient: Map input bytes=50
12/05/23 13:40:17 INFO mapred.JobClient: Combine input records=8
12/05/23 13:40:17 INFO mapred.JobClient: SPLIT_RAW_BYTES=294
12/05/23 13:40:17 INFO mapred.JobClient: Reduce input records=6
12/05/23 13:40:17 INFO mapred.JobClient: Reduce input groups=5
12/05/23 13:40:17 INFO mapred.JobClient: Combine output records=6
12/05/23 13:40:17 INFO mapred.JobClient: Physical memory (bytes) snapshot=561541120
12/05/23 13:40:17 INFO mapred.JobClient: Reduce output records=5
12/05/23 13:40:17 INFO mapred.JobClient: Virtual memory (bytes) snapshot=2018074624
12/05/23 13:40:17 INFO mapred.JobClient: Map output records=8
[cloudera@localhost ~]$

実行結果を確認します。

[cloudera@localhost ~]$ hadoop dfs -ls /usr/joe/wordcount/output
Found 3 items
-rw-r--r-- 1 cloudera supergroup 0 2012-05-23 13:40 /usr/joe/wordcount/output/_SUCCESS
drwxr-xr-x - cloudera supergroup 0 2012-05-23 13:39 /usr/joe/wordcount/output/_logs
-rw-r--r-- 1 cloudera supergroup 41 2012-05-23 13:40 /usr/joe/wordcount/output/part-00000
[cloudera@localhost ~]$

正常に英単語が集計されています。

[cloudera@localhost ~]$ hadoop dfs -cat /usr/joe/wordcount/output/part-00000
Bye 1
Goodbye 1
Hadoop 2
Hello 2
World 2
[cloudera@localhost ~]$