You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrunit.apache.org by db...@apache.org on 2012/10/06 12:34:46 UTC
git commit: MRUNIT-136: PipelineMapReduceDriver needs withAll/addAll
methods
Updated Branches:
refs/heads/trunk 3b1f193d7 -> 03b0d58cb
MRUNIT-136: PipelineMapReduceDriver needs withAll/addAll methods
Project: http://git-wip-us.apache.org/repos/asf/mrunit/repo
Commit: http://git-wip-us.apache.org/repos/asf/mrunit/commit/03b0d58c
Tree: http://git-wip-us.apache.org/repos/asf/mrunit/tree/03b0d58c
Diff: http://git-wip-us.apache.org/repos/asf/mrunit/diff/03b0d58c
Branch: refs/heads/trunk
Commit: 03b0d58cb51268bef16ac1bc20dbd5467acf95ba
Parents: 3b1f193
Author: Dave Beech <da...@newsint.co.uk>
Authored: Sat Oct 6 11:31:54 2012 +0100
Committer: Dave Beech <da...@newsint.co.uk>
Committed: Sat Oct 6 11:31:54 2012 +0100
----------------------------------------------------------------------
.../hadoop/mrunit/PipelineMapReduceDriver.java | 24 +++++++++++++++
.../hadoop/mrunit/TestPipelineMapReduceDriver.java | 22 +++++++++++++
2 files changed, 46 insertions(+), 0 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mrunit/blob/03b0d58c/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
index 5c4d9f2..9dd7086 100644
--- a/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
+++ b/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
@@ -169,6 +169,18 @@ public class PipelineMapReduceDriver<K1, V1, K2, V2> extends
}
/**
+ * Adds list of inputs to send to the mapper
+ *
+ * @param inputs
+ * list of (K, V) pairs
+ */
+ public void addAll(final List<Pair<K1, V1>> inputs) {
+ for (Pair<K1, V1> input : inputs) {
+ addInput(input);
+ }
+ }
+
+ /**
* Identical to addInput() but returns self for fluent programming style
*
* @param key
@@ -236,6 +248,18 @@ public class PipelineMapReduceDriver<K1, V1, K2, V2> extends
}
/**
+ * Identical to addAll() but returns self for fluent programming style
+ *
+ * @param inputRecords input key/value pairs
+ * @return this
+ */
+ public PipelineMapReduceDriver<K1, V1, K2, V2> withAll(
+ final List<Pair<K1, V1>> inputRecords) {
+ addAll(inputRecords);
+ return this;
+ }
+
+ /**
* @return the path passed to the mapper InputSplit
*/
public Path getMapInputPath() {
http://git-wip-us.apache.org/repos/asf/mrunit/blob/03b0d58c/src/test/java/org/apache/hadoop/mrunit/TestPipelineMapReduceDriver.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/hadoop/mrunit/TestPipelineMapReduceDriver.java b/src/test/java/org/apache/hadoop/mrunit/TestPipelineMapReduceDriver.java
index 67fc16f..398e584 100644
--- a/src/test/java/org/apache/hadoop/mrunit/TestPipelineMapReduceDriver.java
+++ b/src/test/java/org/apache/hadoop/mrunit/TestPipelineMapReduceDriver.java
@@ -110,6 +110,28 @@ public class TestPipelineMapReduceDriver {
}
@Test
+ public void testSumAtEndWithInputAddAll() throws IOException {
+
+ final List<Pair<Text, LongWritable>> inputs = new ArrayList<Pair<Text, LongWritable>>();
+ inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_A)));
+ inputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(BAR_IN)));
+ inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_B)));
+
+ final PipelineMapReduceDriver<Text, LongWritable, Text, LongWritable> driver
+ = PipelineMapReduceDriver.newPipelineMapReduceDriver();
+ driver
+ .withMapReduce(new IdentityMapper<Text, LongWritable>(),
+ new IdentityReducer<Text, LongWritable>())
+ .withMapReduce(new IdentityMapper<Text, LongWritable>(),
+ new IdentityReducer<Text, LongWritable>())
+ .withMapReduce(new IdentityMapper<Text, LongWritable>(),
+ new LongSumReducer<Text>())
+ .withAll(inputs)
+ .withOutput(new Text("bar"), new LongWritable(BAR_OUT))
+ .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest();
+ }
+
+ @Test
public void testSumInMiddle() throws IOException {
final PipelineMapReduceDriver<Text, LongWritable, Text, LongWritable> driver = PipelineMapReduceDriver.newPipelineMapReduceDriver();
driver