You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@crunch.apache.org by jw...@apache.org on 2012/09/13 16:11:58 UTC
[2/2] git commit: CRUNCH-58. Add PObjects to Crunch.
CRUNCH-58. Add PObjects to Crunch.
This commit adds PObjects to Crunch. A PObject encapsulates a singleton
value produced from a distributed computation. The changes in this commit
include:
1. Adding a PObject interface to the Java code base.
2. Adding an abstract class PObjectImpl that implements a PObject backed by
a PCollection. Concrete subclasses implement the PObjectImpl#process method
to transform an iterable obtained from materializing the backing PCollection
into the singleton value encapsulated by the PObject.
3. Adding concrete subclasses of PObjectImpl that a) Use the first element of
the backing PCollection as the PObject value, b) Use a Java collection
containing the elements of the backing PCollection as the PObject value and
c) Use a Java Map containing the mappings defined by Pairs in the backing
PCollection as the PObject value.
4. Modifying min() and max() on PCollection to return PObjects.
5. Adding an asCollection method to PCollection<S> that returns a
PObject<Collection<S>> of the PCollectin's elements.
6. Adding an asMap method to PTable<K, V> that returns a PObject<Map<K,V>>
of the PTable's elements.
7. Adding PObject to the Scala code base and modifying min() and max()
in Scala's PCollection to return PObjects.
Tests have been added for PObjectImpl and its concrete subclasses. Tests
for the new asCollection and asMap methods have also been added. Existing
tests were modified to accomodate changes to min() and max().
Signed-off-by: Josh Wills <jw...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-crunch/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-crunch/commit/2a44d18e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-crunch/tree/2a44d18e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-crunch/diff/2a44d18e
Branch: refs/heads/master
Commit: 2a44d18e8c4c201bc827067245eccabf89e7d34b
Parents: 67142f6
Author: Kiyan Ahmadizadeh <ki...@wibidata.com>
Authored: Thu Sep 6 17:36:27 2012 -0700
Committer: Josh Wills <jw...@apache.org>
Committed: Tue Sep 11 17:55:13 2012 -0700
----------------------------------------------------------------------
.../apache/crunch/scrunch/PageRankClassTest.scala | 4 +-
.../org/apache/crunch/scrunch/PageRankTest.scala | 4 +-
.../org/apache/crunch/scrunch/PCollection.scala | 4 +-
.../scala/org/apache/crunch/scrunch/PObject.scala | 72 ++++++++++
.../org/apache/crunch/CollectionPObjectIT.java | 108 +++++++++++++++
.../org/apache/crunch/FirstElementPObjectIT.java | 72 ++++++++++
.../it/java/org/apache/crunch/MapPObjectIT.java | 105 ++++++++++++++
.../src/it/java/org/apache/crunch/PObjectsIT.java | 108 +++++++++++++++
.../src/it/java/org/apache/crunch/PageRankIT.java | 7 +-
.../it/java/org/apache/crunch/lib/AggregateIT.java | 5 +-
.../main/java/org/apache/crunch/PCollection.java | 34 +++--
.../src/main/java/org/apache/crunch/PObject.java | 55 ++++++++
crunch/src/main/java/org/apache/crunch/PTable.java | 24 +++-
.../crunch/impl/mem/collect/MemCollection.java | 12 ++-
.../apache/crunch/impl/mem/collect/MemTable.java | 8 +
.../crunch/impl/mr/collect/PCollectionImpl.java | 13 ++-
.../apache/crunch/impl/mr/collect/PTableBase.java | 8 +
.../main/java/org/apache/crunch/lib/Aggregate.java | 14 ++-
.../materialize/pobject/CollectionPObject.java | 55 ++++++++
.../materialize/pobject/FirstElementPObject.java | 50 +++++++
.../crunch/materialize/pobject/MapPObject.java | 62 ++++++++
.../crunch/materialize/pobject/PObjectImpl.java | 104 ++++++++++++++
22 files changed, 889 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankClassTest.scala
----------------------------------------------------------------------
diff --git a/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankClassTest.scala b/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankClassTest.scala
index 317e662..847d65b 100644
--- a/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankClassTest.scala
+++ b/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankClassTest.scala
@@ -99,7 +99,7 @@ class PageRankClassTest extends CrunchTestSupport with JUnitSuite {
var delta = 1.0f
while (delta > 0.01f) {
prev = update(prev, 0.5f)
- delta = prev.values.map(_.delta).max.materialize.head
+ delta = prev.values.map(_.delta).max.value()
}
assertEquals(0.0048, delta, 0.001)
pipeline.done
@@ -111,7 +111,7 @@ class PageRankClassTest extends CrunchTestSupport with JUnitSuite {
var delta = 1.0f
while (delta > 0.01f) {
prev = fastUpdate(prev, 0.5f)
- delta = prev.values.map(_.delta).max.materialize.head
+ delta = prev.values.map(_.delta).max.value()
}
assertEquals(0.0048, delta, 0.001)
pipeline.done
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankTest.scala
----------------------------------------------------------------------
diff --git a/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankTest.scala b/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankTest.scala
index 01f2c2a..02e3d82 100644
--- a/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankTest.scala
+++ b/crunch-scrunch/src/it/scala/org/apache/crunch/scrunch/PageRankTest.scala
@@ -86,7 +86,7 @@ class PageRankTest extends CrunchTestSupport with JUnitSuite {
var delta = 1.0f
while (delta > 0.01f) {
prev = update(prev, 0.5f)
- delta = prev.map((k, v) => math.abs(v._1 - v._2)).max.materialize.head
+ delta = prev.map((k, v) => math.abs(v._1 - v._2)).max.value()
}
assertEquals(0.0048, delta, 0.001)
pipeline.done
@@ -97,7 +97,7 @@ class PageRankTest extends CrunchTestSupport with JUnitSuite {
var delta = 1.0f
while (delta > 0.01f) {
prev = fastUpdate(prev, 0.5f)
- delta = prev.map((k, v) => math.abs(v._1 - v._2)).max.materialize.head
+ delta = prev.map((k, v) => math.abs(v._1 - v._2)).max.value()
}
assertEquals(0.0048, delta, 0.001)
pipeline.done
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PCollection.scala
----------------------------------------------------------------------
diff --git a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PCollection.scala b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PCollection.scala
index c99da60..17a8c07 100644
--- a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PCollection.scala
+++ b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PCollection.scala
@@ -72,7 +72,9 @@ class PCollection[S](val native: JCollection[S]) extends PCollectionLike[S, PCol
count.mapValues(_.longValue())
}
- def max() = wrap(Aggregate.max(native))
+ def max() = PObject(Aggregate.max(native))
+
+ def min() = PObject(Aggregate.min(native))
def pType = native.getPType()
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PObject.scala
----------------------------------------------------------------------
diff --git a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PObject.scala b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PObject.scala
new file mode 100644
index 0000000..5d3ddb3
--- /dev/null
+++ b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PObject.scala
@@ -0,0 +1,72 @@
+/**
+ * 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.crunch.scrunch
+
+import org.apache.crunch.{PObject => JPObject}
+import org.apache.crunch.Target
+
+/**
+ * Represents a singleton value that results from a distributed computation.
+ *
+ * @param native The Java PObject that backs this Scala PObject.
+ * @tparam T The type of value encapsulated by this PObject.
+ */
+class PObject[T] private (private val native: JPObject[T]) {
+ /**
+ * Writes the contents of this PObject to the specified Target, using the storage format
+ * specified by the target.
+ *
+ * @param target
+ * @return This PObject.
+ */
+ def write(target: Target): PObject[T] = PObject(native.write(target))
+
+ /**
+ * Gets the value associated with this PObject. Calling this method will trigger
+ * whatever computation is necessary to obtain the value and block until that computation
+ * succeeds.
+ *
+ * @return The value associated with this PObject.
+ */
+ def value(): T = native.getValue()
+
+ /**
+ * @return A shorthand name for this PObject.
+ */
+ def name: String = native.getName()
+
+ /**
+ * @return The Pipeline associated with this PObject.
+ */
+ def pipeline: Pipeline = new Pipeline(native.getPipeline())
+}
+
+/**
+ * The companion object for PObject that provides factory methods for creating PObjects.
+ */
+protected[scrunch] object PObject {
+
+ /**
+ * Creates a new Scala PObject from a Java PObject.
+ *
+ * @param native The Java PObject that will back this Scala PObject.
+ * @tparam T The type of value encapsulated by the PObject.
+ * @return A Scala PObject backed by the Java PObject specified.
+ */
+ def apply[T](native: JPObject[T]): PObject[T] = new PObject[T](native)
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/CollectionPObjectIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/CollectionPObjectIT.java b/crunch/src/it/java/org/apache/crunch/CollectionPObjectIT.java
new file mode 100644
index 0000000..f697a38
--- /dev/null
+++ b/crunch/src/it/java/org/apache/crunch/CollectionPObjectIT.java
@@ -0,0 +1,108 @@
+/**
+ * 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.crunch;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.lang.Integer;
+import java.lang.Iterable;
+import java.lang.String;
+import java.util.Collection;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
+import org.apache.crunch.Pipeline;
+import org.apache.crunch.impl.mem.MemPipeline;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.materialize.pobject.CollectionPObject;
+import org.apache.crunch.materialize.pobject.PObjectImpl;
+import org.apache.crunch.test.TemporaryPath;
+import org.apache.crunch.test.TemporaryPaths;
+import org.apache.crunch.types.PTypeFamily;
+import org.apache.crunch.types.avro.AvroTypeFamily;
+import org.apache.crunch.types.writable.WritableTypeFamily;
+import org.junit.Rule;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+@SuppressWarnings("serial")
+public class CollectionPObjectIT {
+
+ private static final int LINES_IN_SHAKES = 3667;
+
+ private static final String FIRST_SHAKESPEAR_LINE =
+ "***The Project Gutenberg's Etext of Shakespeare's First Folio***";
+
+ private static final String LAST_SHAKESPEAR_LINE =
+ "FINIS. THE TRAGEDIE OF MACBETH.";
+
+ @Rule
+ public TemporaryPath tmpDir = TemporaryPaths.create();
+
+ @Test
+ public void testPObjectMRPipeline() throws IOException {
+ runPObject(new MRPipeline(CollectionPObjectIT.class, tmpDir.getDefaultConfiguration()));
+ }
+
+ @Test
+ public void testAsCollectionMRPipeline() throws IOException {
+ runAsCollection(new MRPipeline(CollectionPObjectIT.class, tmpDir.getDefaultConfiguration()));
+ }
+
+ @Test
+ public void testPObjectMemPipeline() throws IOException {
+ runPObject(MemPipeline.getInstance());
+ }
+
+ @Test
+ public void testAsCollectionMemPipeline() throws IOException {
+ runAsCollection(MemPipeline.getInstance());
+ }
+
+ private PCollection<String> getPCollection(Pipeline pipeline) throws IOException {
+ String shakesInputPath = tmpDir.copyResourceFileName("shakes.txt");
+ PCollection<String> shakespeare = pipeline.readTextFile(shakesInputPath);
+ return shakespeare;
+ }
+
+ private void verifyLines(String[] lines) {
+ assertEquals("Not enough lines in Shakespeare.", LINES_IN_SHAKES, lines.length);
+ assertEquals("First line in Shakespeare is wrong.", FIRST_SHAKESPEAR_LINE, lines[0]);
+ assertEquals("Last line in Shakespeare is wrong.", LAST_SHAKESPEAR_LINE,
+ lines[lines.length - 1]);
+ }
+
+ public void runPObject(Pipeline pipeline) throws IOException {
+ PCollection<String> shakespeare = getPCollection(pipeline);
+ PObject<Collection<String>> linesP = new CollectionPObject<String>(shakespeare);
+ String[] lines = new String[LINES_IN_SHAKES];
+ lines = linesP.getValue().toArray(lines);
+ verifyLines(lines);
+ }
+
+ public void runAsCollection(Pipeline pipeline) throws IOException {
+ PCollection<String> shakespeare = getPCollection(pipeline);
+ String[] lines = new String[LINES_IN_SHAKES];
+ lines = shakespeare.asCollection().getValue().toArray(lines);
+ verifyLines(lines);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/FirstElementPObjectIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/FirstElementPObjectIT.java b/crunch/src/it/java/org/apache/crunch/FirstElementPObjectIT.java
new file mode 100644
index 0000000..a47b1a3
--- /dev/null
+++ b/crunch/src/it/java/org/apache/crunch/FirstElementPObjectIT.java
@@ -0,0 +1,72 @@
+/**
+ * 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.crunch;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.lang.Integer;
+import java.lang.Iterable;
+import java.lang.String;
+import java.util.Collection;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
+import org.apache.crunch.impl.mem.MemPipeline;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.materialize.pobject.FirstElementPObject;
+import org.apache.crunch.materialize.pobject.PObjectImpl;
+import org.apache.crunch.test.TemporaryPath;
+import org.apache.crunch.test.TemporaryPaths;
+import org.apache.crunch.types.PTypeFamily;
+import org.apache.crunch.types.avro.AvroTypeFamily;
+import org.apache.crunch.types.writable.WritableTypeFamily;
+import org.junit.Rule;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+@SuppressWarnings("serial")
+public class FirstElementPObjectIT {
+
+ private static final String FIRST_SHAKESPEAR_LINE =
+ "***The Project Gutenberg's Etext of Shakespeare's First Folio***";
+
+ @Rule
+ public TemporaryPath tmpDir = TemporaryPaths.create();
+
+ @Test
+ public void testMRPipeline() throws IOException {
+ run(new MRPipeline(FirstElementPObjectIT.class, tmpDir.getDefaultConfiguration()));
+ }
+
+ @Test
+ public void testInMemoryPipeline() throws IOException {
+ run(MemPipeline.getInstance());
+ }
+
+ public void run(Pipeline pipeline) throws IOException {
+ String shakesInputPath = tmpDir.copyResourceFileName("shakes.txt");
+ PCollection<String> shakespeare = pipeline.readTextFile(shakesInputPath);
+ PObject<String> firstLine = new FirstElementPObject<String>(shakespeare);
+ String first = firstLine.getValue();
+ assertEquals("First line in Shakespeare is wrong.", FIRST_SHAKESPEAR_LINE, first);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/MapPObjectIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/MapPObjectIT.java b/crunch/src/it/java/org/apache/crunch/MapPObjectIT.java
new file mode 100644
index 0000000..832b35c
--- /dev/null
+++ b/crunch/src/it/java/org/apache/crunch/MapPObjectIT.java
@@ -0,0 +1,105 @@
+/**
+ * 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.crunch;
+
+import static junit.framework.Assert.assertTrue;
+
+import java.io.IOException;
+import java.lang.Integer;
+import java.util.Map;
+
+import org.apache.crunch.PObject;
+import org.apache.crunch.PTable;
+import org.apache.crunch.impl.mem.MemPipeline;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.materialize.pobject.MapPObject;
+import org.apache.crunch.test.TemporaryPath;
+import org.apache.crunch.test.TemporaryPaths;
+import org.apache.crunch.types.PTypeFamily;
+import org.junit.Rule;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+
+public class MapPObjectIT {
+
+ static final ImmutableList<Pair<Integer, String>> kvPairs = ImmutableList.of(Pair.of(0, "a"), Pair.of(1, "b"),
+ Pair.of(2, "c"), Pair.of(3, "e"));
+
+ public void assertMatches(Map<Integer, String> m) {
+ for (Integer k : m.keySet()) {
+ System.out.println(k + " " + kvPairs.get(k).second() + " " + m.get(k));
+ assertTrue(kvPairs.get(k).second().equals(m.get(k)));
+ }
+ }
+
+ private static class Set1Mapper extends MapFn<String, Pair<Integer, String>> {
+ @Override
+ public Pair<Integer, String> map(String input) {
+
+ int k = -1;
+ if (input.equals("a"))
+ k = 0;
+ else if (input.equals("b"))
+ k = 1;
+ else if (input.equals("c"))
+ k = 2;
+ else if (input.equals("e"))
+ k = 3;
+ return Pair.of(k, input);
+ }
+ }
+ @Rule
+ public TemporaryPath tmpDir = TemporaryPaths.create();
+
+ @Test
+ public void testMemMapPObject() {
+ PTable<Integer, String> table = MemPipeline.tableOf(kvPairs);
+ PObject<Map<Integer, String>> map = new MapPObject<Integer, String>(table);
+ assertMatches(map.getValue());
+ }
+
+ @Test
+ public void testMemAsMap() {
+ PTable<Integer, String> table = MemPipeline.tableOf(kvPairs);
+ assertMatches(table.asMap().getValue());
+ }
+
+ private PTable<Integer, String> getMRPTable() throws IOException {
+ Pipeline p = new MRPipeline(MaterializeToMapIT.class, tmpDir.getDefaultConfiguration());
+ String inputFile = tmpDir.copyResourceFileName("set1.txt");
+ PCollection<String> c = p.readTextFile(inputFile);
+ PTypeFamily tf = c.getTypeFamily();
+ PTable<Integer, String> table = c.parallelDo(new Set1Mapper(), tf.tableOf(tf.ints(),
+ tf.strings()));
+ return table;
+ }
+
+ @Test
+ public void testMRMapPObject() throws IOException {
+ PTable<Integer, String> table = getMRPTable();
+ PObject<Map<Integer, String>> map = new MapPObject<Integer, String>(table);
+ assertMatches(map.getValue());
+ }
+
+ @Test
+ public void testMRAsMap() throws IOException {
+ PTable<Integer, String> table = getMRPTable();
+ assertMatches(table.asMap().getValue());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/PObjectsIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/PObjectsIT.java b/crunch/src/it/java/org/apache/crunch/PObjectsIT.java
new file mode 100644
index 0000000..a3810dc
--- /dev/null
+++ b/crunch/src/it/java/org/apache/crunch/PObjectsIT.java
@@ -0,0 +1,108 @@
+/**
+ * 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.crunch;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.lang.Integer;
+import java.lang.Iterable;
+import java.lang.String;
+import java.util.Collection;
+import java.util.Iterator;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
+import org.apache.crunch.impl.mem.MemPipeline;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.materialize.pobject.PObjectImpl;
+import org.apache.crunch.test.TemporaryPath;
+import org.apache.crunch.test.TemporaryPaths;
+import org.apache.crunch.types.PTypeFamily;
+import org.apache.crunch.types.avro.AvroTypeFamily;
+import org.apache.crunch.types.writable.WritableTypeFamily;
+import org.junit.Rule;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+@SuppressWarnings("serial")
+public class PObjectsIT {
+
+ private static final Integer LINES_IN_SHAKES = 3667;
+
+ @Rule
+ public TemporaryPath tmpDir = TemporaryPaths.create();
+
+ /**
+ * A mock PObject that should map PCollections of strings to an integer count of the number of
+ * elements in the underlying PCollection.
+ */
+ public static class MockPObjectImpl extends PObjectImpl<String, Integer> {
+ private int numProcessCalls;
+
+ public MockPObjectImpl(PCollection<String> collect) {
+ super(collect);
+ numProcessCalls = 0;
+ }
+
+ @Override
+ public Integer process(Iterable<String> input) {
+ numProcessCalls++;
+ int i = 0;
+ Iterator<String> itr = input.iterator();
+ while (itr.hasNext()) {
+ i++;
+ itr.next();
+ }
+ return i;
+ }
+
+ public int getNumProcessCalls() {
+ return numProcessCalls;
+ }
+ }
+
+ @Test
+ public void testMRPipeline() throws IOException {
+ run(new MRPipeline(PObjectsIT.class, tmpDir.getDefaultConfiguration()));
+ }
+
+ @Test
+ public void testInMemoryPipeline() throws IOException {
+ run(MemPipeline.getInstance());
+ }
+
+ public void run(Pipeline pipeline) throws IOException {
+ String shakesInputPath = tmpDir.copyResourceFileName("shakes.txt");
+ PCollection<String> shakespeare = pipeline.readTextFile(shakesInputPath);
+ MockPObjectImpl lineCount = new MockPObjectImpl(shakespeare);
+ // Get the line count once and verify it's correctness.
+ assertEquals("Incorrect number of lines counted from PCollection.", LINES_IN_SHAKES,
+ lineCount.getValue());
+ // And do it again.
+ assertEquals("Incorrect number of lines counted from PCollection.", LINES_IN_SHAKES,
+ lineCount.getValue());
+ // Make sure process was called only once because the PObject's value was cached after the
+ // first call.
+ assertEquals("Process on PObject not called exactly 1 times.", 1,
+ lineCount.getNumProcessCalls());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/PageRankIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/PageRankIT.java b/crunch/src/it/java/org/apache/crunch/PageRankIT.java
index 761129a..e45bbc7 100644
--- a/crunch/src/it/java/org/apache/crunch/PageRankIT.java
+++ b/crunch/src/it/java/org/apache/crunch/PageRankIT.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertEquals;
import java.util.Collection;
import java.util.List;
+import org.apache.crunch.PObject;
import org.apache.crunch.impl.mem.MemPipeline;
import org.apache.crunch.impl.mr.MRPipeline;
import org.apache.crunch.lib.Aggregate;
@@ -71,7 +72,7 @@ public class PageRankIT {
@Rule
public TemporaryPath tmpDir = TemporaryPaths.create();
-
+
@Test
public void testAvroReflect() throws Exception {
PTypeFamily tf = AvroTypeFamily.getInstance();
@@ -155,13 +156,13 @@ public class PageRankIT {
while (delta > 0.01) {
scores = pageRank(scores, 0.5f);
scores.materialize().iterator(); // force the write
- delta = Iterables.getFirst(Aggregate.max(scores.parallelDo(new MapFn<Pair<String, PageRankData>, Float>() {
+ delta = Aggregate.max(scores.parallelDo(new MapFn<Pair<String, PageRankData>, Float>() {
@Override
public Float map(Pair<String, PageRankData> input) {
PageRankData prd = input.second();
return Math.abs(prd.score - prd.lastScore);
}
- }, ptf.floats())).materialize(), null);
+ }, ptf.floats())).getValue();
}
assertEquals(0.0048, delta, 0.001);
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/it/java/org/apache/crunch/lib/AggregateIT.java
----------------------------------------------------------------------
diff --git a/crunch/src/it/java/org/apache/crunch/lib/AggregateIT.java b/crunch/src/it/java/org/apache/crunch/lib/AggregateIT.java
index 5d40772..88596aa 100644
--- a/crunch/src/it/java/org/apache/crunch/lib/AggregateIT.java
+++ b/crunch/src/it/java/org/apache/crunch/lib/AggregateIT.java
@@ -28,6 +28,7 @@ import java.util.Map;
import org.apache.crunch.MapFn;
import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Pipeline;
@@ -91,8 +92,8 @@ public class AggregateIT {
return -input;
}
}, family.ints());
- Integer maxLengths = Iterables.getFirst(Aggregate.max(lengths).materialize(), null);
- Integer minLengths = Iterables.getFirst(Aggregate.min(negLengths).materialize(), null);
+ Integer maxLengths = Aggregate.max(lengths).getValue();
+ Integer minLengths = Aggregate.min(negLengths).getValue();
assertTrue(maxLengths != null);
assertTrue(minLengths != null);
assertEquals(maxLengths.intValue(), -minLengths.intValue());
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/PCollection.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/PCollection.java b/crunch/src/main/java/org/apache/crunch/PCollection.java
index 4a3f043..8e73159 100644
--- a/crunch/src/main/java/org/apache/crunch/PCollection.java
+++ b/crunch/src/main/java/org/apache/crunch/PCollection.java
@@ -17,6 +17,8 @@
*/
package org.apache.crunch;
+import java.util.Collection;
+
import org.apache.crunch.types.PTableType;
import org.apache.crunch.types.PType;
import org.apache.crunch.types.PTypeFamily;
@@ -24,7 +26,7 @@ import org.apache.crunch.types.PTypeFamily;
/**
* A representation of an immutable, distributed collection of elements that is
* the fundamental target of computations in Crunch.
- *
+ *
*/
public interface PCollection<S> {
/**
@@ -41,7 +43,7 @@ public interface PCollection<S> {
/**
* Applies the given doFn to the elements of this {@code PCollection} and
* returns a new {@code PCollection} that is the output of this processing.
- *
+ *
* @param doFn
* The {@code DoFn} to apply
* @param type
@@ -53,7 +55,7 @@ public interface PCollection<S> {
/**
* Applies the given doFn to the elements of this {@code PCollection} and
* returns a new {@code PCollection} that is the output of this processing.
- *
+ *
* @param name
* An identifier for this processing step, useful for debugging
* @param doFn
@@ -67,7 +69,7 @@ public interface PCollection<S> {
/**
* Similar to the other {@code parallelDo} instance, but returns a
* {@code PTable} instance instead of a {@code PCollection}.
- *
+ *
* @param doFn
* The {@code DoFn} to apply
* @param type
@@ -79,7 +81,7 @@ public interface PCollection<S> {
/**
* Similar to the other {@code parallelDo} instance, but returns a
* {@code PTable} instance instead of a {@code PCollection}.
- *
+ *
* @param name
* An identifier for this processing step
* @param doFn
@@ -93,7 +95,7 @@ public interface PCollection<S> {
/**
* Write the contents of this {@code PCollection} to the given {@code Target},
* using the storage format specified by the target.
- *
+ *
* @param target
* The target to write to
*/
@@ -106,6 +108,12 @@ public interface PCollection<S> {
Iterable<S> materialize();
/**
+ * @return A {@code PObject} encapsulating an in-memory {@link Collection} containing the values
+ * of this {@code PCollection}.
+ */
+ PObject<Collection<S>> asCollection();
+
+ /**
* Returns the {@code PType} of this {@code PCollection}.
*/
PType<S> getPType();
@@ -135,7 +143,7 @@ public interface PCollection<S> {
/**
* Apply the given filter function to this instance and return the resulting
* {@code PCollection}.
- *
+ *
* @param name
* An identifier for this processing step
* @param filterFn
@@ -152,7 +160,7 @@ public interface PCollection<S> {
/**
* Apply the given map function to each element of this instance in order to
* create a {@code PTable}.
- *
+ *
* @param name
* An identifier for this processing step
* @param extractKeyFn
@@ -173,16 +181,14 @@ public interface PCollection<S> {
PTable<S, Long> count();
/**
- * Returns a {@code PCollection} made up of only the maximum element of this
- * instance.
+ * Returns a {@code PObject} of the maximum element of this instance.
*/
- PCollection<S> max();
+ PObject<S> max();
/**
- * Returns a {@code PCollection} made up of only the minimum element of this
- * instance.
+ * Returns a {@code PObject} of the minimum element of this instance.
*/
- PCollection<S> min();
+ PObject<S> min();
/**
* Randomly sample items from this PCollection instance with the given
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/PObject.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/PObject.java b/crunch/src/main/java/org/apache/crunch/PObject.java
new file mode 100644
index 0000000..2a2c1f3
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/PObject.java
@@ -0,0 +1,55 @@
+/**
+ * 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.crunch;
+
+/**
+ * A {@code PObject} represents a singleton object value that results from a distributed
+ * computation. Computation producing the value is deferred until
+ * {@link org.apache.crunch.PObject#getValue()} is called.
+ *
+ * @param <T> The type of value encapsulated by this {@code PObject}.
+ */
+public interface PObject<T> {
+
+ /**
+ * Returns the {@code Pipeline} associated with this {@code PObject}
+ */
+ Pipeline getPipeline();
+
+ /**
+ * Write the contents of this {@code PObject} to the given {@code Target},
+ * using the storage format specified by the target.
+ *
+ * @param target The target to write to
+ */
+ PObject<T> write(Target target);
+
+ /**
+ * Returns a shorthand name for this PObject.
+ */
+ String getName();
+
+ /**
+ * Gets the value associated with this {@code PObject}. Calling this method will trigger
+ * whatever computation is necessary to obtain the value and block until that computation
+ * succeeds.
+ *
+ * @return The value associated with this {@code PObject}.
+ */
+ T getValue();
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/PTable.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/PTable.java b/crunch/src/main/java/org/apache/crunch/PTable.java
index cd10e46..e827603 100644
--- a/crunch/src/main/java/org/apache/crunch/PTable.java
+++ b/crunch/src/main/java/org/apache/crunch/PTable.java
@@ -26,7 +26,7 @@ import org.apache.crunch.types.PType;
/**
* A sub-interface of {@code PCollection} that represents an immutable,
* distributed multi-map of keys and values.
- *
+ *
*/
public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
@@ -37,7 +37,7 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
* Performs a grouping operation on the keys of this table.
- *
+ *
* @return a {@code PGroupedTable} instance that represents the grouping
*/
PGroupedTable<K, V> groupByKey();
@@ -45,7 +45,7 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
* Performs a grouping operation on the keys of this table, using the given
* number of partitions.
- *
+ *
* @param numPartitions
* The number of partitions for the data.
* @return a {@code PGroupedTable} instance that represents this grouping
@@ -55,7 +55,7 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
* Performs a grouping operation on the keys of this table, using the
* additional {@code GroupingOptions} to control how the grouping is executed.
- *
+ *
* @param options
* The grouping options to use
* @return a {@code PGroupedTable} instance that represents the grouping
@@ -91,7 +91,7 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
* Returns a PTable made up of the pairs in this PTable with the largest value
* field.
- *
+ *
* @param count
* The number of pairs to return
*/
@@ -100,7 +100,7 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
/**
* Returns a PTable made up of the pairs in this PTable with the smallest
* value field.
- *
+ *
* @param count
* The number of pairs to return
*/
@@ -136,4 +136,16 @@ public interface PTable<K, V> extends PCollection<Pair<K, V>> {
*/
Map<K, V> materializeToMap();
+ /**
+ * Returns a {@link PObject} encapsulating a {@link Map} made up of the keys and values in this
+ * {@code PTable}.
+ * <p><b>Note:</b>The contents of the returned map may not be exactly the same as this PTable,
+ * as a PTable is a multi-map (i.e. can contain multiple values for a single key).
+ * </p>
+ *
+ * @return The {@code PObject} encapsulating a {@code Map} made up of the keys and values in
+ * this {@code PTable}.
+ */
+ PObject<Map<K, V>> asMap();
+
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemCollection.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemCollection.java b/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemCollection.java
index 76c0c7e..9e5b6f2 100644
--- a/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemCollection.java
+++ b/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemCollection.java
@@ -23,6 +23,7 @@ import org.apache.crunch.DoFn;
import org.apache.crunch.FilterFn;
import org.apache.crunch.MapFn;
import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Pipeline;
@@ -32,6 +33,7 @@ import org.apache.crunch.impl.mem.MemPipeline;
import org.apache.crunch.lib.Aggregate;
import org.apache.crunch.lib.Sample;
import org.apache.crunch.lib.Sort;
+import org.apache.crunch.materialize.pobject.CollectionPObject;
import org.apache.crunch.test.InMemoryEmitter;
import org.apache.crunch.types.PTableType;
import org.apache.crunch.types.PType;
@@ -120,6 +122,12 @@ public class MemCollection<S> implements PCollection<S> {
return collect;
}
+ /** {@inheritDoc} */
+ @Override
+ public PObject<Collection<S>> asCollection() {
+ return new CollectionPObject<S>(this);
+ }
+
public Collection<S> getCollection() {
return collect;
}
@@ -168,12 +176,12 @@ public class MemCollection<S> implements PCollection<S> {
}
@Override
- public PCollection<S> max() {
+ public PObject<S> max() {
return Aggregate.max(this);
}
@Override
- public PCollection<S> min() {
+ public PObject<S> min() {
return Aggregate.min(this);
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemTable.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemTable.java b/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemTable.java
index ff2b2d7..56dc69a 100644
--- a/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemTable.java
+++ b/crunch/src/main/java/org/apache/crunch/impl/mem/collect/MemTable.java
@@ -24,6 +24,7 @@ import java.util.Map;
import org.apache.crunch.GroupingOptions;
import org.apache.crunch.PCollection;
import org.apache.crunch.PGroupedTable;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Target;
@@ -32,6 +33,7 @@ import org.apache.crunch.lib.Cogroup;
import org.apache.crunch.lib.Join;
import org.apache.crunch.lib.PTables;
import org.apache.crunch.materialize.MaterializableMap;
+import org.apache.crunch.materialize.pobject.MapPObject;
import org.apache.crunch.types.PTableType;
import org.apache.crunch.types.PType;
@@ -144,4 +146,10 @@ public class MemTable<K, V> extends MemCollection<Pair<K, V>> implements PTable<
return new MaterializableMap<K, V>(this.materialize());
}
+ /** {@inheritDoc} */
+ @Override
+ public PObject<Map<K, V>> asMap() {
+ return new MapPObject<K, V>(this);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PCollectionImpl.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PCollectionImpl.java b/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PCollectionImpl.java
index 962eeb8..d88da42 100644
--- a/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PCollectionImpl.java
+++ b/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PCollectionImpl.java
@@ -17,6 +17,7 @@
*/
package org.apache.crunch.impl.mr.collect;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -26,6 +27,7 @@ import org.apache.crunch.DoFn;
import org.apache.crunch.FilterFn;
import org.apache.crunch.MapFn;
import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Pipeline;
@@ -37,6 +39,7 @@ import org.apache.crunch.impl.mr.plan.DoNode;
import org.apache.crunch.lib.Aggregate;
import org.apache.crunch.lib.Sample;
import org.apache.crunch.lib.Sort;
+import org.apache.crunch.materialize.pobject.CollectionPObject;
import org.apache.crunch.types.PTableType;
import org.apache.crunch.types.PType;
import org.apache.crunch.types.PTypeFamily;
@@ -112,6 +115,12 @@ public abstract class PCollectionImpl<S> implements PCollection<S> {
return getPipeline().materialize(this);
}
+ /** {@inheritDoc} */
+ @Override
+ public PObject<Collection<S>> asCollection() {
+ return new CollectionPObject<S>(this);
+ }
+
public SourceTarget<S> getMaterializedAt() {
return materializedAt;
}
@@ -151,12 +160,12 @@ public abstract class PCollectionImpl<S> implements PCollection<S> {
}
@Override
- public PCollection<S> max() {
+ public PObject<S> max() {
return Aggregate.max(this);
}
@Override
- public PCollection<S> min() {
+ public PObject<S> min() {
return Aggregate.min(this);
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PTableBase.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PTableBase.java b/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PTableBase.java
index 4126f74..cd77606 100644
--- a/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PTableBase.java
+++ b/crunch/src/main/java/org/apache/crunch/impl/mr/collect/PTableBase.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.crunch.GroupingOptions;
import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Target;
@@ -31,6 +32,7 @@ import org.apache.crunch.lib.Cogroup;
import org.apache.crunch.lib.Join;
import org.apache.crunch.lib.PTables;
import org.apache.crunch.materialize.MaterializableMap;
+import org.apache.crunch.materialize.pobject.MapPObject;
import org.apache.crunch.types.PType;
import com.google.common.collect.Lists;
@@ -115,8 +117,14 @@ public abstract class PTableBase<K, V> extends PCollectionImpl<Pair<K, V>> imple
/**
* Returns a Map<K, V> made up of the keys and values in this PTable.
*/
+ @Override
public Map<K, V> materializeToMap() {
return new MaterializableMap<K, V>(this.materialize());
}
+ /** {@inheritDoc} */
+ @Override
+ public PObject<Map<K, V>> asMap() {
+ return new MapPObject<K, V>(this);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/lib/Aggregate.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/lib/Aggregate.java b/crunch/src/main/java/org/apache/crunch/lib/Aggregate.java
index 360d40b..c2c56c0 100644
--- a/crunch/src/main/java/org/apache/crunch/lib/Aggregate.java
+++ b/crunch/src/main/java/org/apache/crunch/lib/Aggregate.java
@@ -28,9 +28,11 @@ import org.apache.crunch.DoFn;
import org.apache.crunch.Emitter;
import org.apache.crunch.MapFn;
import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.fn.MapValuesFn;
+import org.apache.crunch.materialize.pobject.FirstElementPObject;
import org.apache.crunch.types.PTableType;
import org.apache.crunch.types.PType;
import org.apache.crunch.types.PTypeFamily;
@@ -40,7 +42,7 @@ import com.google.common.collect.Lists;
/**
* Methods for performing various types of aggregations over {@link PCollection}
* instances.
- *
+ *
*/
public class Aggregate {
@@ -146,14 +148,14 @@ public class Aggregate {
/**
* Returns the largest numerical element from the input collection.
*/
- public static <S> PCollection<S> max(PCollection<S> collect) {
+ public static <S> PObject<S> max(PCollection<S> collect) {
Class<S> clazz = collect.getPType().getTypeClass();
if (!clazz.isPrimitive() && !Comparable.class.isAssignableFrom(clazz)) {
throw new IllegalArgumentException("Can only get max for Comparable elements, not for: "
+ collect.getPType().getTypeClass());
}
PTypeFamily tf = collect.getTypeFamily();
- return PTables.values(collect.parallelDo("max", new DoFn<S, Pair<Boolean, S>>() {
+ PCollection<S> maxCollect = PTables.values(collect.parallelDo("max", new DoFn<S, Pair<Boolean, S>>() {
private transient S max = null;
public void process(S input, Emitter<Pair<Boolean, S>> emitter) {
@@ -178,19 +180,20 @@ public class Aggregate {
emitter.emit(Pair.of(input.first(), max));
}
}));
+ return new FirstElementPObject<S>(maxCollect);
}
/**
* Returns the smallest numerical element from the input collection.
*/
- public static <S> PCollection<S> min(PCollection<S> collect) {
+ public static <S> PObject<S> min(PCollection<S> collect) {
Class<S> clazz = collect.getPType().getTypeClass();
if (!clazz.isPrimitive() && !Comparable.class.isAssignableFrom(clazz)) {
throw new IllegalArgumentException("Can only get min for Comparable elements, not for: "
+ collect.getPType().getTypeClass());
}
PTypeFamily tf = collect.getTypeFamily();
- return PTables.values(collect.parallelDo("min", new DoFn<S, Pair<Boolean, S>>() {
+ PCollection<S> minCollect = PTables.values(collect.parallelDo("min", new DoFn<S, Pair<Boolean, S>>() {
private transient S min = null;
public void process(S input, Emitter<Pair<Boolean, S>> emitter) {
@@ -215,6 +218,7 @@ public class Aggregate {
emitter.emit(Pair.of(input.first(), min));
}
}));
+ return new FirstElementPObject<S>(minCollect);
}
public static <K, V> PTable<K, Collection<V>> collectValues(PTable<K, V> collect) {
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/materialize/pobject/CollectionPObject.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/pobject/CollectionPObject.java b/crunch/src/main/java/org/apache/crunch/materialize/pobject/CollectionPObject.java
new file mode 100644
index 0000000..60e64b1
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/pobject/CollectionPObject.java
@@ -0,0 +1,55 @@
+/**
+ * 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.crunch.materialize.pobject;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+
+import org.apache.crunch.PCollection;
+
+/**
+ * A concrete implementation of {@link org.apache.crunch.materialize.pobject.PObjectImpl} whose
+ * value is a Java {@link java.util.Collection} containing the elements of the underlying {@link
+ * PCollection} for this {@link org.apache.crunch.PObject}.
+ *
+ * @param <S> The value type for elements contained in the {@code Collection} value encapsulated
+ * by this {@code PObject}.
+ */
+public class CollectionPObject<S> extends PObjectImpl<S, Collection<S>> {
+
+ /**
+ * Constructs a new instance of this {@code PObject} implementation.
+ *
+ * @param collect The backing {@code PCollection} for this {@code PObject}.
+ */
+ public CollectionPObject(PCollection<S> collect) {
+ super(collect);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Collection<S> process(Iterable<S> input) {
+ Collection<S> target = new ArrayList<S>();
+ Iterator<S> itr = input.iterator();
+ while (itr.hasNext()) {
+ target.add(itr.next());
+ }
+ return target;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/materialize/pobject/FirstElementPObject.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/pobject/FirstElementPObject.java b/crunch/src/main/java/org/apache/crunch/materialize/pobject/FirstElementPObject.java
new file mode 100644
index 0000000..aa5fd9e
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/pobject/FirstElementPObject.java
@@ -0,0 +1,50 @@
+/**
+ * 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.crunch.materialize.pobject;
+
+import java.util.Iterator;
+
+import org.apache.crunch.PCollection;
+
+/**
+ * A concrete implementation of {@link PObjectImpl} that uses the first element in the backing
+ * {@link PCollection} as the {@link org.apache.crunch.PObject} value.
+ *
+ * @param <T> The value type of this {@code PObject}.
+ */
+public class FirstElementPObject<T> extends PObjectImpl<T, T> {
+
+ /**
+ * Constructs a new instance of this {@code PObject} implementation.
+ *
+ * @param collect The backing {@code PCollection} for this {@code PObject}.
+ */
+ public FirstElementPObject(PCollection<T> collect) {
+ super(collect);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public T process(Iterable<T> input) {
+ Iterator<T> itr = input.iterator();
+ if (itr.hasNext()) {
+ return itr.next();
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/materialize/pobject/MapPObject.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/pobject/MapPObject.java b/crunch/src/main/java/org/apache/crunch/materialize/pobject/MapPObject.java
new file mode 100644
index 0000000..243997f
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/pobject/MapPObject.java
@@ -0,0 +1,62 @@
+/**
+ * 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.crunch.materialize.pobject;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.Pair;
+
+/**
+ * A concrete implementation of {@link PObjectImpl} whose
+ * value is a Java {@link Map}. The underlying {@link PCollection} for this
+ * {@link org.apache.crunch.PObject} must contain {@link Pair}s of values. The
+ * first element of the pair will be used as the map key, while the second element will be used
+ * as the map value. Note that the contents of the underlying {@code PCollection} may not be
+ * reflected in the returned {@code Map}, since a single key may be mapped to several values in
+ * the underlying {@code PCollection}, and only one of those values will appear in the {@code
+ * Map} encapsulated by this {@code PObject}.
+ *
+ * @param <K> The type of keys for the Map.
+ * @param <V> The type of values for the Map.
+ */
+public class MapPObject<K, V> extends PObjectImpl<Pair<K, V>, Map<K, V>> {
+
+ /**
+ * Constructs a new instance of this {@code PObject} implementation.
+ *
+ * @param collect The backing {@code PCollection} for this {@code PObject}.
+ */
+ public MapPObject(PCollection<Pair<K, V>> collect) {
+ super(collect);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Map<K, V> process(Iterable<Pair<K, V>> input) {
+ Map<K, V> target = new HashMap<K, V>();
+ Iterator<Pair<K, V>> itr = input.iterator();
+ while (itr.hasNext()) {
+ Pair<K, V> pair = itr.next();
+ target.put(pair.first(), pair.second());
+ }
+ return target;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/2a44d18e/crunch/src/main/java/org/apache/crunch/materialize/pobject/PObjectImpl.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/pobject/PObjectImpl.java b/crunch/src/main/java/org/apache/crunch/materialize/pobject/PObjectImpl.java
new file mode 100644
index 0000000..20e43bc
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/pobject/PObjectImpl.java
@@ -0,0 +1,104 @@
+/**
+ * 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.crunch.materialize.pobject;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PObject;
+import org.apache.crunch.Pipeline;
+import org.apache.crunch.Target;
+
+/**
+ * An abstract implementation of {@link PObject} that is backed by a {@link PCollection}.
+ * Clients creating a concrete implementation should override the method
+ * {@link PObjectImpl#process(Iterable)}, which transforms the backing PCollection into the
+ * singleton value encapsulated by the PObject. Once this {code PObject}'s value has been
+ * calculated, the value is cached to prevent subsequent materializations of the backing
+ * {@code PCollection}.
+ *
+ * @param <S> The type contained in the underlying PCollection.
+ * @param <T> The type encapsulated by this PObject.
+ */
+public abstract class PObjectImpl<S, T> implements PObject<T> {
+
+ // The underlying PCollection whose contents will be used to generate the value for this
+ // PObject.
+ private PCollection<S> collection;
+
+ // A variable to hold a cached copy of the value of this {@code PObject},
+ // to prevent unnecessary materializations of the backing {@code PCollection}.
+ private T cachedValue;
+
+ // A flag indicating if a value for this {@code PObject} has been cached.
+ private boolean isCached;
+
+ /**
+ * Constructs a new instance of this {@code PObject} implementation.
+ *
+ * @param collect The backing {@code PCollection} for this {@code PObject}.
+ */
+ public PObjectImpl(PCollection<S> collect) {
+ this.collection = collect;
+ this.cachedValue = null;
+ this.isCached = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public final Pipeline getPipeline() {
+ return collection.getPipeline();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public final PObject<T> write(Target target) {
+ collection.write(target);
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public final String getName() {
+ return collection.getName();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String toString() {
+ return collection.toString();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public final T getValue() {
+ if (!isCached) {
+ cachedValue = process(collection.materialize());
+ isCached = true;
+ }
+ return cachedValue;
+ }
+
+ /**
+ * Transforms the provided Iterable, obtained from the backing {@link PCollection},
+ * into the value encapsulated by this {@code PObject}.
+ *
+ * @param input An Iterable whose elements correspond to those of the backing {@code
+ * PCollection}.
+ * @return The value of this {@code PObject}.
+ */
+ protected abstract T process(Iterable<S> input);
+}