第一個Hadoop程序

Hadoop資料之前倒是有看過，只是工作一直沒涉及，自己也沒有動手過。之前嘗試高版本的apache版本機編譯，居然都不成功。初衷只是要搭建環境，卻由於都放在業余時間搞，時間不連續，嘗試的方法不同(中間嘗試安裝cdh4.40,結果發現面對一堆文檔還不是自己想要的，最後還是回到hadoop權威指南附錄)，導致這個工作一直難產。雖說最後結果簡單，還是記錄一下。

下載的是apache穩定版hadoop-0.23.9,0.23之後apache版本往兩個分支走:hadoop 1.0和hadoop 2.0。由於這個版本較新卻和0.20接近，配置上相似一些，有了之前嘗試的教訓，就是這個了。解壓之後要設定好path，path加上hadoop bin目錄。默認配置文件在hadoop etc/hadoop下。

公共配置:core-site.xml

hdfs配置:hdfs-site.xml

mapreduce配置: mapred-site.xml

有三種模式,standalone模式，偽分布式,和分布式。什麼都不改啟動的是本機模式。這裡就是跑在本機模式。

WordCount2.java:

package com.imd.examples;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;

public class WordCount2 {
public static void main(String[] args) throws Exception {
// 不配置一個入口類會導致底層無法調用setJarByClass方法,運行時ClassNotFound.
JobConf conf = new JobConf(WordCount2.class);
conf.setMapperClass(WordCountMapper.class);
conf.setReducerClass(WordCountReducer.class);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);

// 輸入參數除最後一個外都是input path,最後一個是output path
for (int i = 0; i < args.length - 1; ++i) {
FileInputFormat.addInputPath(conf, new Path(args[i]));
}

FileOutputFormat.setOutputPath(conf, new Path(args[args.length - 1]));
JobClient.runJob(conf);
}
}

WordCountMapper.java:

package com.imd.examples;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

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

@Override
// 怎麼切分的呢?key是什麼?
public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
StringTokenizer st = new StringTokenizer(value.toString());
Text text = new Text();
while (st.hasMoreTokens()) {
text.set(st.nextToken());
output.collect(text, one);
}
}
}