【总结】Hadoop中的MultipleOutputs实践

package StationPatitioner;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleOutputs;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * hadoop Version 1.1.2
 * MultipleOutputs例子
 * @author 巧克力黑
 *
 */
public class PatitionByStationUsingMultipleOutputs extends Configured implements Tool {
	enum Counter 
	{
		LINESKIP,	//出错的行
	}
	static class StationMapper extends MapReduceBase implements Mapper<LongWritable , Text, Text , Text>{
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			try {
				parser.parse(value);
				output.collect(new Text(parser.getStationid()), value);
			} catch (Exception e) {
				reporter.getCounter(Counter.LINESKIP).increment(1);	//出错令计数器+1
			}
			
		}
	}

	static class MultipleOutputReducer extends MapReduceBase implements Reducer<Text, Text, NullWritable, Text>{
		private MultipleOutputs multipleOutputs;
		@Override
		public void configure(JobConf jobconf) {
			multipleOutputs = new MultipleOutputs(jobconf);//初始化一个MultipleOutputs
		}
		
		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<NullWritable, Text> output, Reporter reporter)
				throws IOException {
			//得到OutputCollector
			OutputCollector collector = multipleOutputs.getCollector("station", key.toString().replace("-", ""), reporter);
			while(values.hasNext()){
				collector.collect(NullWritable.get(), values.next());//MultipleOutputs用OutputCollector输出数据
			}
		}
		
		@Override
		public void close() throws IOException {
			multipleOutputs.close();
		}
	}
	
	@Override
	public int run(String[] as) throws Exception {
		System.setProperty("HADOOP_USER_NAME", "root");//windows下用户与linux用户不一直，采用此方法避免报Permission相关错误
		JobConf conf = new JobConf();
		
		conf.setMapperClass(StationMapper.class);
		conf.setReducerClass(MultipleOutputReducer.class);
		conf.setMapOutputKeyClass(Text.class);
		conf.setOutputKeyClass(NullWritable.class);
		conf.setOutputFormat(NullOutputFormat.class);
	    FileInputFormat.setInputPaths(conf, new Path("hdfs://ubuntu:9000/sample1.txt"));//input路径
	    FileOutputFormat.setOutputPath(conf, new Path("hdfs://ubuntu:9000/temperature"));//output路径
		
		MultipleOutputs.addMultiNamedOutput(conf, "station", TextOutputFormat.class, NullWritable.class, Text.class);
		
		JobClient.runJob(conf);
		return 0;
	}
	
	public static void main(String[] args) throws Exception{
		int exitCode = ToolRunner.run(new PatitionByStationUsingMultipleOutputs(), args);
		System.exit(exitCode);
	}
	
}

package StationPatitioner;

import org.apache.hadoop.io.Text;

public class NcdcRecordParser {
	private static final int MISSING_TEMPERATURE = 9999;

	private String year;
	private int airTemperature;
	private String quality;
	private String stationid;

	public void parse(String record) {
		stationid = record.substring(0, 5);
		year = record.substring(15, 19);
		String airTemperatureString;
		// Remove leading plus sign as parseInt doesn't like them
		if (record.charAt(87) == '+') {
			airTemperatureString = record.substring(88, 92);
		} else {
			airTemperatureString = record.substring(87, 92);
		}
		airTemperature = Integer.parseInt(airTemperatureString);
		quality = record.substring(92, 93);
	}
	
	public String getStationid(){
		return stationid;
	}

	public void parse(Text record) {
		parse(record.toString());
	}

	public boolean isValidTemperature() {
		return airTemperature != MISSING_TEMPERATURE && quality.matches("[01459]");
	}

	public String getYear() {
		return year;
	}

	public int getAirTemperature() {
		return airTemperature;
	}
}