Hadoop（二）：MapReduce程序（Java）

Java版本程序开发过程主要包含三个步骤，一是map、reduce程序开发；第二是将程序编译成JAR包；第三使用Hadoop jar命令进行任务提交。

下面拿一个具体的例子进行说明，一个简单的词频统计，输入数据是一个单词文本，输出每个单词的出现个数。

一、MapReduce程序

　　标准的MapReduce程序包含一个Mapper函数、一个Reducer函数和一个main函数

　　

1、主程序

package hadoop;

import org.apache.hadoop.conf.Configuration;　　// 读写和保存各种配置资源
import org.apache.hadoop.fs.Path;　　// 保存文件或者目录的路径
import org.apache.hadoop.io.IntWritable;　　// hadoop自身定义的整形类
import org.apache.hadoop.io.Text;　　// hadoop自身定义的存储字符串的类
import org.apache.hadoop.mapreduce.Job;　　// 每个hadoop任务是一个Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;　　// 读取输入
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;　　// 将结果存到输出文件
import org.apache.hadoop.util.GenericOptionsParser;　　// 解析hadoop的命里行参数

public class WordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();　　// 从hadoop配置文件里读取参数
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();　　// 从hadoop命令行读取参数
        if (otherArgs.length != 2) {　　// 从命令行读取的参数正常是两个，分别是输入文件和输出文件的目录
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf, "wordcount");　　// 定义一个新的Job，第一个参数是hadoop配置信息，第二个参数是Job的名字
        job.setJarByClass(WordCount.class);　　// 根据WordCount类的位置设置Jar文件
        job.setMapperClass(WordCountMapper.class);　　// 设置mapper文件　　
        job.setReducerClass(WordCountReducer.class);　　// 设置reducer文件
        job.setOutputKeyClass(Text.class);　　// 设定输出键的类型
        job.setOutputValueClass(IntWritable.class);　　// 设定输出值的类型
        FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 设定输入文件
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 设定输出文件
        System.exit(job.waitForCompletion(true) ? 0 : 1); // 开始执行Job
    }
}

2、mapper程序

package hadoop;

import java.io.IOException;
import java.util.StringTokenizer;　　// java提供的字符串分隔函数

import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;　　// hadoop提供的mapper基类，用户在此基础上进行自己的mapper程序开发

public class WordCountMapper extends Mapper<Object,Text,Text,IntWritable>{　　// ①
    IntWritable one = new IntWritable(1);
    Text word = new Text();

    public void map(Object key,Text value,Context context) throws IOException,InterruptedException{　　// ②
        StringTokenizer itr = new StringTokenizer(value.toString());　　// 将字符串根据空格进行分割（value是Text类型的，所以需要将其转化成String类型进行处理）
        while(itr.hasMoreTokens()){
            word.set(itr.nextToken());
            context.write(word,one);
        }
    }
}

　　① Mapper类包含四个参数，分别用来表示输入数据的key类型、value类型、输出数据的key类型和value类型。在本案例中，输入数据只有一个value没有key，所以将key类型设置为了object，值的类型是Text；对于输出数据，key类型是Text，value的类型是IntWritable。

　　② map方法包含三个参数，分别是输入数据的key类型、value类型和输出数据类型（包含了key和value）

package hadoop;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer; // Reducer基类　　

public class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> {　　// ①
    IntWritable result = new IntWritable();
    public void reduce(Text key,Iterable<IntWritable>values,Context context) throws IOException,InterruptedException{ 　　// ②
        int sum = 0;
        for(IntWritable val:values){
            sum += val.get();
        }
        result.set(sum);
        context.write(key,result);
    }

}

 　　① 和Mapper类一致，Reducer类同样包含四个参数，分别用来表示输入数据的key类型、value类型、输出数据的key类型和value类型。在本案例中，reducer的输入数据key类型为Text，值的类型是一个IntWritable的list；对于输出数据，key类型是Text，value的类型是IntWritable。

　　② reduce方法包含三个参数，分别是输入数据的key类型、value类型和输出数据类型（包含了key和value）

mapper阶段的输入 hello world hello hadoop

mapper阶段的输出 <hello 1> <world 1><hello 1> <hadoop 1>

reducer阶段的输入 <hello <1,1>> <world 1><hadoop 1>

reducer阶段的输出 <hello 2> <world 1><hadoop 1>

二、编译打包

1、编译（*.java —>*.class）

　　首先进入代码目录，运行以下命令：

　　javac -classpath /home/work/usr/hadoop/hadoop-1.2.1/hadoop-core-1.2.1.jar:/home/.../hadoop-1.2.1/lib/commons-cli-1.2.jar

　　　　  -d ./classes/ ./src/*.java

　　（1）javac：JDK的命令行编译器

　　（2）-classpath：设置需要用到的jar包路径，各个jar包之间用":"分隔

　　（3）-d：设置编译后的文件存储路径，本案例中存储在./classes/下，即当前目录的classes子目录

　　（4）最后一个参数是要被编译的java文件，本案例中是存储在./src/目录下的所有java文件，包含上面所讲的三个类

　　注意：hadoop-2.*版本所需要用到的jar包和hadoop-1.*版本有所不同

2、打包

　　jar -cvf wordcount.jar -C ./classes/ .

　　（1）jar：JDK的打包命令行工具

　　（2）-cvf：jar命令的参数

　　

　　（3）注意最后有一个.代表当前目录，把打包结果放在当前目录下

三、任务提交

1、将处理数据提交到HDFS上

　　进入hadoop的安装目录，如上文 cd /home/work/usr/hadoop/hadoop-1.2.1

　　（1）在集群上创建输入文件夹：./bin/hadoop fs -mkdir input

　　（2）上传本地的数据文件到集群input目录：./bin/hadoop fs -put input/* input

　　（3）删除集群上的输出目录（如果目录已经存在会报错）：./bin/hadoop fs -rmr output（删除的时候小心点...）

2、运行程序

　　./bin/hadoop jar /../wordcount.jar hadoop.WordCount input output

　　（1）jar：指定jar包的位置

　　（2）hadoop.WordCount：用户自己定义的包名+主类

　　（3）指定输入和输出路径

3、查看输出结果 

　　./bin/hadoop fs -cat output/part-00000

注意：

（1）mapreduce程序最后的输出文件通常都是以part-00*这种方式命名的

（2）上述用到了很多hdfs的相关命令，对于hdfs上数据的访问，如果知道它的存储位置，也可以直接进入其目录进行一些查看、删除操作

（3）启动任务之后，命令行会返回当前任务的运行进度