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/07/11 07:14:46 UTC
[8/33] CRUNCH-8: Moving the code into multiple Maven modules.
Contributed by Matthias Friedrich
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/lib/join/MapsideJoin.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/lib/join/MapsideJoin.java b/crunch/src/main/java/org/apache/crunch/lib/join/MapsideJoin.java
new file mode 100644
index 0000000..28a0829
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/lib/join/MapsideJoin.java
@@ -0,0 +1,160 @@
+/**
+ * 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.lib.join;
+
+import java.io.IOException;
+
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.crunch.DoFn;
+import org.apache.crunch.Emitter;
+import org.apache.crunch.PTable;
+import org.apache.crunch.Pair;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.impl.mr.run.CrunchRuntimeException;
+import org.apache.crunch.io.ReadableSourceTarget;
+import org.apache.crunch.io.impl.SourcePathTargetImpl;
+import org.apache.crunch.types.PType;
+import org.apache.crunch.types.PTypeFamily;
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Multimap;
+
+/**
+ * Utility for doing map side joins on a common key between two {@link PTable}s.
+ * <p>
+ * A map side join is an optimized join which doesn't use a reducer; instead,
+ * the right side of the join is loaded into memory and the join is performed in
+ * a mapper. This style of join has the important implication that the output of
+ * the join is not sorted, which is the case with a conventional (reducer-based)
+ * join.
+ * <p>
+ * <b>Note:</b>This utility is only supported when running with a
+ * {@link MRPipeline} as the pipeline.
+ */
+public class MapsideJoin {
+
+ /**
+ * Join two tables using a map side join. The right-side table will be loaded
+ * fully in memory, so this method should only be used if the right side
+ * table's contents can fit in the memory allocated to mappers. The join
+ * performed by this method is an inner join.
+ *
+ * @param left
+ * The left-side table of the join
+ * @param right
+ * The right-side table of the join, whose contents will be fully
+ * read into memory
+ * @return A table keyed on the join key, containing pairs of joined values
+ */
+ public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) {
+
+ if (!(right.getPipeline() instanceof MRPipeline)) {
+ throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context");
+ }
+
+ MRPipeline pipeline = (MRPipeline) right.getPipeline();
+ pipeline.materialize(right);
+
+ // TODO Move necessary logic to MRPipeline so that we can theoretically
+ // optimize his by running the setup of multiple map-side joins concurrently
+ pipeline.run();
+
+ ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline
+ .getMaterializeSourceTarget(right);
+ if (!(readableSourceTarget instanceof SourcePathTargetImpl)) {
+ throw new CrunchRuntimeException("Right-side contents can't be read from a path");
+ }
+
+ // Suppress warnings because we've just checked this cast via instanceof
+ @SuppressWarnings("unchecked")
+ SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget;
+
+ Path path = sourcePathTarget.getPath();
+ DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration());
+
+ MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(),
+ right.getPType());
+ PTypeFamily typeFamily = left.getTypeFamily();
+ return left.parallelDo(
+ "mapjoin",
+ mapJoinDoFn,
+ typeFamily.tableOf(left.getKeyType(),
+ typeFamily.pairs(left.getValueType(), right.getValueType())));
+
+ }
+
+ static class MapsideJoinDoFn<K, U, V> extends DoFn<Pair<K, U>, Pair<K, Pair<U, V>>> {
+
+ private String inputPath;
+ private PType<Pair<K, V>> ptype;
+ private Multimap<K, V> joinMap;
+
+ public MapsideJoinDoFn(String inputPath, PType<Pair<K, V>> ptype) {
+ this.inputPath = inputPath;
+ this.ptype = ptype;
+ }
+
+ private Path getCacheFilePath() {
+ try {
+ for (Path localPath : DistributedCache.getLocalCacheFiles(getConfiguration())) {
+ if (localPath.toString().endsWith(inputPath)) {
+ return localPath.makeQualified(FileSystem.getLocal(getConfiguration()));
+
+ }
+ }
+ } catch (IOException e) {
+ throw new CrunchRuntimeException(e);
+ }
+
+ throw new CrunchRuntimeException("Can't find local cache file for '" + inputPath + "'");
+ }
+
+ @Override
+ public void initialize() {
+ super.initialize();
+
+ ReadableSourceTarget<Pair<K, V>> sourceTarget = (ReadableSourceTarget<Pair<K, V>>) ptype
+ .getDefaultFileSource(getCacheFilePath());
+ Iterable<Pair<K, V>> iterable = null;
+ try {
+ iterable = sourceTarget.read(getConfiguration());
+ } catch (IOException e) {
+ throw new CrunchRuntimeException("Error reading right-side of map side join: ", e);
+ }
+
+ joinMap = ArrayListMultimap.create();
+ for (Pair<K, V> joinPair : iterable) {
+ joinMap.put(joinPair.first(), joinPair.second());
+ }
+ }
+
+ @Override
+ public void process(Pair<K, U> input, Emitter<Pair<K, Pair<U, V>>> emitter) {
+ K key = input.first();
+ U value = input.second();
+ for (V joinValue : joinMap.get(key)) {
+ Pair<U, V> valuePair = Pair.of(value, joinValue);
+ emitter.emit(Pair.of(key, valuePair));
+ }
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/lib/join/RightOuterJoinFn.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/lib/join/RightOuterJoinFn.java b/crunch/src/main/java/org/apache/crunch/lib/join/RightOuterJoinFn.java
new file mode 100644
index 0000000..a7b2a0d
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/lib/join/RightOuterJoinFn.java
@@ -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.crunch.lib.join;
+
+import java.util.List;
+
+import org.apache.crunch.Emitter;
+import org.apache.crunch.Pair;
+import org.apache.crunch.types.PType;
+import com.google.common.collect.Lists;
+
+/**
+ * Used to perform the last step of an right outer join.
+ *
+ * @param <K> Type of the keys.
+ * @param <U> Type of the first {@link org.apache.crunch.PTable}'s values
+ * @param <V> Type of the second {@link org.apache.crunch.PTable}'s values
+ */
+public class RightOuterJoinFn<K, U, V> extends JoinFn<K, U, V> {
+
+ private transient K lastKey;
+ private transient List<U> leftValues;
+
+ public RightOuterJoinFn(PType<U> leftValueType) {
+ super(leftValueType);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void initialize() {
+ lastKey = null;
+ this.leftValues = Lists.newArrayList();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void join(K key, int id, Iterable<Pair<U, V>> pairs,
+ Emitter<Pair<K, Pair<U, V>>> emitter) {
+ if (!key.equals(lastKey)) {
+ lastKey = key;
+ leftValues.clear();
+ }
+ if (id == 0) {
+ for (Pair<U, V> pair : pairs) {
+ if (pair.first() != null)
+ leftValues.add(leftValueType.getDetachedValue(pair.first()));
+ }
+ } else {
+ for (Pair<U, V> pair : pairs) {
+ // Make sure that right side gets emitted.
+ if (leftValues.isEmpty()) {
+ leftValues.add(null);
+ }
+
+ for (U u : leftValues) {
+ emitter.emit(Pair.of(lastKey, Pair.of(u, pair.second())));
+ }
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String getJoinType() { return "rightOuterJoin"; }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/materialize/MaterializableIterable.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/MaterializableIterable.java b/crunch/src/main/java/org/apache/crunch/materialize/MaterializableIterable.java
new file mode 100644
index 0000000..c713d90
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/MaterializableIterable.java
@@ -0,0 +1,65 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.crunch.Pipeline;
+import org.apache.crunch.impl.mr.run.CrunchRuntimeException;
+import org.apache.crunch.io.ReadableSourceTarget;
+
+public class MaterializableIterable<E> implements Iterable<E> {
+
+ private static final Log LOG = LogFactory.getLog(MaterializableIterable.class);
+
+ private final Pipeline pipeline;
+ private final ReadableSourceTarget<E> sourceTarget;
+ private Iterable<E> materialized;
+
+ public MaterializableIterable(Pipeline pipeline, ReadableSourceTarget<E> source) {
+ this.pipeline = pipeline;
+ this.sourceTarget = source;
+ this.materialized = null;
+ }
+
+ public ReadableSourceTarget<E> getSourceTarget() {
+ return sourceTarget;
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ if (materialized == null) {
+ pipeline.run();
+ materialize();
+ }
+ return materialized.iterator();
+ }
+
+ public void materialize() {
+ try {
+ materialized = sourceTarget.read(pipeline.getConfiguration());
+ } catch (IOException e) {
+ LOG.error("Could not materialize: " + sourceTarget, e);
+ throw new CrunchRuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/materialize/MaterializableMap.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/materialize/MaterializableMap.java b/crunch/src/main/java/org/apache/crunch/materialize/MaterializableMap.java
new file mode 100644
index 0000000..e751f50
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/materialize/MaterializableMap.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;
+
+import java.util.AbstractMap;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.crunch.Pair;
+
+public class MaterializableMap<K, V> extends AbstractMap<K, V> {
+
+ private Iterable<Pair<K, V>> iterable;
+ private Set<Map.Entry<K, V>> entrySet;
+
+ public MaterializableMap(Iterable<Pair<K, V>> iterable) {
+ this.iterable = iterable;
+ }
+
+ private Set<Map.Entry<K, V>> toMapEntries(Iterable<Pair<K, V>> xs) {
+ HashMap<K, V> m = new HashMap<K, V>();
+ for (Pair<K, V> x : xs)
+ m.put(x.first(), x.second());
+ return m.entrySet();
+ }
+
+ @Override
+ public Set<Map.Entry<K, V>> entrySet() {
+ if (entrySet == null)
+ entrySet = toMapEntries(iterable);
+ return entrySet;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/test/FileHelper.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/test/FileHelper.java b/crunch/src/main/java/org/apache/crunch/test/FileHelper.java
new file mode 100644
index 0000000..7f5d0c7
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/test/FileHelper.java
@@ -0,0 +1,42 @@
+/**
+ * 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.test;
+
+import static com.google.common.io.Resources.getResource;
+import static com.google.common.io.Resources.newInputStreamSupplier;
+
+import java.io.File;
+import java.io.IOException;
+
+import com.google.common.io.Files;
+
+public class FileHelper {
+
+ public static String createTempCopyOf(String fileResource) throws IOException {
+ File tmpFile = File.createTempFile("tmp", "");
+ tmpFile.deleteOnExit();
+ Files.copy(newInputStreamSupplier(getResource(fileResource)), tmpFile);
+ return tmpFile.getAbsolutePath();
+ }
+
+ public static File createOutputPath() throws IOException {
+ File output = File.createTempFile("output", "");
+ output.delete();
+ return output;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/test/InMemoryEmitter.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/test/InMemoryEmitter.java b/crunch/src/main/java/org/apache/crunch/test/InMemoryEmitter.java
new file mode 100644
index 0000000..46e90ae
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/test/InMemoryEmitter.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.test;
+
+import java.util.List;
+
+import org.apache.crunch.Emitter;
+import com.google.common.collect.Lists;
+
+/**
+ * An {@code Emitter} instance that writes emitted records to a backing {@code List}.
+ *
+ * @param <T>
+ */
+public class InMemoryEmitter<T> implements Emitter<T> {
+
+ private final List<T> output;
+
+ public InMemoryEmitter() {
+ this(Lists.<T>newArrayList());
+ }
+
+ public InMemoryEmitter(List<T> output) {
+ this.output = output;
+ }
+
+ @Override
+ public void emit(T emitted) {
+ output.add(emitted);
+ }
+
+ @Override
+ public void flush() {
+
+ }
+
+ public List<T> getOutput() {
+ return output;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/test/TestCounters.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/test/TestCounters.java b/crunch/src/main/java/org/apache/crunch/test/TestCounters.java
new file mode 100644
index 0000000..8c8823c
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/test/TestCounters.java
@@ -0,0 +1,41 @@
+/**
+ * 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.test;
+
+import org.apache.hadoop.mapreduce.Counter;
+import org.apache.hadoop.mapreduce.Counters;
+
+/**
+ * A utility class used during unit testing to update and read counters.
+ */
+public class TestCounters {
+
+ private static Counters COUNTERS = new Counters();
+
+ public static Counter getCounter(Enum<?> e) {
+ return COUNTERS.findCounter(e);
+ }
+
+ public static Counter getCounter(String group, String name) {
+ return COUNTERS.findCounter(group, name);
+ }
+
+ public static void clearCounters() {
+ COUNTERS = new Counters();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/tool/CrunchTool.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/tool/CrunchTool.java b/crunch/src/main/java/org/apache/crunch/tool/CrunchTool.java
new file mode 100644
index 0000000..a788520
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/tool/CrunchTool.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.tool;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.util.Tool;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PTable;
+import org.apache.crunch.Pipeline;
+import org.apache.crunch.Source;
+import org.apache.crunch.TableSource;
+import org.apache.crunch.Target;
+import org.apache.crunch.impl.mem.MemPipeline;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.io.At;
+import org.apache.crunch.io.From;
+import org.apache.crunch.io.To;
+
+/**
+ * An extension of the {@code Tool} interface that creates a {@code Pipeline}
+ * instance and provides methods for working with the Pipeline from inside of
+ * the Tool's run method.
+ *
+ */
+public abstract class CrunchTool extends Configured implements Tool {
+
+ protected static final From from = new From();
+ protected static final To to = new To();
+ protected static final At at = new At();
+
+ private Pipeline pipeline;
+
+ public CrunchTool() throws IOException {
+ this(false);
+ }
+
+ public CrunchTool(boolean inMemory) throws IOException {
+ this.pipeline = inMemory ? MemPipeline.getInstance() : new MRPipeline(getClass());
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ super.setConf(conf);
+ if (conf != null && pipeline != null) {
+ pipeline.setConfiguration(conf);
+ }
+ }
+
+ @Override
+ public Configuration getConf() {
+ return pipeline.getConfiguration();
+ }
+
+ public void enableDebug() {
+ pipeline.enableDebug();
+ }
+
+ public <T> PCollection<T> read(Source<T> source) {
+ return pipeline.read(source);
+ }
+
+ public <K, V> PTable<K, V> read(TableSource<K, V> tableSource) {
+ return pipeline.read(tableSource);
+ }
+
+ public PCollection<String> readTextFile(String pathName) {
+ return pipeline.readTextFile(pathName);
+ }
+
+ public void write(PCollection<?> pcollection, Target target) {
+ pipeline.write(pcollection, target);
+ }
+
+ public void writeTextFile(PCollection<?> pcollection, String pathName) {
+ pipeline.writeTextFile(pcollection, pathName);
+ }
+
+ public void run() {
+ pipeline.run();
+ }
+
+ public void done() {
+ pipeline.done();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/Converter.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/Converter.java b/crunch/src/main/java/org/apache/crunch/types/Converter.java
new file mode 100644
index 0000000..70705f0
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/Converter.java
@@ -0,0 +1,41 @@
+/**
+ * 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.types;
+
+import java.io.Serializable;
+
+import org.apache.crunch.DoFn;
+
+/**
+ * Converts the input key/value from a MapReduce task into the input to a
+ * {@link DoFn}, or takes the output of a {@code DoFn} and write it to
+ * the output key/values.
+ */
+public interface Converter<K, V, S, T> extends Serializable {
+ S convertInput(K key, V value);
+
+ T convertIterableInput(K key, Iterable<V> value);
+
+ K outputKey(S value);
+
+ V outputValue(S value);
+
+ Class<K> getKeyClass();
+
+ Class<V> getValueClass();
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/PGroupedTableType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/PGroupedTableType.java b/crunch/src/main/java/org/apache/crunch/types/PGroupedTableType.java
new file mode 100644
index 0000000..e29018a
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/PGroupedTableType.java
@@ -0,0 +1,124 @@
+/**
+ * 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.types;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+
+import org.apache.crunch.GroupingOptions;
+import org.apache.crunch.MapFn;
+import org.apache.crunch.PGroupedTable;
+import org.apache.crunch.Pair;
+import org.apache.crunch.SourceTarget;
+
+/**
+ * The {@code PType} instance for {@link PGroupedTable} instances. Its settings are
+ * derived from the {@code PTableType} that was grouped to create the
+ * {@code PGroupedTable} instance.
+ *
+ */
+public abstract class PGroupedTableType<K, V> implements PType<Pair<K, Iterable<V>>> {
+
+ protected static class PTypeIterable<V> implements Iterable<V> {
+ private final Iterable<Object> iterable;
+ private final MapFn<Object, V> mapFn;
+
+ public PTypeIterable(MapFn<Object, V> mapFn, Iterable<Object> iterable) {
+ this.mapFn = mapFn;
+ this.iterable = iterable;
+ }
+
+ public Iterator<V> iterator() {
+ return new Iterator<V>() {
+ Iterator<Object> iter = iterable.iterator();
+
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ public V next() {
+ return mapFn.map(iter.next());
+ }
+
+ public void remove() {
+ iter.remove();
+ }
+ };
+ }
+ }
+
+ public static class PairIterableMapFn<K, V> extends
+ MapFn<Pair<Object, Iterable<Object>>, Pair<K, Iterable<V>>> {
+ private final MapFn<Object, K> keys;
+ private final MapFn<Object, V> values;
+
+ public PairIterableMapFn(MapFn<Object, K> keys, MapFn<Object, V> values) {
+ this.keys = keys;
+ this.values = values;
+ }
+
+ @Override
+ public void initialize() {
+ keys.initialize();
+ values.initialize();
+ }
+
+ @Override
+ public Pair<K, Iterable<V>> map(Pair<Object, Iterable<Object>> input) {
+ return Pair.<K, Iterable<V>>of(keys.map(input.first()),
+ new PTypeIterable(values, input.second()));
+ }
+ }
+
+ protected final PTableType<K, V> tableType;
+
+ public PGroupedTableType(PTableType<K, V> tableType) {
+ this.tableType = tableType;
+ }
+
+ public PTableType<K, V> getTableType() {
+ return tableType;
+ }
+
+ @Override
+ public PTypeFamily getFamily() {
+ return tableType.getFamily();
+ }
+
+ @Override
+ public List<PType> getSubTypes() {
+ return tableType.getSubTypes();
+ }
+
+ @Override
+ public Converter getConverter() {
+ return tableType.getConverter();
+ }
+
+ public abstract Converter getGroupingConverter();
+
+ public abstract void configureShuffle(Job job, GroupingOptions options);
+
+ @Override
+ public SourceTarget<Pair<K, Iterable<V>>> getDefaultFileSource(Path path) {
+ throw new UnsupportedOperationException("Grouped tables cannot be written out directly");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/PTableType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/PTableType.java b/crunch/src/main/java/org/apache/crunch/types/PTableType.java
new file mode 100644
index 0000000..b9e946f
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/PTableType.java
@@ -0,0 +1,44 @@
+/**
+ * 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.types;
+
+import org.apache.crunch.PTable;
+import org.apache.crunch.Pair;
+
+/**
+ * An extension of {@code PType} specifically for {@link PTable} objects. It
+ * allows separate access to the {@code PType}s of the key and value for the
+ * {@code PTable}.
+ *
+ */
+public interface PTableType<K, V> extends PType<Pair<K, V>> {
+ /**
+ * Returns the key type for the table.
+ */
+ PType<K> getKeyType();
+
+ /**
+ * Returns the value type for the table.
+ */
+ PType<V> getValueType();
+
+ /**
+ * Returns the grouped table version of this type.
+ */
+ PGroupedTableType<K, V> getGroupedTableType();
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/PType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/PType.java b/crunch/src/main/java/org/apache/crunch/types/PType.java
new file mode 100644
index 0000000..126b165
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/PType.java
@@ -0,0 +1,82 @@
+/**
+ * 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.types;
+
+import java.io.Serializable;
+import java.util.List;
+
+import org.apache.hadoop.fs.Path;
+
+import org.apache.crunch.DoFn;
+import org.apache.crunch.MapFn;
+import org.apache.crunch.PCollection;
+import org.apache.crunch.SourceTarget;
+
+/**
+ * A {@code PType} defines a mapping between a data type that is used in a
+ * Crunch pipeline and a serialization and storage format that is used to
+ * read/write data from/to HDFS. Every {@link PCollection} has an associated
+ * {@code PType} that tells Crunch how to read/write data from that
+ * {@code PCollection}.
+ *
+ */
+public interface PType<T> extends Serializable {
+ /**
+ * Returns the Java type represented by this {@code PType}.
+ */
+ Class<T> getTypeClass();
+
+ /**
+ * Returns the {@code PTypeFamily} that this {@code PType} belongs to.
+ */
+ PTypeFamily getFamily();
+
+ MapFn<Object, T> getInputMapFn();
+
+ MapFn<T, Object> getOutputMapFn();
+
+ Converter getConverter();
+
+ /**
+ * Returns a copy of a value (or the value itself) that can safely be
+ * retained.
+ * <p>
+ * This is useful when iterable values being processed in a DoFn (via a
+ * reducer) need to be held on to for more than the scope of a single
+ * iteration, as a reducer (and therefore also a DoFn that has an Iterable as
+ * input) re-use deserialized values. More information on object reuse is
+ * available in the {@link DoFn} class documentation.
+ *
+ * @param value
+ * The value to be deep-copied
+ * @return A deep copy of the input value
+ */
+ T getDetachedValue(T value);
+
+ /**
+ * Returns a {@code SourceTarget} that is able to read/write data using the
+ * serialization format specified by this {@code PType}.
+ */
+ SourceTarget<T> getDefaultFileSource(Path path);
+
+ /**
+ * Returns the sub-types that make up this PType if it is a composite
+ * instance, such as a tuple.
+ */
+ List<PType> getSubTypes();
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/PTypeFamily.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/PTypeFamily.java b/crunch/src/main/java/org/apache/crunch/types/PTypeFamily.java
new file mode 100644
index 0000000..2ae9493
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/PTypeFamily.java
@@ -0,0 +1,79 @@
+/**
+ * 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.types;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.crunch.MapFn;
+import org.apache.crunch.Pair;
+import org.apache.crunch.Tuple;
+import org.apache.crunch.Tuple3;
+import org.apache.crunch.Tuple4;
+import org.apache.crunch.TupleN;
+
+/**
+ * An abstract factory for creating {@code PType} instances that have the same
+ * serialization/storage backing format.
+ *
+ */
+public interface PTypeFamily {
+ PType<Void> nulls();
+
+ PType<String> strings();
+
+ PType<Long> longs();
+
+ PType<Integer> ints();
+
+ PType<Float> floats();
+
+ PType<Double> doubles();
+
+ PType<Boolean> booleans();
+
+ PType<ByteBuffer> bytes();
+
+ <T> PType<T> records(Class<T> clazz);
+
+ <T> PType<Collection<T>> collections(PType<T> ptype);
+
+ <T> PType<Map<String, T>> maps(PType<T> ptype);
+
+ <V1, V2> PType<Pair<V1, V2>> pairs(PType<V1> p1, PType<V2> p2);
+
+ <V1, V2, V3> PType<Tuple3<V1, V2, V3>> triples(PType<V1> p1, PType<V2> p2,
+ PType<V3> p3);
+
+ <V1, V2, V3, V4> PType<Tuple4<V1, V2, V3, V4>> quads(PType<V1> p1,
+ PType<V2> p2, PType<V3> p3, PType<V4> p4);
+
+ PType<TupleN> tuples(PType<?>... ptypes);
+
+ <T extends Tuple> PType<T> tuples(Class<T> clazz, PType<?>... ptypes);
+
+ <S, T> PType<T> derived(Class<T> clazz, MapFn<S, T> inputFn, MapFn<T, S> outputFn, PType<S> base);
+
+ <K, V> PTableType<K, V> tableOf(PType<K> key, PType<V> value);
+
+ /**
+ * Returns the equivalent of the given ptype for this family, if it exists.
+ */
+ <T> PType<T> as(PType<T> ptype);
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/PTypeUtils.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/PTypeUtils.java b/crunch/src/main/java/org/apache/crunch/types/PTypeUtils.java
new file mode 100644
index 0000000..4dda78d
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/PTypeUtils.java
@@ -0,0 +1,66 @@
+/**
+ * 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.types;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.crunch.Pair;
+import org.apache.crunch.Tuple;
+import org.apache.crunch.Tuple3;
+import org.apache.crunch.Tuple4;
+import org.apache.crunch.TupleN;
+
+/**
+ * Utilities for converting between {@code PType}s from different {@code PTypeFamily}
+ * implementations.
+ *
+ */
+public class PTypeUtils {
+
+ public static <T> PType<T> convert(PType<T> ptype, PTypeFamily tf) {
+ if (ptype instanceof PTableType) {
+ PTableType ptt = (PTableType) ptype;
+ return tf.tableOf(tf.as(ptt.getKeyType()), tf.as(ptt.getValueType()));
+ }
+ Class<T> typeClass = ptype.getTypeClass();
+ if (Tuple.class.isAssignableFrom(typeClass)) {
+ List<PType> subTypes = ptype.getSubTypes();
+ if (Pair.class.equals(typeClass)) {
+ return tf.pairs(tf.as(subTypes.get(0)), tf.as(subTypes.get(1)));
+ } else if (Tuple3.class.equals(typeClass)) {
+ return tf.triples(tf.as(subTypes.get(0)), tf.as(subTypes.get(1)), tf.as(subTypes.get(2)));
+ } else if (Tuple4.class.equals(typeClass)) {
+ return tf.quads(tf.as(subTypes.get(0)), tf.as(subTypes.get(1)),
+ tf.as(subTypes.get(2)), tf.as(subTypes.get(3)));
+ } else if (TupleN.class.equals(typeClass)) {
+ PType[] newPTypes = subTypes.toArray(new PType[0]);
+ for (int i = 0; i < newPTypes.length; i++) {
+ newPTypes[i] = tf.as(subTypes.get(i));
+ }
+ return (PType<T>) tf.tuples(newPTypes);
+ }
+ }
+ if (Collection.class.isAssignableFrom(typeClass)) {
+ return tf.collections(tf.as(ptype.getSubTypes().get(0)));
+ }
+ return tf.records(typeClass);
+ }
+
+ private PTypeUtils() {}
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/TupleFactory.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/TupleFactory.java b/crunch/src/main/java/org/apache/crunch/types/TupleFactory.java
new file mode 100644
index 0000000..49aae0d
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/TupleFactory.java
@@ -0,0 +1,98 @@
+/**
+ * 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.types;
+
+import java.io.Serializable;
+import java.lang.reflect.Constructor;
+
+import org.apache.crunch.Pair;
+import org.apache.crunch.Tuple;
+import org.apache.crunch.Tuple3;
+import org.apache.crunch.Tuple4;
+import org.apache.crunch.TupleN;
+import org.apache.crunch.impl.mr.run.CrunchRuntimeException;
+
+public abstract class TupleFactory<T extends Tuple> implements Serializable {
+
+ public void initialize() { }
+
+ public abstract T makeTuple(Object...values);
+
+ public static final TupleFactory<Pair> PAIR = new TupleFactory<Pair>() {
+ @Override
+ public Pair makeTuple(Object... values) {
+ return Pair.of(values[0], values[1]);
+ }
+ };
+
+ public static final TupleFactory<Tuple3> TUPLE3 = new TupleFactory<Tuple3>() {
+ @Override
+ public Tuple3 makeTuple(Object... values) {
+ return Tuple3.of(values[0], values[1], values[2]);
+ }
+ };
+
+ public static final TupleFactory<Tuple4> TUPLE4 = new TupleFactory<Tuple4>() {
+ @Override
+ public Tuple4 makeTuple(Object... values) {
+ return Tuple4.of(values[0], values[1], values[2], values[3]);
+ }
+ };
+
+ public static final TupleFactory<TupleN> TUPLEN = new TupleFactory<TupleN>() {
+ @Override
+ public TupleN makeTuple(Object... values) {
+ return new TupleN(values);
+ }
+ };
+
+ public static <T extends Tuple> TupleFactory<T> create(Class<T> clazz, Class... typeArgs) {
+ return new CustomTupleFactory<T>(clazz, typeArgs);
+ }
+
+ private static class CustomTupleFactory<T extends Tuple> extends TupleFactory<T> {
+
+ private final Class<T> clazz;
+ private final Class[] typeArgs;
+
+ private transient Constructor<T> constructor;
+
+ public CustomTupleFactory(Class<T> clazz, Class[] typeArgs) {
+ this.clazz = clazz;
+ this.typeArgs = typeArgs;
+ }
+
+ @Override
+ public void initialize() {
+ try {
+ constructor = clazz.getConstructor(typeArgs);
+ } catch (Exception e) {
+ throw new CrunchRuntimeException(e);
+ }
+ }
+
+ @Override
+ public T makeTuple(Object... values) {
+ try {
+ return constructor.newInstance(values);
+ } catch (Exception e) {
+ throw new CrunchRuntimeException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroDeepCopier.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroDeepCopier.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroDeepCopier.java
new file mode 100644
index 0000000..be9f5e9
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroDeepCopier.java
@@ -0,0 +1,151 @@
+/**
+ * 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.types.avro;
+
+import java.io.ByteArrayOutputStream;
+import java.io.Serializable;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericData.Record;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.reflect.ReflectDatumReader;
+import org.apache.avro.reflect.ReflectDatumWriter;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.avro.specific.SpecificDatumWriter;
+
+import org.apache.crunch.impl.mr.run.CrunchRuntimeException;
+
+/**
+ * Performs deep copies of Avro-serializable objects.
+ * <p>
+ * <b>Warning:</b> Methods in this class are not thread-safe. This shouldn't be
+ * a problem when running in a map-reduce context where each mapper/reducer is
+ * running in its own JVM, but it may well be a problem in any other kind of
+ * multi-threaded context.
+ */
+public abstract class AvroDeepCopier<T> implements Serializable {
+
+ private BinaryEncoder binaryEncoder;
+ private BinaryDecoder binaryDecoder;
+ protected DatumWriter<T> datumWriter;
+ protected DatumReader<T> datumReader;
+
+ protected AvroDeepCopier(DatumWriter<T> datumWriter, DatumReader<T> datumReader) {
+ this.datumWriter = datumWriter;
+ this.datumReader = datumReader;
+ }
+
+ protected abstract T createCopyTarget();
+
+ /**
+ * Deep copier for Avro specific data objects.
+ */
+ public static class AvroSpecificDeepCopier<T> extends AvroDeepCopier<T> {
+
+ private Class<T> valueClass;
+
+ public AvroSpecificDeepCopier(Class<T> valueClass, Schema schema) {
+ super(new SpecificDatumWriter<T>(schema), new SpecificDatumReader(schema));
+ this.valueClass = valueClass;
+ }
+
+ @Override
+ protected T createCopyTarget() {
+ return createNewInstance(valueClass);
+ }
+
+ }
+
+ /**
+ * Deep copier for Avro generic data objects.
+ */
+ public static class AvroGenericDeepCopier extends AvroDeepCopier<Record> {
+
+ private Schema schema;
+
+ public AvroGenericDeepCopier(Schema schema) {
+ super(new GenericDatumWriter<Record>(schema), new GenericDatumReader<Record>(schema));
+ this.schema = schema;
+ }
+
+ @Override
+ protected Record createCopyTarget() {
+ return new GenericData.Record(schema);
+ }
+ }
+
+ /**
+ * Deep copier for Avro reflect data objects.
+ */
+ public static class AvroReflectDeepCopier<T> extends AvroDeepCopier<T> {
+ private Class<T> valueClass;
+
+ public AvroReflectDeepCopier(Class<T> valueClass, Schema schema) {
+ super(new ReflectDatumWriter<T>(schema), new ReflectDatumReader<T>(schema));
+ this.valueClass = valueClass;
+ }
+
+ @Override
+ protected T createCopyTarget() {
+ return createNewInstance(valueClass);
+ }
+ }
+
+ /**
+ * Create a deep copy of an Avro value.
+ *
+ * @param source
+ * The value to be copied
+ * @return The deep copy of the value
+ */
+ public T deepCopy(T source) {
+ ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
+ binaryEncoder = EncoderFactory.get().binaryEncoder(byteOutStream, binaryEncoder);
+ T target = createCopyTarget();
+ try {
+ datumWriter.write(source, binaryEncoder);
+ binaryEncoder.flush();
+ binaryDecoder = DecoderFactory.get()
+ .binaryDecoder(byteOutStream.toByteArray(), binaryDecoder);
+ datumReader.read(target, binaryDecoder);
+ } catch (Exception e) {
+ throw new CrunchRuntimeException("Error while deep copying avro value " + source, e);
+ }
+
+ return target;
+ }
+
+ protected T createNewInstance(Class<T> targetClass) {
+ try {
+ return targetClass.newInstance();
+ } catch (InstantiationException e) {
+ throw new CrunchRuntimeException(e);
+ } catch (IllegalAccessException e) {
+ throw new CrunchRuntimeException(e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroGroupedTableType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroGroupedTableType.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroGroupedTableType.java
new file mode 100644
index 0000000..5ce970f
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroGroupedTableType.java
@@ -0,0 +1,110 @@
+/**
+ * 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.types.avro;
+
+import java.util.Collection;
+
+import org.apache.avro.mapred.AvroJob;
+import org.apache.avro.mapred.AvroKey;
+import org.apache.avro.mapred.AvroKeyComparator;
+import org.apache.avro.mapred.AvroValue;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Job;
+
+import org.apache.crunch.GroupingOptions;
+import org.apache.crunch.MapFn;
+import org.apache.crunch.Pair;
+import org.apache.crunch.fn.PairMapFn;
+import org.apache.crunch.lib.PTables;
+import org.apache.crunch.types.Converter;
+import org.apache.crunch.types.PGroupedTableType;
+
+/**
+ *
+ *
+ */
+public class AvroGroupedTableType<K, V> extends PGroupedTableType<K, V> {
+
+ private static final AvroPairConverter CONVERTER = new AvroPairConverter();
+ private final MapFn inputFn;
+ private final MapFn outputFn;
+
+ public AvroGroupedTableType(AvroTableType<K, V> tableType) {
+ super(tableType);
+ AvroType keyType = (AvroType) tableType.getKeyType();
+ AvroType valueType = (AvroType) tableType.getValueType();
+ this.inputFn = new PairIterableMapFn(keyType.getInputMapFn(),
+ valueType.getInputMapFn());
+ this.outputFn = new PairMapFn(keyType.getOutputMapFn(),
+ valueType.getOutputMapFn());
+ }
+
+ @Override
+ public Class<Pair<K, Iterable<V>>> getTypeClass() {
+ return (Class<Pair<K, Iterable<V>>>) Pair.of(null, null).getClass();
+ }
+
+ @Override
+ public Converter getGroupingConverter() {
+ return CONVERTER;
+ }
+
+ @Override
+ public MapFn getInputMapFn() {
+ return inputFn;
+ }
+
+ @Override
+ public MapFn getOutputMapFn() {
+ return outputFn;
+ }
+
+ @Override
+ public Pair<K, Iterable<V>> getDetachedValue(Pair<K, Iterable<V>> value) {
+ return PTables.getGroupedDetachedValue(this, value);
+ }
+
+ @Override
+ public void configureShuffle(Job job, GroupingOptions options) {
+ AvroTableType<K, V> att = (AvroTableType<K, V>) tableType;
+ String schemaJson = att.getSchema().toString();
+ Configuration conf = job.getConfiguration();
+
+ if (!att.isSpecific()) {
+ conf.setBoolean(AvroJob.MAP_OUTPUT_IS_REFLECT, true);
+ }
+ conf.set(AvroJob.MAP_OUTPUT_SCHEMA, schemaJson);
+ job.setSortComparatorClass(AvroKeyComparator.class);
+ job.setMapOutputKeyClass(AvroKey.class);
+ job.setMapOutputValueClass(AvroValue.class);
+ if (options != null) {
+ options.configure(job);
+ }
+
+ Avros.configureReflectDataFactory(conf);
+
+ Collection<String> serializations =
+ job.getConfiguration().getStringCollection("io.serializations");
+ if (!serializations.contains(SafeAvroSerialization.class.getName())) {
+ serializations.add(SafeAvroSerialization.class.getName());
+ job.getConfiguration().setStrings("io.serializations",
+ serializations.toArray(new String[0]));
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroInputFormat.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroInputFormat.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroInputFormat.java
new file mode 100644
index 0000000..da5bbb2
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroInputFormat.java
@@ -0,0 +1,58 @@
+/**
+ * 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.types.avro;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.avro.Schema;
+import org.apache.avro.mapred.AvroJob;
+import org.apache.avro.mapred.AvroWrapper;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+
+/** An {@link org.apache.hadoop.mapreduce.InputFormat} for Avro data files. */
+public class AvroInputFormat<T> extends FileInputFormat<AvroWrapper<T>, NullWritable> {
+
+ @Override
+ protected List<FileStatus> listStatus(JobContext job) throws IOException {
+ List<FileStatus> result = new ArrayList<FileStatus>();
+ for (FileStatus file : super.listStatus(job)) {
+ if (file.getPath().getName().endsWith(org.apache.avro.mapred.AvroOutputFormat.EXT)) {
+ result.add(file);
+ }
+ }
+ return result;
+ }
+
+ @Override
+ public RecordReader<AvroWrapper<T>, NullWritable> createRecordReader(InputSplit split,
+ TaskAttemptContext context) throws IOException, InterruptedException {
+ context.setStatus(split.toString());
+ String jsonSchema = context.getConfiguration().get(AvroJob.INPUT_SCHEMA);
+ Schema schema = new Schema.Parser().parse(jsonSchema);
+ return new AvroRecordReader<T>(schema);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroKeyConverter.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroKeyConverter.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroKeyConverter.java
new file mode 100644
index 0000000..60aa69b
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroKeyConverter.java
@@ -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.crunch.types.avro;
+
+import org.apache.avro.mapred.AvroWrapper;
+import org.apache.hadoop.io.NullWritable;
+
+import org.apache.crunch.types.Converter;
+
+public class AvroKeyConverter<K> implements Converter<AvroWrapper<K>, NullWritable, K, Iterable<K>> {
+
+ private transient AvroWrapper<K> wrapper = null;
+
+ @Override
+ public K convertInput(AvroWrapper<K> key, NullWritable value) {
+ return key.datum();
+ }
+
+ @Override
+ public AvroWrapper<K> outputKey(K value) {
+ getWrapper().datum(value);
+ return wrapper;
+ }
+
+ @Override
+ public NullWritable outputValue(K value) {
+ return NullWritable.get();
+ }
+
+ @Override
+ public Class<AvroWrapper<K>> getKeyClass() {
+ return (Class<AvroWrapper<K>>) getWrapper().getClass();
+ }
+
+ @Override
+ public Class<NullWritable> getValueClass() {
+ return NullWritable.class;
+ }
+
+ private AvroWrapper<K> getWrapper() {
+ if (wrapper == null) {
+ wrapper = new AvroWrapper<K>();
+ }
+ return wrapper;
+ }
+
+ @Override
+ public Iterable<K> convertIterableInput(AvroWrapper<K> key,
+ Iterable<NullWritable> value) {
+ throw new UnsupportedOperationException("Should not be possible");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroOutputFormat.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroOutputFormat.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroOutputFormat.java
new file mode 100644
index 0000000..838c21a
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroOutputFormat.java
@@ -0,0 +1,71 @@
+/**
+ * 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.types.avro;
+
+import java.io.IOException;
+
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.mapred.AvroJob;
+import org.apache.avro.mapred.AvroWrapper;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+
+/** An {@link org.apache.hadoop.mapreduce.OutputFormat} for Avro data files. */
+public class AvroOutputFormat<T> extends FileOutputFormat<AvroWrapper<T>, NullWritable> {
+
+ @Override
+ public RecordWriter<AvroWrapper<T>, NullWritable> getRecordWriter(
+ TaskAttemptContext context) throws IOException, InterruptedException {
+
+ Configuration conf = context.getConfiguration();
+ Schema schema = null;
+ String outputName = conf.get("crunch.namedoutput");
+ if (outputName != null && !outputName.isEmpty()) {
+ schema = (new Schema.Parser()).parse(conf.get("avro.output.schema." + outputName));
+ } else {
+ schema = AvroJob.getOutputSchema(context.getConfiguration());
+ }
+
+ ReflectDataFactory factory = Avros.getReflectDataFactory(conf);
+ final DataFileWriter<T> WRITER = new DataFileWriter<T>(factory.<T>getWriter());
+
+ Path path = getDefaultWorkFile(context,
+ org.apache.avro.mapred.AvroOutputFormat.EXT);
+ WRITER.create(schema,
+ path.getFileSystem(context.getConfiguration()).create(path));
+
+ return new RecordWriter<AvroWrapper<T>, NullWritable>() {
+ @Override
+ public void write(AvroWrapper<T> wrapper, NullWritable ignore)
+ throws IOException {
+ WRITER.append(wrapper.datum());
+ }
+ @Override
+ public void close(TaskAttemptContext context) throws IOException,
+ InterruptedException {
+ WRITER.close();
+ }
+ };
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroPairConverter.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroPairConverter.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroPairConverter.java
new file mode 100644
index 0000000..6ec3972
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroPairConverter.java
@@ -0,0 +1,109 @@
+/**
+ * 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.types.avro;
+
+import java.util.Iterator;
+
+import org.apache.avro.mapred.AvroKey;
+import org.apache.avro.mapred.AvroValue;
+
+import org.apache.crunch.Pair;
+import org.apache.crunch.types.Converter;
+
+public class AvroPairConverter<K, V> implements Converter<AvroKey<K>, AvroValue<V>, Pair<K, V>, Pair<K, Iterable<V>>> {
+
+ private transient AvroKey<K> keyWrapper = null;
+ private transient AvroValue<V> valueWrapper = null;
+
+ @Override
+ public Pair<K, V> convertInput(AvroKey<K> key, AvroValue<V> value) {
+ return Pair.of(key.datum(), value.datum());
+ }
+
+ public Pair<K, Iterable<V>> convertIterableInput(AvroKey<K> key, Iterable<AvroValue<V>> iter) {
+ Iterable<V> it = new AvroWrappedIterable<V>(iter);
+ return Pair.of(key.datum(), it);
+ }
+
+ @Override
+ public AvroKey<K> outputKey(Pair<K, V> value) {
+ getKeyWrapper().datum(value.first());
+ return keyWrapper;
+ }
+
+ @Override
+ public AvroValue<V> outputValue(Pair<K, V> value) {
+ getValueWrapper().datum(value.second());
+ return valueWrapper;
+ }
+
+ @Override
+ public Class<AvroKey<K>> getKeyClass() {
+ return (Class<AvroKey<K>>) getKeyWrapper().getClass();
+ }
+
+ @Override
+ public Class<AvroValue<V>> getValueClass() {
+ return (Class<AvroValue<V>>) getValueWrapper().getClass();
+ }
+
+ private AvroKey<K> getKeyWrapper() {
+ if (keyWrapper == null) {
+ keyWrapper = new AvroKey<K>();
+ }
+ return keyWrapper;
+ }
+
+ private AvroValue<V> getValueWrapper() {
+ if (valueWrapper == null) {
+ valueWrapper = new AvroValue<V>();
+ }
+ return valueWrapper;
+ }
+
+ private static class AvroWrappedIterable<V> implements Iterable<V> {
+
+ private final Iterable<AvroValue<V>> iters;
+
+ public AvroWrappedIterable(Iterable<AvroValue<V>> iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public Iterator<V> iterator() {
+ return new Iterator<V>() {
+ private final Iterator<AvroValue<V>> it = iters.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override
+ public V next() {
+ return it.next().datum();
+ }
+
+ @Override
+ public void remove() {
+ it.remove();
+ }
+ };
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroRecordReader.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroRecordReader.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroRecordReader.java
new file mode 100644
index 0000000..3bcab5c
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroRecordReader.java
@@ -0,0 +1,115 @@
+/**
+ * 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.types.avro;
+
+import java.io.IOException;
+
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileReader;
+import org.apache.avro.file.FileReader;
+import org.apache.avro.file.SeekableInput;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.mapred.AvroJob;
+import org.apache.avro.mapred.AvroWrapper;
+import org.apache.avro.mapred.FsInput;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+
+/** An {@link RecordReader} for Avro data files. */
+public class AvroRecordReader<T> extends RecordReader<AvroWrapper<T>, NullWritable> {
+
+ private FileReader<T> reader;
+ private long start;
+ private long end;
+ private AvroWrapper<T> key;
+ private NullWritable value;
+ private Schema schema;
+
+ public AvroRecordReader(Schema schema) {
+ this.schema = schema;
+ }
+
+ @Override
+ public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException,
+ InterruptedException {
+ FileSplit split = (FileSplit) genericSplit;
+ Configuration conf = context.getConfiguration();
+ SeekableInput in = new FsInput(split.getPath(), conf);
+ DatumReader<T> datumReader = null;
+ if (context.getConfiguration().getBoolean(AvroJob.INPUT_IS_REFLECT, true)) {
+ ReflectDataFactory factory = Avros.getReflectDataFactory(conf);
+ datumReader = factory.getReader(schema);
+ } else {
+ datumReader = new SpecificDatumReader<T>(schema);
+ }
+ this.reader = DataFileReader.openReader(in, datumReader);
+ reader.sync(split.getStart()); // sync to start
+ this.start = reader.tell();
+ this.end = split.getStart() + split.getLength();
+ }
+
+ @Override
+ public boolean nextKeyValue() throws IOException, InterruptedException {
+ if (!reader.hasNext() || reader.pastSync(end)) {
+ key = null;
+ value = null;
+ return false;
+ }
+ if (key == null) {
+ key = new AvroWrapper<T>();
+ }
+ if (value == null) {
+ value = NullWritable.get();
+ }
+ key.datum(reader.next(key.datum()));
+ return true;
+ }
+
+ @Override
+ public AvroWrapper<T> getCurrentKey() throws IOException, InterruptedException {
+ return key;
+ }
+
+ @Override
+ public NullWritable getCurrentValue() throws IOException, InterruptedException {
+ return value;
+ }
+
+ @Override
+ public float getProgress() throws IOException {
+ if (end == start) {
+ return 0.0f;
+ } else {
+ return Math.min(1.0f, (getPos() - start) / (float) (end - start));
+ }
+ }
+
+ public long getPos() throws IOException {
+ return reader.tell();
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroTableType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroTableType.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroTableType.java
new file mode 100644
index 0000000..6d21122
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroTableType.java
@@ -0,0 +1,167 @@
+/**
+ * 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.types.avro;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.crunch.MapFn;
+import org.apache.crunch.Pair;
+import org.apache.crunch.lib.PTables;
+import org.apache.crunch.types.PGroupedTableType;
+import org.apache.crunch.types.PTableType;
+import org.apache.crunch.types.PType;
+
+/**
+ * The implementation of the PTableType interface for Avro-based serialization.
+ *
+ */
+public class AvroTableType<K, V> extends AvroType<Pair<K, V>> implements
+ PTableType<K, V> {
+
+ private static class PairToAvroPair extends
+ MapFn<Pair, org.apache.avro.mapred.Pair> {
+ private final MapFn keyMapFn;
+ private final MapFn valueMapFn;
+ private final String firstJson;
+ private final String secondJson;
+
+ private String pairSchemaJson;
+ private transient Schema pairSchema;
+
+ public PairToAvroPair(AvroType keyType, AvroType valueType) {
+ this.keyMapFn = keyType.getOutputMapFn();
+ this.firstJson = keyType.getSchema().toString();
+ this.valueMapFn = valueType.getOutputMapFn();
+ this.secondJson = valueType.getSchema().toString();
+ }
+
+ @Override
+ public void configure(Configuration conf) {
+ keyMapFn.configure(conf);
+ valueMapFn.configure(conf);
+ }
+
+ @Override
+ public void setConfigurationForTest(Configuration conf) {
+ keyMapFn.setConfigurationForTest(conf);
+ valueMapFn.setConfigurationForTest(conf);
+ }
+
+ @Override
+ public void initialize() {
+ keyMapFn.setContext(getContext());
+ valueMapFn.setContext(getContext());
+ pairSchemaJson = org.apache.avro.mapred.Pair.getPairSchema(
+ new Schema.Parser().parse(firstJson),
+ new Schema.Parser().parse(secondJson)).toString();
+ }
+
+ @Override
+ public org.apache.avro.mapred.Pair map(Pair input) {
+ if (pairSchema == null) {
+ pairSchema = new Schema.Parser().parse(pairSchemaJson);
+ }
+ org.apache.avro.mapred.Pair avroPair = new org.apache.avro.mapred.Pair(
+ pairSchema);
+ avroPair.key(keyMapFn.map(input.first()));
+ avroPair.value(valueMapFn.map(input.second()));
+ return avroPair;
+ }
+ }
+
+ private static class IndexedRecordToPair extends MapFn<IndexedRecord, Pair> {
+
+ private final MapFn firstMapFn;
+ private final MapFn secondMapFn;
+
+ public IndexedRecordToPair(MapFn firstMapFn, MapFn secondMapFn) {
+ this.firstMapFn = firstMapFn;
+ this.secondMapFn = secondMapFn;
+ }
+
+ @Override
+ public void configure(Configuration conf) {
+ firstMapFn.configure(conf);
+ secondMapFn.configure(conf);
+ }
+
+ @Override
+ public void setConfigurationForTest(Configuration conf) {
+ firstMapFn.setConfigurationForTest(conf);
+ secondMapFn.setConfigurationForTest(conf);
+ }
+
+ @Override
+ public void initialize() {
+ firstMapFn.setContext(getContext());
+ secondMapFn.setContext(getContext());
+ }
+
+ @Override
+ public Pair map(IndexedRecord input) {
+ return Pair.of(firstMapFn.map(input.get(0)),
+ secondMapFn.map(input.get(1)));
+ }
+ }
+
+ private final AvroType<K> keyType;
+ private final AvroType<V> valueType;
+
+ public AvroTableType(AvroType<K> keyType, AvroType<V> valueType,
+ Class<Pair<K, V>> pairClass) {
+ super(pairClass, org.apache.avro.mapred.Pair.getPairSchema(
+ keyType.getSchema(), valueType.getSchema()),
+ new IndexedRecordToPair(keyType.getInputMapFn(),
+ valueType.getInputMapFn()), new PairToAvroPair(keyType,
+ valueType), keyType, valueType);
+ this.keyType = keyType;
+ this.valueType = valueType;
+ }
+
+ @Override
+ public boolean isSpecific() {
+ return keyType.isSpecific() || valueType.isSpecific();
+ }
+
+ @Override
+ public boolean isGeneric() {
+ return keyType.isGeneric() || valueType.isGeneric();
+ }
+
+ @Override
+ public PType<K> getKeyType() {
+ return keyType;
+ }
+
+ @Override
+ public PType<V> getValueType() {
+ return valueType;
+ }
+
+ @Override
+ public PGroupedTableType<K, V> getGroupedTableType() {
+ return new AvroGroupedTableType<K, V>(this);
+ }
+
+ @Override
+ public Pair<K, V> getDetachedValue(Pair<K, V> value) {
+ return PTables.getDetachedValue(this, value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroType.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroType.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroType.java
new file mode 100644
index 0000000..d5d22a8
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroType.java
@@ -0,0 +1,172 @@
+/**
+ * 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.types.avro;
+
+import java.util.List;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.specific.SpecificRecord;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.crunch.MapFn;
+import org.apache.crunch.SourceTarget;
+import org.apache.crunch.fn.IdentityFn;
+import org.apache.crunch.io.avro.AvroFileSourceTarget;
+import org.apache.crunch.types.Converter;
+import org.apache.crunch.types.PType;
+import org.apache.crunch.types.PTypeFamily;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+/**
+ * The implementation of the PType interface for Avro-based serialization.
+ *
+ */
+public class AvroType<T> implements PType<T> {
+
+ private static final Converter AVRO_CONVERTER = new AvroKeyConverter();
+
+ private final Class<T> typeClass;
+ private final String schemaString;
+ private transient Schema schema;
+ private final MapFn baseInputMapFn;
+ private final MapFn baseOutputMapFn;
+ private final List<PType> subTypes;
+ private AvroDeepCopier<T> deepCopier;
+
+ public AvroType(Class<T> typeClass, Schema schema, PType... ptypes) {
+ this(typeClass, schema, IdentityFn.getInstance(), IdentityFn
+ .getInstance(), ptypes);
+ }
+
+ public AvroType(Class<T> typeClass, Schema schema, MapFn inputMapFn,
+ MapFn outputMapFn, PType... ptypes) {
+ this.typeClass = typeClass;
+ this.schema = Preconditions.checkNotNull(schema);
+ this.schemaString = schema.toString();
+ this.baseInputMapFn = inputMapFn;
+ this.baseOutputMapFn = outputMapFn;
+ this.subTypes = ImmutableList.<PType> builder().add(ptypes).build();
+ }
+
+ @Override
+ public Class<T> getTypeClass() {
+ return typeClass;
+ }
+
+ @Override
+ public PTypeFamily getFamily() {
+ return AvroTypeFamily.getInstance();
+ }
+
+ @Override
+ public List<PType> getSubTypes() {
+ return subTypes;
+ }
+
+ public Schema getSchema() {
+ if (schema == null){
+ schema = new Schema.Parser().parse(schemaString);
+ }
+ return schema;
+ }
+
+ /**
+ * Determine if the wrapped type is a specific data avro type.
+ *
+ * @return true if the wrapped type is a specific data type
+ */
+ public boolean isSpecific() {
+ if (SpecificRecord.class.isAssignableFrom(typeClass)) {
+ return true;
+ }
+ for (PType ptype : subTypes) {
+ if (SpecificRecord.class.isAssignableFrom(ptype.getTypeClass())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Determine if the wrapped type is a generic data avro type.
+ *
+ * @return true if the wrapped type is a generic type
+ */
+ public boolean isGeneric() {
+ return GenericData.Record.class.equals(typeClass);
+ }
+
+ public MapFn<Object, T> getInputMapFn() {
+ return baseInputMapFn;
+ }
+
+ public MapFn<T, Object> getOutputMapFn() {
+ return baseOutputMapFn;
+ }
+
+ @Override
+ public Converter getConverter() {
+ return AVRO_CONVERTER;
+ }
+
+ @Override
+ public SourceTarget<T> getDefaultFileSource(Path path) {
+ return new AvroFileSourceTarget<T>(path, this);
+ }
+
+ private AvroDeepCopier<T> getDeepCopier() {
+ if (deepCopier == null) {
+ if (isSpecific()) {
+ deepCopier = new AvroDeepCopier.AvroSpecificDeepCopier<T>(typeClass, getSchema());
+ } else if (isGeneric()) {
+ deepCopier = (AvroDeepCopier<T>) new AvroDeepCopier.AvroGenericDeepCopier(getSchema());
+ } else {
+ deepCopier = new AvroDeepCopier.AvroReflectDeepCopier<T>(typeClass, getSchema());
+ }
+ }
+ return deepCopier;
+ }
+
+ public T getDetachedValue(T value) {
+ if (this.baseInputMapFn instanceof IdentityFn && !Avros.isPrimitive(this)) {
+ return getDeepCopier().deepCopy(value);
+ }
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof AvroType)) {
+ return false;
+ }
+ AvroType at = (AvroType) other;
+ return (typeClass.equals(at.typeClass) && subTypes.equals(at.subTypes));
+
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder hcb = new HashCodeBuilder();
+ hcb.append(typeClass).append(subTypes);
+ return hcb.toHashCode();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/83acb813/crunch/src/main/java/org/apache/crunch/types/avro/AvroTypeFamily.java
----------------------------------------------------------------------
diff --git a/crunch/src/main/java/org/apache/crunch/types/avro/AvroTypeFamily.java b/crunch/src/main/java/org/apache/crunch/types/avro/AvroTypeFamily.java
new file mode 100644
index 0000000..b7d5598
--- /dev/null
+++ b/crunch/src/main/java/org/apache/crunch/types/avro/AvroTypeFamily.java
@@ -0,0 +1,169 @@
+/**
+ * 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.types.avro;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+
+import org.apache.crunch.MapFn;
+import org.apache.crunch.Pair;
+import org.apache.crunch.Tuple;
+import org.apache.crunch.Tuple3;
+import org.apache.crunch.Tuple4;
+import org.apache.crunch.TupleN;
+import org.apache.crunch.types.PGroupedTableType;
+import org.apache.crunch.types.PTableType;
+import org.apache.crunch.types.PType;
+import org.apache.crunch.types.PTypeFamily;
+import org.apache.crunch.types.PTypeUtils;
+
+public class AvroTypeFamily implements PTypeFamily {
+
+ private static final AvroTypeFamily INSTANCE = new AvroTypeFamily();
+
+ public static AvroTypeFamily getInstance() {
+ return INSTANCE;
+ }
+
+ // There can only be one instance.
+ private AvroTypeFamily() {
+ }
+
+
+ @Override
+ public PType<Void> nulls() {
+ return Avros.nulls();
+ }
+
+ @Override
+ public PType<String> strings() {
+ return Avros.strings();
+ }
+
+ @Override
+ public PType<Long> longs() {
+ return Avros.longs();
+ }
+
+ @Override
+ public PType<Integer> ints() {
+ return Avros.ints();
+ }
+
+ @Override
+ public PType<Float> floats() {
+ return Avros.floats();
+ }
+
+ @Override
+ public PType<Double> doubles() {
+ return Avros.doubles();
+ }
+
+ @Override
+ public PType<Boolean> booleans() {
+ return Avros.booleans();
+ }
+
+ @Override
+ public PType<ByteBuffer> bytes() {
+ return Avros.bytes();
+ }
+
+ @Override
+ public <T> PType<T> records(Class<T> clazz) {
+ return Avros.records(clazz);
+ }
+
+ public PType<GenericData.Record> generics(Schema schema) {
+ return Avros.generics(schema);
+ }
+
+ public <T> PType<T> containers(Class<T> clazz) {
+ return Avros.containers(clazz);
+ }
+
+ @Override
+ public <T> PType<Collection<T>> collections(PType<T> ptype) {
+ return Avros.collections(ptype);
+ }
+
+ @Override
+ public <T> PType<Map<String, T>> maps(PType<T> ptype) {
+ return Avros.maps(ptype);
+ }
+
+ @Override
+ public <V1, V2> PType<Pair<V1, V2>> pairs(PType<V1> p1, PType<V2> p2) {
+ return Avros.pairs(p1, p2);
+ }
+
+ @Override
+ public <V1, V2, V3> PType<Tuple3<V1, V2, V3>> triples(PType<V1> p1,
+ PType<V2> p2, PType<V3> p3) {
+ return Avros.triples(p1, p2, p3);
+ }
+
+ @Override
+ public <V1, V2, V3, V4> PType<Tuple4<V1, V2, V3, V4>> quads(PType<V1> p1,
+ PType<V2> p2, PType<V3> p3, PType<V4> p4) {
+ return Avros.quads(p1, p2, p3, p4);
+ }
+
+ @Override
+ public PType<TupleN> tuples(PType<?>... ptypes) {
+ return Avros.tuples(ptypes);
+ }
+
+ @Override
+ public <K, V> PTableType<K, V> tableOf(PType<K> key, PType<V> value) {
+ return Avros.tableOf(key, value);
+ }
+
+ @Override
+ public <T> PType<T> as(PType<T> ptype) {
+ if (ptype instanceof AvroType || ptype instanceof AvroGroupedTableType) {
+ return ptype;
+ }
+ if (ptype instanceof PGroupedTableType) {
+ PTableType ptt = ((PGroupedTableType) ptype).getTableType();
+ return new AvroGroupedTableType((AvroTableType) as(ptt));
+ }
+ Class<T> typeClass = ptype.getTypeClass();
+ PType<T> prim = Avros.getPrimitiveType(typeClass);
+ if (prim != null) {
+ return prim;
+ }
+ return PTypeUtils.convert(ptype, this);
+ }
+
+ @Override
+ public <T extends Tuple> PType<T> tuples(Class<T> clazz, PType<?>... ptypes) {
+ return Avros.tuples(clazz, ptypes);
+ }
+
+ @Override
+ public <S, T> PType<T> derived(Class<T> clazz, MapFn<S, T> inputFn,
+ MapFn<T, S> outputFn, PType<S> base) {
+ return Avros.derived(clazz, inputFn, outputFn, base);
+ }
+}