使用MRUnit测试hadoop的MapReduce

kevin0490

　　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。