[Hadoop]HadoopユニットテストMRUnit

MapReduceでは,map関数とreduce関数の独立したテストが非常に便利であり,これは関数スタイルによって決定される.MRUnitは、既知の入力をmapperに渡したり、reducerの出力が予想に合っているかどうかを確認したりするのに便利なテストライブラリです.MRUnitは、標準の実行フレームワーク(JUnit)とともに使用される.

1.開発環境の設定

(https://repository.apache.org/content/repositories/releases/org/apache/mrunit/mrunit/)MRUnit jarの最新バージョンをダウンロードします.たとえばhadoopバージョンが1.0.3の場合、ダウンロードする必要があります.
mrunit-x.x.x-incubating-hadoop2.jar.JUnitの最新バージョンjarもダウンロードする必要があります.
Maven方式を使用する場合は、次の方法を使用します.

<junit.version>4.12junit.version>
<mrunit.version>1.1.0mrunit.version>

<dependency>
    <groupId>junitgroupId>
    <artifactId>junitartifactId>
    <version>${junit.version}version>
    <scope>testscope>
dependency>

<dependency>
   <groupId>org.apache.mrunitgroupId>
   <artifactId>mrunitartifactId>
   <version>${mrunit.version}version>
   <classifier>hadoop2classifier>
   <scope>testscope>
dependency>

コメント:
もしあなたがhadoopを使っていたら.xバージョン、classifierはhadoop 2に設定

2.MRUnit試験用例

MRUnitテストフレームワークはJunitに基づいて、hadoopバージョンが0.20,0.23であることをテストすることができる.x,1.0.x,2.xのmap reduceプログラム.
次はMRUnitを用いて年間最高気温を集計するMap Reduceプログラムのユニットテストである.
テストデータは次のとおりです.

0096007026999992016062218244+00000+000000FM-15+702699999V0209999C000019999999N999999999+03401+01801999999ADDMA1101731999999REMMET069MOBOB0 METAR 7026 //008 000000 221824Z AUTO 00000KT  34/18 A3004=

これは1日のデータだけで、気温は340で、Mapperはこの天気の温度340に出力します
次は対応するMapperとReducerです.
MaxTemperatureMapper:

package com.sjf.open.maxTemperature;
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;
import com.google.common.base.Objects;
/**
 * Created by xiaosi on 16-7-27.
 */
public class MaxTemperatureMapper extends Mapper {
    private static final int MISSING = 9999;
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        //  
        String year = line.substring(15, 19);
        //  
        int airTemperature;
        if(Objects.equal(line.charAt(87),"+")){
            airTemperature = Integer.parseInt(line.substring(88,92));
        }
        else{
            airTemperature = Integer.parseInt(line.substring(87,92));
        }
        //  
        String quality = line.substring(92, 93);
        if(!Objects.equal(airTemperature, MISSING) && quality.matches("[01459]")){
            context.write(new Text(year), new IntWritable(airTemperature));
        }
    }
}

MaxTemperatureReducer:

package com.sjf.open.maxTemperature;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
 * Created by xiaosi on 16-7-27.
 */
public class MaxTemperatureReducer extends Reducer, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //  
        int maxValue = Integer.MIN_VALUE;
        for(IntWritable value : values){
            maxValue = Math.max(maxValue, value.get());
        }//for
        //  
        context.write(key, new IntWritable(maxValue));
    }
}

次はMRUnitテストクラスです.

package com.sjf.open.maxTemperature;
import com.google.common.collect.Lists;
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.apache.hadoop.mrunit.types.Pair;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.List;
/**
 * Created by xiaosi on 16-12-8.
 */
public class MaxTemperatureTest {
    private MapDriver mapDriver;
    private ReduceDriver reduceDriver;
    private MapReduceDriver mapReduceDriver;
    @Before
    public void setUp(){
        MaxTemperatureMapper mapper = new MaxTemperatureMapper();
        mapDriver = MapDriver.newMapDriver(mapper);
        MaxTemperatureReducer reducer = new MaxTemperatureReducer();
        reduceDriver = ReduceDriver.newReduceDriver();
        reduceDriver.withReducer(reducer);
        mapReduceDriver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
    }
    @Test
    public void testMapper() throws IOException {
        Text text = new Text("0096007026999992016062218244+00000+000000FM-15+702699999V0209999C000019999999N999999999+03401+01801999999ADDMA1101731999999REMMET069MOBOB0 METAR 7026 //008 000000 221824Z AUTO 00000KT  34/18 A3004=");
        mapDriver.withInput(new LongWritable(), text);
        mapDriver.withOutput(new Text("2016"), new IntWritable(340));
        mapDriver.runTest();
        //  
        List<Pair> expectedOutputList = mapDriver.getExpectedOutputs();
        for(Pair pair : expectedOutputList){
            System.out.println(pair.getFirst() + " --- " + pair.getSecond()); // 2016 --- 340
        }
    }
    @Test
    public void testReducer() throws IOException {
        List IntWritableList = Lists.newArrayList();
        IntWritableList.add(new IntWritable(340));
        IntWritableList.add(new IntWritable(240));
        IntWritableList.add(new IntWritable(320));
        IntWritableList.add(new IntWritable(330));
        IntWritableList.add(new IntWritable(310));
        reduceDriver.withInput(new Text("2016"), IntWritableList);
        reduceDriver.withOutput(new Text("2016"), new IntWritable(340));
        reduceDriver.runTest();
        //  
        List<Pair> expectedOutputList = reduceDriver.getExpectedOutputs();
        for(Pair pair : expectedOutputList){
            System.out.println(pair.getFirst() + " --- " + pair.getSecond());
        }
    }
    @Test
    public void testMapperAndReducer() throws IOException {
        Text text = new Text("0089010010999992014010114004+70933-008667FM-12+000999999V0201201N006019999999N999999999+00121-00361100681ADDMA1999990100561MD1810171+9990REMSYN04801001 46/// /1206 10012 21036 30056 40068 58017=");
        mapReduceDriver.withInput(new LongWritable(), text);
        mapReduceDriver.withOutput(new Text("2014"), new IntWritable(12));
        mapReduceDriver.runTest();
        //  
        List<Pair> expectedOutputList = mapReduceDriver.getExpectedOutputs();
        for(Pair pair : expectedOutputList){
            System.out.println(pair.getFirst() + " --- " + pair.getSecond()); // 2014 --- 12
        }
    }
}

テストがMapperの場合はMRUnitのMapDiver、Reducerの場合はReduceDriver、MapReduceプログラム全体をテストする場合はMapReduceDriverが必要です.runTest()メソッドを呼び出す前に、mapper(またはReducer)、入力値、所望の出力key、所望の出力値などを構成する必要があります.所望の出力値と一致しない場合、MRUnitテストに失敗します.withOutput()が呼び出された回数に応じて、MapDiver(ReduceDriver,MapReduceDriver)は、0、1、または複数の出力レコードをチェックすることができる.
コメント:
注意MapDriver、ReduceDriver、MapReduceDriverが導入したjarパッケージのバージョン:

import org.apache.hadoop.mrunit.mapreduce.MapDriver;
import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;

ではなく、

import org.apache.hadoop.mrunit.MapDriver;
import org.apache.hadoop.mrunit.MapReduceDriver;
import org.apache.hadoop.mrunit.ReduceDriver;

これは、Hadoopの古いAPIに対応し、新しいバージョンのMapperとReducerの第1のクラスに対応します.

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

第2のクラスは、古いバージョンのMapperとReducerに対応しています.

import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;

参照先:https://cwiki.apache.org/confluence/display/MRUNIT/MRUnit+Tutorial
データソース:ftp://ftp.ncdc.noaa.gov/pub/data/noaa