使用MRUnit测试hadoop的MapReduce

 

1、MRUnit下载地址：http://mrunit.apache.org/

2、简单的mapred输入、输出测试，示例参考来自：https://cwiki.apache.org/confluence/display/MRUNIT/MRUnit+Tutorial

 

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.mapreduce.Mapper;

public class SMSCDRMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

	private Text status = new Text();

	private final static IntWritable addOne = new IntWritable(1);

	@Override
	// 不建议在方法中使用全类名称：org.apache.hadoop.mapreduce.Mapper.Context
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

		// 655209;1;796764372490213;804422938115889;6
		String[] line = value.toString().split(";");
		if (Integer.parseInt(line[1]) == 1) {
			status.set(line[4]);
			context.write(status, addOne);
		}

	}
}

 

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SMSCDRReducer1 extends Reducer<Text, IntWritable, Text, IntWritable> {

	// Context 不可以直接在方法里头使用org.apache.hadoop.mapreduce.Reducer.Context，而应该在import中使用。
	// 对以上一点，实在是引起大大的疑问题
	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException,
			InterruptedException {
		int sum = 0;
		for (IntWritable value : values) {
			sum += value.get();
		}
		context.write(key, new IntWritable(sum));
	}
}

 

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;
import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
import org.junit.Before;
import org.junit.Test;

public class SMSCDRMapperReducerTest {
	MapDriver<LongWritable,Text,Text,IntWritable> mapDriver;
	ReduceDriver<Text,IntWritable,Text,IntWritable> reducerDriver;
	MapReduceDriver<LongWritable,Text,Text,IntWritable,Text,IntWritable> mapReducerDriver;
	
	@Before
	public void setUp(){
		SMSCDRMapper mapper = new SMSCDRMapper();
		SMSCDRReducer1 reducer = new SMSCDRReducer1();
		mapDriver = MapDriver.newMapDriver(mapper);
		reducerDriver = ReduceDriver.newReduceDriver(reducer);
		mapReducerDriver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
	}
	
	@Test
	public void testMapper() throws Exception{
		mapDriver.withInput(new LongWritable(),new Text("655209;1;796764372490213;804422938115889;6"));
		mapDriver.withOutput(new Text("6"), new IntWritable(1));
		mapDriver.runTest();
	}
	@Test
	public void testReducer() throws Exception {
		List<IntWritable> values = new ArrayList<IntWritable>();
		values.add( new IntWritable(1));
		values.add( new IntWritable(1));
		reducerDriver.withInput(new Text("6"), values);
		reducerDriver.withOutput(new Text("6"),  new IntWritable(2));
		reducerDriver.runTest();
	}
}

 3、通过本地系统进行测试：

1）测试数据：

CDRID;CDRType;Phone1;Phone2;SMS Status Code
655209;1;796764372490213;804422938115889;6
353415;0;356857119806206;287572231184798;4
835699;1;252280313968413;889717902341635;0

 2）Mapper和Reducer与第二点的一样。

3）编写读取本地文件系统的Job。