java.lang.NullPointerException с использованием MRUnit. Ошибка стандартной сериализации ключа

Question

Я пытаюсь протестировать простой проект MapReduce, используя MRUnit. Я устанавливаю вход для mapDriver, а затем вызываю mapDriver.runTest() (я также пробовал с mapDriver.run(), но выдает ту же ошибку).

Я написал специальный ключ, который перегружает write(DataOutput out), readFields(DataInput in) и compareTo(...) Методы. При отладке ключ правильно записывает свои данные с помощью write(DataOutput out). Однако после того, как метод ключа readFields(DataInput in) (который корректно извлекает данные, которые были ранее записаны с использованием write(DataOutput out)), заканчивается, появляется ошибка ниже.

Я нашел здесь похожие записи и попробовал переопределить методы hashCode() и equals() безрезультатно. Требует ли MRUnit каких-либо дополнительных методов для переопределения при использовании пользовательских ключей? Этот пост больше всего похож на MRUnit with Avro NullPointerException in Serialization. Тем не менее, я не использую avro, и насколько мне известно, я использую сериализацию по умолчанию. Ура!

java.lang.NullPointerException 
at org.apache.hadoop.mrunit.Serialization.copy(Serialization.java:61) 
at org.apache.hadoop.mrunit.Serialization.copy(Serialization.java:81) 
at org.apache.hadoop.mrunit.mapreduce.mock.MockContextWrapper$4.answer(MockContextWrapper.java:78) 
at org.mockito.internal.stubbing.StubbedInvocationMatcher.answer(StubbedInvocationMatcher.java:31) 
at org.mockito.internal.MockHandler.handle(MockHandler.java:97) 
at org.mockito.internal.creation.MethodInterceptorFilter.intercept(MethodInterceptorFilter.java:47) 
at org.apache.hadoop.mapreduce.Mapper$Context$$EnhancerByMockitoWithCGLIB$$f555e120.write(<generated>) 
at model.RMSEEvaluation$Mapper.map(RMSEEvaluation.java:57) 
at model.RMSEEvaluation$Mapper.map(RMSEEvaluation.java:1) 
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144) 
at org.apache.hadoop.mrunit.mapreduce.MapDriver.run(MapDriver.java:221) 
at org.apache.hadoop.mrunit.MapDriverBase.runTest(MapDriverBase.java:150) 
at org.apache.hadoop.mrunit.TestDriver.runTest(TestDriver.java:137) 
at test.TestRMSEEvaluation.testSetValues(TestRMSEEvaluation.java:77) 
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) 
at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) 
at java.lang.reflect.Method.invoke(Unknown Source) 
at junit.framework.TestCase.runTest(TestCase.java:168) 
at junit.framework.TestCase.runBare(TestCase.java:134) 
at junit.framework.TestResult$1.protect(TestResult.java:110) 
at junit.framework.TestResult.runProtected(TestResult.java:128) 
at junit.framework.TestResult.run(TestResult.java:113) 
at junit.framework.TestCase.run(TestCase.java:124) 
at junit.framework.TestSuite.runTest(TestSuite.java:243) 
at junit.framework.TestSuite.run(TestSuite.java:238) 
at org.junit.internal.runners.JUnit38ClassRunner.run(JUnit38ClassRunner.java:83) 
at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:50) 
at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38) 
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467) 
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683) 
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390) 
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197) 

Answer 1

Я нашел решение этой ошибки. Ошибка состояла в том, что тип сериализации не был установлен в Configuration для MapDriver mapDriver. Мне пришлось явно установить сериализацию, используя следующее:

Configuration conf = new Configuration(); 
conf.set("io.serializations","org.apache.hadoop.io.serializer.JavaSerialization," 
      + "org.apache.hadoop.io.serializer.WritableSerialization"); 
mapDriver.setConfiguration(conf); 

Надеюсь, это поможет любому, у кого есть аналогичная проблема!

Answer 2

Во-первых, это стоит test работает ли сериализации/десериализации действительно, как и ожидалось.
Не зная, как вы написали тест, следующая простые один хорошо работает с MRUnit 0.9.0-incubating и JUnit 4,10:

import java.io.IOException; 
import java.util.ArrayList; 
import java.util.List; 

import junit.framework.Assert; 

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
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; 

public class TestCustom { 

    private MapDriver<CustomRecord, Text, CustomRecord, Text> mapDriver; 

    private Mapper<CustomRecord, Text, CustomRecord, Text> map = 
     new Mapper<CustomRecord, Text, CustomRecord, Text>(); 

    private Reducer<CustomRecord, Text, CustomRecord, Text> reduce = 
     new Reducer<CustomRecord, Text, CustomRecord, Text>(); 

    private ReduceDriver<CustomRecord, Text, CustomRecord, Text> reduceDriver 
     = ReduceDriver.newReduceDriver(reduce); 

    private MapReduceDriver<CustomRecord, Text, CustomRecord, 
     Text, CustomRecord, Text> mapReduceDriver; 

    private Configuration conf = new Configuration(); 

    //test data 
    private Pair<CustomRecord, Text> data; 

    //shuffled and sorted data 
    private static List<Pair<CustomRecord, List<Text>>> shuffledData; 

    @Before 
    public void init() { 

     mapDriver = MapDriver.newMapDriver(map); 
     mapReduceDriver = MapReduceDriver.newMapReduceDriver(map, reduce); 
     mapDriver.withConfiguration(conf); 
     initData(); 
    } 

    private void initData() { 

     CustomRecord key = new CustomRecord("first", 1); 
     Text value = new Text("key1"); 
     data = new Pair<CustomRecord, Text>(key, value); 
    } 

    @Test 
    public void testMapper() throws IOException { 

     mapDriver.withInput(data); 
     //expected output result 
     mapDriver.withOutput(data); 
     mapDriver.runTest(true); 

     //shuffle and sort 
     List<Pair<CustomRecord, Text>> pairs = 
      new ArrayList<Pair<CustomRecord, Text>>(); 
     pairs.add(data); 

     shuffledData = mapReduceDriver.shuffle(pairs); 

    } 

    @Test 
    public void testReducer() throws IOException { 

     // feed input to one single reduce call 
     Pair<CustomRecord, List<Text>> pair = shuffledData.get(0); 
     reduceDriver.withInput(pair.getFirst(), pair.getSecond()); 

     //reducer's output 
     List<Pair<CustomRecord, Text>> result = reduceDriver.run(); 

     Assert.assertEquals("Key mismatch!", 
      data.getFirst(), result.get(0).getFirst()); 
     Assert.assertEquals("Value mismatch!", 
      data.getSecond(), result.get(0).getSecond()); 
    } 
} 

Он проверяет картограф идентичности и редуктор с настраиваемой доступна для записи в качестве ключа (CustomRecord).
Обратите внимание, что ключ реализует WritableComparable и переопределяет hashCode и равно.