自定义输出格式

import java.io.IOException;
import java.io.PrintWriter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CustomFileOutPutFormat extends FileOutputFormat<Text, IntWritable> {

	@Override
	public RecordWriter<Text, IntWritable> getRecordWriter(
			TaskAttemptContext job) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		// 得到文件输出的目录
		Path fileDir = FileOutputFormat.getOutputPath(job);
		// 指定输出的文件名,这里我们为文件取名为1.txt
		//如果有父级目录另作处理
		Path fileName = new Path(fileDir.toString()+"/1.txt");
		System.out.println(fileName.getName());
		Configuration conf = job.getConfiguration();
		FSDataOutputStream file = fileName.getFileSystem(conf).create(fileName);
		return new CustomRecordWrite(file);
	}
}

class CustomRecordWrite extends RecordWriter<Text, IntWritable> {
	private PrintWriter write = null;

	public CustomRecordWrite(FSDataOutputStream file) {
		this.write = new PrintWriter(file);
	}

	@Override
	public void write(Text key, IntWritable value) throws IOException,
			InterruptedException {
		// TODO Auto-generated method stub
		write.println("Word: " + key.toString() + "\t" + "Counts:  " + value);
	}

	@Override
	public void close(TaskAttemptContext context) throws IOException,
			InterruptedException {
		// TODO Auto-generated method stub
		write.close();
	}
}

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCount extends Configured implements Tool {
	@Override
	public int run(String[] arg0) throws Exception {
		// TODO Auto-generated method stub
		Configuration conf = getConf();
		Job job = new Job(conf, "worldcount");
		job.setJarByClass(WordCount.class);
		FileInputFormat.addInputPath(job, new Path("/value/*.txt"));
		FileOutputFormat.setOutputPath(job, new Path("/wordcount/out"));
		job.setMapperClass(WordCountMap.class);
		job.setReducerClass(WordCountReduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		job.setInputFormatClass(TextInputFormat.class);
		// 默认为TextOutputFormat,
		//这里我们设置自定义的输出格式
		job.setOutputFormatClass(CustomFileOutPutFormat.class);
		job.submit();
		return job.isSuccessful() ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		ToolRunner.run(new Configuration(), new WordCount(), null);
	}

}

class WordCountMap extends Mapper<Object, Text, Text, IntWritable> {
	private final static IntWritable one = new IntWritable(1);
	private Text word = new Text();

	public void map(Object key, Text value,
			Mapper<Object, Text, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		StringTokenizer itr = new StringTokenizer(value.toString());
		while (itr.hasMoreTokens()) {
			word.set(itr.nextToken());
			System.out.println(word.toString());
			context.write(word, one);
		}
	}
}

class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
	private 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);
	}
}