博客

网上虽然有不少关于MultipleOutputs实现多文件输出的文章，但发现要不还是使用mapred.lib旧接口，要不就是说明不清楚。

Mapper

package com.yy.hiido.itemcf.hadoop.mapper;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class ReadFilesMapper extends
		Mapper<LongWritable, Text, Text, Text> {
	private final String SPLITTER="\\s+";
	private Text outkey;
	private Text values;
	@Override
	protected void setup(Context context)
			throws IOException, InterruptedException {
		super.setup(context);
		outkey = new Text();
		values = new Text();
	}

	@Override
	protected void map(LongWritable key, Text value,Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		//System.out.println("line : "+line);
		String [] fields = line.split(this.SPLITTER);
		outkey.set(fields[0]);
		values.set(fields[1]);
		System.out.println("key : "+fields[0]+"  |  values : "+fields[1]);
		context.write(outkey,values);
	}
}

Reducer

package com.yy.hiido.itemcf.hadoop.reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReadFilesReducer extends
		Reducer<Text, Text, Text, Text> {
	private MultipleOutputs<Text,Text> mos;
	@Override
	protected void setup(Context context)
			throws IOException, InterruptedException {
		mos = new MultipleOutputs<Text,Text>(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values,Context context)
			throws IOException, InterruptedException {
		String temp="";
		for(Text t : values)
			temp+=t.toString()+" | ";
		//mos.write(key.toString(), key, new Text(temp));//这样需要预定义named output
		mos.write(key, new Text(temp), key.toString());//这样不需要与定义named output
	}

	@Override
	protected void cleanup(Context context)
			throws IOException, InterruptedException {
		mos.close();;
	}
}

main

package com.yy.hiido.itemcf.hadoop.job;

import java.io.IOException;
import java.net.URI;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
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.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.yy.hiido.itemcf.hadoop.mapper.ReadFilesMapper;
import com.yy.hiido.itemcf.hadoop.reducer.ReadFilesReducer;

/**
 * @author BlackWing 测试multioutput
 * */

public class TestMultiOutputJob extends Configured implements Tool {
	// 配置文件名
	public final static String propertyFileName = "config.xml";
	private static final Log LOG = LogFactory.getLog(TestMultiOutputJob.class);
	private static Configuration conf = HBaseConfiguration.create();

	public int myjob() {
		// 读配置文件
		conf.addResource(propertyFileName);
		String input = conf.get("file.to.read");
		String outputDir = conf.get("test.output");
		System.out.println("input : "+input);
		System.out.println("output : "+outputDir);

		// 若输出目录存在,则删除
		try {
			FileSystem fs = FileSystem.get(URI.create(outputDir),
					new Configuration());
			fs.delete(new Path(outputDir), true);
			fs.close();
		} catch (Exception e) {
			e.printStackTrace();
		}

		Job myjob = null;

		try {
			myjob = new Job(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
		myjob.setJarByClass(ReadFilesMapper.class);

		try {
			FileInputFormat.setInputPaths(myjob, input);
			FileOutputFormat.setOutputPath(myjob, new Path(outputDir));
		} catch (IOException e1) {
			e1.printStackTrace();
		}
		myjob.setMapperClass(ReadFilesMapper.class);
		myjob.setInputFormatClass(TextInputFormat.class);
//		myjob.setOutputFormatClass(TextOutputFormat.class);
		LazyOutputFormat.setOutputFormatClass(myjob, TextOutputFormat.class);
		myjob.setReducerClass(ReadFilesReducer.class);
		myjob.setOutputKeyClass(Text.class);
		myjob.setOutputValueClass(Text.class);

		//MultipleOutputs.addNamedOutput(myjob, "moshouzhengba", TextOutputFormat.class, Text.class, Text.class);
		//MultipleOutputs.addNamedOutput(myjob, "maoxiandao", TextOutputFormat.class, Text.class, Text.class);
		//MultipleOutputs.addNamedOutput(myjob, "yingxionglianmen", TextOutputFormat.class, Text.class, Text.class);

		boolean succeeded = false;
		try {
			succeeded = myjob.waitForCompletion(true);
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
		if (!succeeded)
			return -1;
		// status记录job运行状态
		LOG.info("Job complete !");
		return 1;
	}

	@Override
	public int run(String[] as) throws Exception {
		// 读配置文件
		conf.addResource(propertyFileName);
		myjob();
		return 0;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		long start = System.currentTimeMillis();
		try {
			ToolRunner.run(new TestMultiOutputJob(), args);
		} catch (Exception e) {
			e.printStackTrace();
		}
		long end = System.currentTimeMillis();
		System.out.println("run the program costs time:" + (end - start)
				/ 60000 + "Minutes");
	}
}

其中注意的是：

1.reducer中调用时，要调用MultipleOutputs以下接口：

write

public void write(KEYOUT key,
                  VALUEOUT value,
                  String baseOutputPath)
           throws IOException,
                  InterruptedException

如果调用

write

public <K,V> void write(String namedOutput,
                        K key,
                        V value)
           throws IOException,
                  InterruptedException

则需要在job中，预先声明named output（如下），不然会报错：named output xxx not defined：

MultipleOutputs.addNamedOutput(myjob, "moshouzhengba", TextOutputFormat.class, Text.class, Text.class);
		MultipleOutputs.addNamedOutput(myjob, "maoxiandao", TextOutputFormat.class, Text.class, Text.class);
		MultipleOutputs.addNamedOutput(myjob, "yingxionglianmen", TextOutputFormat.class, Text.class, Text.class);

2.默认情况下，输出目录会生成part-r-00000或者part-m-00000的空文件，需要如下设置后，才不会生成：

//		myjob.setOutputFormatClass(TextOutputFormat.class);
		LazyOutputFormat.setOutputFormatClass(myjob, TextOutputFormat.class);

就是去掉job设置outputFormatClass，改为通过LazyOutputFormat设置

这里只是以Text的输入输出格式说明。

官方的文档：

https://hadoop.apache.org/docs/current2/api/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.html#write(java.lang.String, K, V)