MapReduce

Java MapReduce

明白MapReduce 程序的工作原理之后，下一步便是通过代码来实现它。我们需要三样东西：一个map 函数、一个reduce 函数和一些用来运行作业的代码。map函数由Mapper 接口实现来表示，后者声明了一个map()方法。例2-3 显示了我们的map函数实现。

例2-3. 查找最高气温的Mapper

import java.io.IOException;   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 MaxTemperatureMapper extends MapReduceBase       implements Mapper<LongWritable, Text, Text, IntWritable> {     private static final int MISSING = 9999;          public void map(LongWritable key, Text value,               OutputCollector<Text, IntWritable> output, Reporter reporter)         throws IOException {                String line = value.toString();           String year = line.substring(15, 19);           int airTemperature;           if (line.charAt(87) == '+') { // parseIntdoesn't like leading plus signs               airTemperature = Integer.parseInt(line.substring(88, 92));           } else {               airTemperature = Integer.parseInt(line.substring(87, 92));           }           String quality = line.substring(92, 93);           if (airTemperature != MISSING && quality.matches("[01459]")) {               output.collect(new Text(year), new IntWritable(airTemperature));           }       }   }  该Mapper接口是一个泛型类型，它有四个形参类型，分别指定map函数的输入键、输入值、输出键和输出值的类型。就目前的示例来说，输入键是一个长整数偏移量，输入值是一行文本，输出键是年份，输出值是气温(整数)。Hadoop自身提供一套可优化网络序列化传输的基本类型，而不直接使用Java内嵌的类型。这些类型均可在org.apache.hadoop.io包中找到。这里我们使用LongWritable类型(相当于Java中的Long类型)、Text类型(相当于Java中的String类型)和IntWritable类型(相当于Java 中的Integer类型)。

map()方法的输入是一个键和一个值。我们首先将包含有一行输入的Text值转换成Java的String类型，之后使用substring()方法提取我们感兴趣的列。

map()方法还提供了OutputCollector实例用于输出内容的写入。在这种情况下，我们将年份数据按Text对象进行读/写 (因为我们把年份当作键)，将气温值封装在IntWritable 类型中。

我们只在气温数据不缺失并且所对应质量代码显示为正确的气温读数时，才将其写入输出记录中。

reduce函数通过Reducer进行类似的定义，如例2-4 所示。

例2-4. 查找最高气温的Reducer

import java.io.IOException;   import java.util.Iterator;   import org.apache.hadoop.io.IntWritable;   import org.apache.hadoop.io.Text;   import org.apache.hadoop.mapred.MapReduceBase;   import org.apache.hadoop.mapred.OutputCollector;   import org.apache.hadoop.mapred.Reducer;   import org.apache.hadoop.mapred.Reporter;   public class MaxTemperatureReducer 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 maxValue = Integer.MIN_VALUE;           while (values.hasNext()) {               maxValue = Math.max(maxValue, values.next().get());      }           output.collect(key, new IntWritable(maxValue));       }  }  同样，针对reduce函数也有四个形式参数类型用于指定其输入和输出类型。reduce 函数的输入类型必须与map 函数的输出类型相匹配：即Text类型和IntWritable类型。在这种情况下，reduce函数的输出类型也必须是Text和IntWritable这两种类型，分别输出年份和最高气温。该最高气温是通过循环比较当前气温与已看到的最高气温获得的。

第三部分代码负责运行MapReduce 作业(请参见例2-5)。

例2-5. 该应用程序在气象数据集中找出最高气温

import java.io.IOException;   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 MaxTemperature {     public static void main(String[] args) throws IOException {           if (args.length != 2) {        System.err.println("Usage: MaxTemperature<input path> <output path>");               System.exit(-1);           }                JobConf conf = new JobConf(MaxTemperature.class);           conf.setJobName("Max temperature");            FileInputFormat.addInputPath(conf, new Path(args[0]));           FileOutputFormat.setOutputPath(conf, new Path(args[1]));       conf.setMapperClass(MaxTemperatureMapper.class);               conf.setReducerClass(MaxTemperatureReducer.class);      conf.setOutputKeyClass(Text.class);           conf.setOutputValueClass(IntWritable.class);            JobClient.runJob(conf);       }  }  JobConf对象指定了作业执行规范。我们可以用它来控制整个作业的运行。在Hadoop 集群上运行这个作业时，我们需要将代码打包成一个JAR文件(Hadoop会在集群上分发这个文件)。我们无需明确指定JAR 文件的名称，而只需在JobConf的构造函数中传递一个类，Hadoop将通过该类查找包含有该类的JAR文件进而找到相关的JAR文件。

构造JobConf对象之后，需要指定输入和输出数据的路径。调用 FileInputFormat类的静态函数addInputPath()来定义输入数据的路径，该路径可以是单个文件、目录(此时，将目录下所有文件当作输入)或符合特定文件模式的一组文件。由函数名可知，可以多次调用addInputPath()实现多路径的输入。

通过调用FileOutputFormat 类中的静态函数 setOutputPath()来指定输出路径。该函数指定了reduce 函数输出文件的写入目录。在运行任务前该目录不应该存在，否则Hadoop 会报错并拒绝运行该任务。这种预防措施是为了防止数据丢失(一个长时间运行任务的结果被意外地覆盖将是非常恼人的)。

接着，通过setMapperClass()和setReducerClass()指定map和reduce类型。

setOutputKeyClass()和setOutputValueClass()控制map和reduce函数的输出类型，正如本例所示，这两个输出类型往往相同。如果不同，map函数的输出类型则通过setMapOutputKeyClass()和setMapOutputValueClass()函数来设置。

输入的类型通过InputFormat类来控制，我们的例子中没有设置，因为使用的是默认的TextInputFormat(文本输入格式)。

在设置定义map 和reduce 函数的类后，便可以开始运行任务。JobClient类的静态函数runJob()会提交作业并等待完成，最后将其进展情况写到控制台。