使用MRUnit进行MapReduce单元测试

1.前言

在写完MR之后，通常都会自己造一些数据本地测一下保证基本逻辑没问题。这里使用MRUnit进行MR的单元测试
官网地址：https://mrunit.apache.org/

             这里笨小葱使用MRUnit来测试一下最简单的WordCount的MR代码。

2.maven配置

        这里需要注意 引入mrunit的jar包时需要加上<classifier>hadoop2</classifier>,来区分对应的hadoop版本

<dependency>
    <groupId>org.apache.mrunit</groupId>
    <artifactId>mrunit</artifactId>
    <version>1.0.0</version>
    <classifier>hadoop2</classifier>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.10</version>
    <scope>test</scope>
</dependency>

3.WordCount代码

public class WordCount {
    public static class WCMap extends Mapper<LongWritable, Text, Text, IntWritable>
    {
        private Text k=new Text();
        private final static IntWritable addOne = new IntWritable(1);
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] valueArr=value.toString().split(" ");
            for (String val:valueArr)
            {
                k.set(val);
                context.write(k,addOne);
            }
        }
    }

    public static class WCReduce extends Reducer<Text,IntWritable,Text,IntWritable>
    {
        private  static  IntWritable sum=new IntWritable();
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int count=0;
            for(IntWritable i:values)
            {
                count+=i.get();
            }
            sum.set(count);
            context.write(key,sum);
        }
    }
}

3.MRunit测试代码

public class WordCountMRTest {
    MapDriver<LongWritable, Text, Text, IntWritable> mapDriver;
    ReduceDriver<Text, IntWritable, Text, IntWritable> reduceDriver;
    MapReduceDriver<LongWritable, Text, Text, IntWritable, Text, IntWritable> mapReduceDriver;

    @Before
    public void setUp() {
        //测试mapreduce
        WordCount.WCMap mapper = new WordCount.WCMap();
        WordCount.WCReduce reducer = new WordCount.WCReduce();
        mapDriver = MapDriver.newMapDriver(mapper);
        reduceDriver = ReduceDriver.newReduceDriver(reducer);
        mapReduceDriver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
    }

    //测试mapper
    @Test
    public void testMapper() throws IOException {
        //output  {(a,1),(b,1),(c,1).....}
        List<Pair<Text, IntWritable>> outputRecords=new ArrayList<Pair<Text, IntWritable>>();
        outputRecords.add(new Pair<Text, IntWritable>(new Text("a"),new IntWritable(1)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("b"),new IntWritable(1)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("c"),new IntWritable(1)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("d"),new IntWritable(1)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("a"),new IntWritable(1)));

        //input "a b c d a"
        mapDriver.withInput(new LongWritable(), new Text(
                "a b c d a"));
        mapDriver.withAllOutput(outputRecords);
        mapDriver.runTest();
    }

    //测试mapper
    @Test
    public void testReducer() throws Exception{
        //input {(a,{1,1}),(b,{1,1,1})}
        List<Pair<Text, List<IntWritable>>> allInput=new ArrayList<Pair<Text, List<IntWritable>>>();
        List<IntWritable> list=new ArrayList<IntWritable>();
        list.add(new IntWritable(1));
        list.add(new IntWritable(1));

        List<IntWritable> list2=new ArrayList<IntWritable>();
        list2.add(new IntWritable(1));
        list2.add(new IntWritable(1));
        list2.add(new IntWritable(1));

        allInput.add(new Pair<Text, List<IntWritable>>(new Text("a"),list));
        allInput.add(new Pair<Text, List<IntWritable>>(new Text("b"),list2));

        //output {(a,2),(b,3)}
        List<Pair<Text, IntWritable>> outputRecords=new ArrayList<Pair<Text, IntWritable>>();
        outputRecords.add(new Pair<Text, IntWritable>(new Text("a"),new IntWritable(2)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("b"),new IntWritable(3)));

        reduceDriver.withAll(allInput);
        reduceDriver.withAllOutput(outputRecords);
        reduceDriver.runTest();
    }


    @Test
    public void testMR() throws Exception{
        //output {(a,2),(b,1),(c,2)}
        List<Pair<Text, IntWritable>> outputRecords=new ArrayList<Pair<Text, IntWritable>>();
        outputRecords.add(new Pair<Text, IntWritable>(new Text("a"),new IntWritable(2)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("b"),new IntWritable(1)));
        outputRecords.add(new Pair<Text, IntWritable>(new Text("c"),new IntWritable(2)));

        //input "a b c c a"
        mapReduceDriver.withInput(new LongWritable(), new Text("a b c c a"));
        mapReduceDriver.withAllOutput(outputRecords);
        mapReduceDriver.runTest();
    }
}