You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrunit.apache.org by br...@apache.org on 2011/11/28 22:30:32 UTC
svn commit: r1207626 - in /incubator/mrunit/trunk/src:
main/java/org/apache/hadoop/mrunit/
main/java/org/apache/hadoop/mrunit/mapreduce/
main/java/org/apache/hadoop/mrunit/mapreduce/mock/
test/java/org/apache/hadoop/mrunit/mapreduce/
Author: brock
Date: Mon Nov 28 21:30:29 2011
New Revision: 1207626
URL: http://svn.apache.org/viewvc?rev=1207626&view=rev
Log:
MRUNIT-19: MRunit ReduceDriver does not fully replicate the reducers behaviour with reusing pointers over the iterator
Added:
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/Serialization.java
incubator/mrunit/trunk/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestReducerInputValueResuse.java
Modified:
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriver.java
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReduceDriver.java
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockContextWrapper.java
incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockMapContextWrapper.java
Modified: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriver.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriver.java?rev=1207626&r1=1207625&r2=1207626&view=diff
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriver.java (original)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriver.java Mon Nov 28 21:30:29 2011
@@ -196,7 +196,7 @@ public class ReduceDriver<K1, V1, K2, V2
new MockOutputCollector<K2, V2>(getConfiguration());
MockReporter reporter = new MockReporter(MockReporter.ReporterType.Reducer, getCounters());
- myReducer.reduce(inputKey, inputValues.iterator(), outputCollector,
+ myReducer.reduce(inputKey, getInputValues().iterator(), outputCollector,
reporter);
List<Pair<K2, V2>> outputs = outputCollector.getOutputs();
Modified: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java?rev=1207626&r1=1207625&r2=1207626&view=diff
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java (original)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java Mon Nov 28 21:30:29 2011
@@ -17,36 +17,51 @@
*/
package org.apache.hadoop.mrunit;
-
+import static org.apache.hadoop.mrunit.Serialization.copy;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.types.Pair;
+import org.apache.hadoop.util.ReflectionUtils;
/**
* Harness that allows you to test a Reducer instance. You provide a key and a
- * set of intermediate values for that key that represent inputs that should
- * be sent to the Reducer (as if they came from a Mapper), and outputs you
- * expect to be sent by the Reducer to the collector. By calling runTest(),
- * the harness will deliver the input to the Reducer and will check its
- * outputs against the expected results. This is designed to handle a single
- * (k, v*) -> (k, v)* case from the Reducer, representing a single unit test.
- * Multiple input (k, v*) sets should go in separate unit tests.
+ * set of intermediate values for that key that represent inputs that should be
+ * sent to the Reducer (as if they came from a Mapper), and outputs you expect
+ * to be sent by the Reducer to the collector. By calling runTest(), the harness
+ * will deliver the input to the Reducer and will check its outputs against the
+ * expected results. This is designed to handle a single (k, v*) -> (k, v)* case
+ * from the Reducer, representing a single unit test. Multiple input (k, v*)
+ * sets should go in separate unit tests.
*/
-public abstract class ReduceDriverBase<K1, V1, K2, V2> extends TestDriver<K1, V1, K2, V2> {
+public abstract class ReduceDriverBase<K1, V1, K2, V2> extends
+ TestDriver<K1, V1, K2, V2> {
protected K1 inputKey;
- protected List<V1> inputValues;
+ private List<V1> inputValues;
public ReduceDriverBase() {
inputValues = new ArrayList<V1>();
}
/**
+ * Returns a list which when iterated over, returns
+ * the same instance of the value each time with different
+ * contents similar to how Hadoop currently works with
+ * Writables.
+ *
+ * @return List of values
+ */
+ public List<V1> getInputValues() {
+ return new ValueClassInstanceReuseList<V1>(inputValues);
+ }
+ /**
* Sets the input key to send to the Reducer
- *
+ *
*/
public void setInputKey(K1 key) {
inputKey = key;
@@ -54,7 +69,7 @@ public abstract class ReduceDriverBase<K
/**
* adds an input value to send to the reducer
- *
+ *
* @param val
*/
public void addInputValue(V1 val) {
@@ -63,7 +78,7 @@ public abstract class ReduceDriverBase<K
/**
* Sets the input values to send to the reducer; overwrites existing ones
- *
+ *
* @param values
*/
public void setInputValues(List<V1> values) {
@@ -73,7 +88,7 @@ public abstract class ReduceDriverBase<K
/**
* Adds a set of input values to send to the reducer
- *
+ *
* @param values
*/
public void addInputValues(List<V1> values) {
@@ -82,7 +97,7 @@ public abstract class ReduceDriverBase<K
/**
* Sets the input to send to the reducer
- *
+ *
* @param values
*/
public void setInput(K1 key, List<V1> values) {
@@ -92,7 +107,7 @@ public abstract class ReduceDriverBase<K
/**
* Adds an output (k, v) pair we expect from the Reducer
- *
+ *
* @param outputRecord
* The (k, v) pair to add
*/
@@ -106,9 +121,11 @@ public abstract class ReduceDriverBase<K
/**
* Adds an output (k, v) pair we expect from the Reducer
- *
- * @param key The key part of a (k, v) pair to add
- * @param val The val part of a (k, v) pair to add
+ *
+ * @param key
+ * The key part of a (k, v) pair to add
+ * @param val
+ * The val part of a (k, v) pair to add
*/
public void addOutput(K2 key, V2 val) {
addOutput(new Pair<K2, V2>(key, val));
@@ -117,7 +134,7 @@ public abstract class ReduceDriverBase<K
/**
* Expects an input of the form "key \t val, val, val..." Forces the Reducer
* input types to Text.
- *
+ *
* @param input
* A string of the form "key \t val,val,val". Trims any whitespace.
*/
@@ -132,10 +149,9 @@ public abstract class ReduceDriverBase<K
// this.
setInputKey((K1) inputPair.getFirst());
setInputValues((List<V1>) parseCommaDelimitedList(inputPair.getSecond()
- .toString()));
+ .toString()));
} else {
- throw new IllegalArgumentException(
- "Could not parse input pair");
+ throw new IllegalArgumentException("Could not parse input pair");
}
}
}
@@ -143,7 +159,7 @@ public abstract class ReduceDriverBase<K
/**
* Expects an input of the form "key \t val" Forces the Reducer output types
* to Text.
- *
+ *
* @param output
* A string of the form "key \t val". Trims any whitespace.
*/
@@ -188,5 +204,45 @@ public abstract class ReduceDriverBase<K
throw new RuntimeException("IOException in reducer: ", ioe);
}
}
-}
+ protected static class ValueClassInstanceReuseList<T> extends ArrayList<T> {
+ private static final long serialVersionUID = 1L;
+ private T value;
+ private final Configuration conf = new Configuration();
+ @SuppressWarnings("unchecked")
+ public ValueClassInstanceReuseList(List<T> list) {
+ super(list);
+ if (!list.isEmpty()) {
+ T first = list.get(0);
+ Class<T> klass = (Class<T>) first.getClass();
+ this.value = ReflectionUtils.newInstance(klass, conf);
+ }
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ final Iterator<T> listIterator = super.iterator();
+ final T currentValue = value;
+ return new Iterator<T>() {
+ private T value = currentValue;
+ private Iterator<T> iterator = listIterator;
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public T next() {
+ T next = iterator.next();
+ copy(next, value, conf);
+ return value;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ }
+}
Added: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/Serialization.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/Serialization.java?rev=1207626&view=auto
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/Serialization.java (added)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/Serialization.java Mon Nov 28 21:30:29 2011
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mrunit;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.SerializationFactory;
+import org.apache.hadoop.io.serializer.Serializer;
+import org.apache.hadoop.util.ReflectionUtils;
+
+public class Serialization {
+
+ @SuppressWarnings("unchecked")
+ public static void copy(Object orig, Object copy, Configuration conf) {
+ if(orig == null || copy == null) {
+ return;
+ }
+ if(orig.getClass() != copy.getClass()) {
+ throw new IllegalArgumentException(orig.getClass() + "!=" + copy.getClass());
+ }
+ Class<?> clazz = orig.getClass();
+ SerializationFactory serializationFactory = new SerializationFactory(conf);
+ Serializer<Object> serializer = (Serializer<Object>) serializationFactory.getSerializer(orig.getClass());
+ Deserializer<Object> deserializer = (Deserializer<Object>) serializationFactory.getDeserializer(clazz);
+ try {
+ DataOutputBuffer outputBuffer = new DataOutputBuffer();
+ serializer.open(outputBuffer);
+ serializer.serialize(orig);
+ DataInputBuffer inputBuffer = new DataInputBuffer();
+ inputBuffer.reset(outputBuffer.getData(), outputBuffer.getLength());
+ deserializer.open(inputBuffer);
+ deserializer.deserialize(copy);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static Object copy(Object orig, Configuration conf) {
+ if(orig == null) {
+ return null;
+ }
+ Class<?> clazz = orig.getClass();
+ Object copy = ReflectionUtils.newInstance(clazz, conf);
+ copy(orig, copy, conf);
+ return copy;
+ }
+
+}
Modified: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReduceDriver.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReduceDriver.java?rev=1207626&r1=1207625&r2=1207626&view=diff
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReduceDriver.java (original)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReduceDriver.java Mon Nov 28 21:30:29 2011
@@ -194,7 +194,7 @@ public class ReduceDriver<K1, V1, K2, V2
@Override
public List<Pair<K2, V2>> run() throws IOException {
List<Pair<K1, List<V1>>> inputs = new ArrayList<Pair<K1, List<V1>>>();
- inputs.add(new Pair<K1, List<V1>>(inputKey, inputValues));
+ inputs.add(new Pair<K1, List<V1>>(inputKey, getInputValues()));
try {
MockReduceContextWrapper<K1, V1, K2, V2> wrapper =
Modified: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockContextWrapper.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockContextWrapper.java?rev=1207626&r1=1207625&r2=1207626&view=diff
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockContextWrapper.java (original)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockContextWrapper.java Mon Nov 28 21:30:29 2011
@@ -20,6 +20,7 @@ package org.apache.hadoop.mrunit.mapredu
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.hadoop.mrunit.Serialization.copy;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
@@ -30,10 +31,6 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.DataInputBuffer;
-import org.apache.hadoop.io.DataOutputBuffer;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
@@ -79,34 +76,14 @@ public abstract class MockContextWrapper
public Object answer(InvocationOnMock invocation) {
Object[] args = checkNotNull(invocation.getArguments());
checkArgument(args.length == 2);
- outputs.add(new Pair(copy(args[0]), copy(args[1])));
+ outputs.add(new Pair(copy(args[0], conf), copy(args[1], conf)));
return null;
}}).when(context).write((KEYOUT)any(), (VALUEOUT)any());
}
+
+
- public static Object copy(Object orig) {
- if(orig instanceof Writable) {
- if(orig instanceof NullWritable) {
- return orig;
- }
- try {
- Writable original = (Writable) orig;
- DataOutputBuffer out = new DataOutputBuffer();
- original.write(out);
- DataInputBuffer in = new DataInputBuffer();
- byte[] buff = out.getData();
- in.reset(buff, buff.length);
- Writable copy;
- copy = original.getClass().newInstance();
- copy.readFields(in);
- return copy;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- return orig;
- }
}
Modified: incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockMapContextWrapper.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockMapContextWrapper.java?rev=1207626&r1=1207625&r2=1207626&view=diff
==============================================================================
--- incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockMapContextWrapper.java (original)
+++ incubator/mrunit/trunk/src/main/java/org/apache/hadoop/mrunit/mapreduce/mock/MockMapContextWrapper.java Mon Nov 28 21:30:29 2011
@@ -48,8 +48,8 @@ public class MockMapContextWrapper<KEYIN
protected static final Log LOG = LogFactory.getLog(MockMapContextWrapper.class);
protected final List<Pair<KEYIN, VALUEIN>> inputs;
protected Pair<KEYIN, VALUEIN> currentKeyValue;
- protected final Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context;
-
+ protected final Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context;
+
public MockMapContextWrapper(List<Pair<KEYIN, VALUEIN>> inputs,
Counters counters, Configuration conf) throws IOException, InterruptedException {
super(counters, conf);
Added: incubator/mrunit/trunk/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestReducerInputValueResuse.java
URL: http://svn.apache.org/viewvc/incubator/mrunit/trunk/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestReducerInputValueResuse.java?rev=1207626&view=auto
==============================================================================
--- incubator/mrunit/trunk/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestReducerInputValueResuse.java (added)
+++ incubator/mrunit/trunk/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestReducerInputValueResuse.java Mon Nov 28 21:30:29 2011
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit.mapreduce;
+
+import java.io.IOException;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+
+public class TestReducerInputValueResuse extends TestCase {
+
+ private class TestReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
+ public LongWritable outputValue = new LongWritable();
+ protected boolean instanceCheckOccured = false;
+ protected boolean instanceCheckFailed = false;
+ public void reduce(Text key, Iterable<LongWritable> vals, Context context)
+ throws IOException, InterruptedException {
+ long sum = 0;
+ LongWritable inputValue = null;
+ for(LongWritable val : vals) {
+ if(inputValue != null) {
+ instanceCheckOccured = true;
+ if(inputValue != val) {
+ instanceCheckFailed = true;
+ }
+ }
+ if(inputValue == null) {
+ inputValue = val;
+ }
+ sum += val.get();
+ }
+ outputValue.set(sum);
+ context.write(key, outputValue);
+ }
+ }
+
+ @Test
+ public void testReduce() throws IOException {
+ TestReducer reducer = new TestReducer();
+ ReduceDriver<Text, LongWritable, Text, LongWritable> driver =
+ new ReduceDriver<Text, LongWritable, Text, LongWritable>();
+ driver.setReducer(reducer);
+ List<LongWritable> values = Lists.newArrayList();
+ values.add(new LongWritable(1));
+ values.add(new LongWritable(1));
+ values.add(new LongWritable(1));
+ values.add(new LongWritable(1));
+ driver.withInput(new Text("foo"), values);
+ driver.withOutput(new Text("foo"), new LongWritable(4));
+ driver.runTest();
+ assertTrue(reducer.instanceCheckOccured);
+ assertFalse(reducer.instanceCheckFailed);
+ }
+
+}
+