You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2014/08/06 03:09:20 UTC
[4/4] git commit: TEZ-1379. Allow EdgeConfigurers to accept
Configuration for Comparators. Change the way partitioner, comparator,
combiner confs are set (from Hadoop Configuration to Map). Rename specific
Input/Output classes from *Configuration to *Con
TEZ-1379. Allow EdgeConfigurers to accept Configuration for Comparators.
Change the way partitioner, comparator, combiner confs are set (from
Hadoop Configuration to Map). Rename specific Input/Output classes from
*Configuration to *Configurer. (sseth)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/675b2372
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/675b2372
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/675b2372
Branch: refs/heads/master
Commit: 675b237244f6bee827b8967541fda58ff1ff8c52
Parents: 1831fda
Author: Siddharth Seth <ss...@apache.org>
Authored: Tue Aug 5 18:08:43 2014 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Tue Aug 5 18:08:43 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../mapreduce/examples/IntersectExample.java | 2 +-
.../mapreduce/examples/IntersectValidate.java | 2 +-
.../tez/mapreduce/examples/MRRSleepJob.java | 8 +-
.../mapreduce/examples/OrderedWordCount.java | 4 +-
.../tez/mapreduce/examples/UnionExample.java | 2 +-
.../tez/mapreduce/examples/WordCount.java | 2 +-
.../apache/tez/mapreduce/client/YARNRunner.java | 10 +-
.../tez/runtime/library/common/ConfigUtils.java | 4 +-
.../HadoopKeyValuesBasedBaseEdgeConfigurer.java | 8 +-
.../conf/OnFileSortedOutputConfiguration.java | 356 --------------
.../conf/OnFileSortedOutputConfigurer.java | 416 ++++++++++++++++
.../OnFileUnorderedKVOutputConfiguration.java | 250 ----------
.../conf/OnFileUnorderedKVOutputConfigurer.java | 257 ++++++++++
...orderedPartitionedKVOutputConfiguration.java | 289 -----------
...eUnorderedPartitionedKVOutputConfigurer.java | 302 ++++++++++++
.../OrderedPartitionedKVEdgeConfigurer.java | 94 +++-
.../conf/ShuffledMergedInputConfiguration.java | 457 -----------------
.../conf/ShuffledMergedInputConfigurer.java | 489 +++++++++++++++++++
.../ShuffledUnorderedKVInputConfiguration.java | 343 -------------
.../ShuffledUnorderedKVInputConfigurer.java | 331 +++++++++++++
.../UnorderedPartitionedKVEdgeConfigurer.java | 72 ++-
.../UnorderedUnpartitionedKVEdgeConfigurer.java | 41 +-
.../TestOnFileSortedOutputConfiguration.java | 194 --------
.../conf/TestOnFileSortedOutputConfigurer.java | 195 ++++++++
...estOnFileUnorderedKVOutputConfiguration.java | 132 -----
.../TestOnFileUnorderedKVOutputConfigurer.java | 132 +++++
.../TestOnFileUnorderedPartitionedKVOutput.java | 180 -------
...eUnorderedPartitionedKVOutputConfigurer.java | 181 +++++++
.../TestOrderedPartitionedKVEdgeConfigurer.java | 47 +-
.../TestShuffledMergedInputConfiguration.java | 162 ------
.../conf/TestShuffledMergedInputConfigurer.java | 163 +++++++
...stShuffledUnorderedKVInputConfiguration.java | 134 -----
.../TestShuffledUnorderedKVInputConfigurer.java | 134 +++++
...estUnorderedPartitionedKVEdgeConfigurer.java | 36 +-
...tUnorderedUnpartitionedKVEdgeConfigurer.java | 24 +-
36 files changed, 2823 insertions(+), 2632 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 5e4f2d8..80a1168 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -46,6 +46,8 @@ INCOMPATIBLE CHANGES
TEZ-1351. MROutput needs a flush method to ensure data is materialized for
FileOutputCommitter
TEZ-1317. Simplify MRinput/MROutput configuration
+ TEZ-1379. Allow EdgeConfigurers to accept Configuration for Comparators. Change the way partitioner, comparator, combiner confs are set (from Hadoop Configuration to Map).
+ Rename specific Input/Output classes from *Configuration to *Configurer.
Release 0.4.0-incubating: 2014-04-05
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectExample.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectExample.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectExample.java
index 9a28014..cc85324 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectExample.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectExample.java
@@ -181,7 +181,7 @@ public class IntersectExample extends Configured implements Tool {
UnorderedPartitionedKVEdgeConfigurer edgeConf =
UnorderedPartitionedKVEdgeConfigurer
.newBuilder(Text.class.getName(), NullWritable.class.getName(),
- HashPartitioner.class.getName(), null).build();
+ HashPartitioner.class.getName()).build();
// Change the way resources are setup - no MRHelpers
Vertex streamFileVertex = new Vertex("partitioner1", new ProcessorDescriptor(
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectValidate.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectValidate.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectValidate.java
index ac671a4..9e076ca 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectValidate.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/IntersectValidate.java
@@ -187,7 +187,7 @@ public class IntersectValidate extends Configured implements Tool {
// better mechanism to configure the IOs.
OrderedPartitionedKVEdgeConfigurer edgeConf = OrderedPartitionedKVEdgeConfigurer
.newBuilder(Text.class.getName(), NullWritable.class.getName(),
- HashPartitioner.class.getName(), null).build();
+ HashPartitioner.class.getName()).build();
// Change the way resources are setup - no MRHelpers
Vertex lhsVertex = new Vertex(LHS_INPUT_NAME, new ProcessorDescriptor(
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
index 7279136..8aad617 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -84,6 +85,7 @@ import org.apache.tez.mapreduce.processor.reduce.ReduceProcessor;
import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfigurer;
import com.google.common.annotations.VisibleForTesting;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
/**
* Dummy class for testing MR framefork. Sleeps for a defined period
@@ -587,11 +589,11 @@ public class MRRSleepJob extends Configured implements Tool {
}
- Configuration partitionerConf = new Configuration(false);
- partitionerConf.set(MRJobConfig.PARTITIONER_CLASS_ATTR, MRRSleepJobPartitioner.class.getName());
+ Map<String, String> partitionerConf = Maps.newHashMap();
+ partitionerConf.put(MRJobConfig.PARTITIONER_CLASS_ATTR, MRRSleepJobPartitioner.class.getName());
OrderedPartitionedKVEdgeConfigurer edgeConf = OrderedPartitionedKVEdgeConfigurer
.newBuilder(IntWritable.class.getName(), IntWritable.class.getName(),
- MRPartitioner.class.getName(), partitionerConf).configureInput().useLegacyInput()
+ HashPartitioner.class.getName(), partitionerConf).configureInput().useLegacyInput()
.done().build();
for (int i = 0; i < vertices.size(); ++i) {
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
index 14240ea..3513c0e 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
@@ -266,14 +266,14 @@ public class OrderedWordCount extends Configured implements Tool {
OrderedPartitionedKVEdgeConfigurer edgeConf1 = OrderedPartitionedKVEdgeConfigurer
.newBuilder(Text.class.getName(), IntWritable.class.getName(),
- HashPartitioner.class.getName(), null).configureInput().useLegacyInput().done().build();
+ HashPartitioner.class.getName()).configureInput().useLegacyInput().done().build();
dag.addEdge(
new Edge(dag.getVertex("initialmap"), dag.getVertex("intermediate_reducer"),
edgeConf1.createDefaultEdgeProperty()));
OrderedPartitionedKVEdgeConfigurer edgeConf2 = OrderedPartitionedKVEdgeConfigurer
.newBuilder(IntWritable.class.getName(), Text.class.getName(),
- HashPartitioner.class.getName(), null).configureInput().useLegacyInput().done().build();
+ HashPartitioner.class.getName()).configureInput().useLegacyInput().done().build();
dag.addEdge(
new Edge(dag.getVertex("intermediate_reducer"), dag.getVertex("finalreduce"),
edgeConf2.createDefaultEdgeProperty()));
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
index 259a314..77c6656 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
@@ -213,7 +213,7 @@ public class UnionExample {
OrderedPartitionedKVEdgeConfigurer edgeConf = OrderedPartitionedKVEdgeConfigurer
.newBuilder(Text.class.getName(), IntWritable.class.getName(),
- HashPartitioner.class.getName(), null).build();
+ HashPartitioner.class.getName()).build();
dag.addVertex(mapVertex1)
.addVertex(mapVertex2)
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
index ab5ac57..cb13025 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
@@ -134,7 +134,7 @@ public class WordCount extends Configured implements Tool {
OrderedPartitionedKVEdgeConfigurer edgeConf = OrderedPartitionedKVEdgeConfigurer
.newBuilder(Text.class.getName(), IntWritable.class.getName(),
- HashPartitioner.class.getName(), null).build();
+ HashPartitioner.class.getName()).build();
DAG dag = new DAG("WordCount");
dag.addVertex(tokenizerVertex)
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
index 2bc2ff0..4158d12 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
@@ -29,6 +29,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;
+import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -473,11 +474,18 @@ public class YARNRunner implements ClientProtocol {
if (i > 0) {
// Set edge conf based on Input conf (compression etc properties for MapReduce are
// w.r.t Outputs - MAP_OUTPUT_COMPRESS for example)
+ Map<String, String> partitionerConf = null;
+ if (stageConfs[i-1] != null) {
+ partitionerConf = Maps.newHashMap();
+ for (Map.Entry<String, String> entry : stageConfs[i - 1]) {
+ partitionerConf.put(entry.getKey(), entry.getValue());
+ }
+ }
OrderedPartitionedKVEdgeConfigurer edgeConf =
OrderedPartitionedKVEdgeConfigurer.newBuilder(stageConfs[i - 1].get(
TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS),
stageConfs[i - 1].get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS),
- MRPartitioner.class.getName(), stageConfs[i - 1])
+ MRPartitioner.class.getName(), partitionerConf)
.configureInput().useLegacyInput().done()
.setFromConfiguration(stageConfs[i - 1]).build();
Edge edge = new Edge(vertices[i-1], vertices[i], edgeConf.createDefaultEdgeProperty());
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ConfigUtils.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ConfigUtils.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ConfigUtils.java
index a376dd6..1744452 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ConfigUtils.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ConfigUtils.java
@@ -219,10 +219,10 @@ public class ConfigUtils {
}
@InterfaceAudience.Private
- public static void mergeConfsWithExclusions(Configuration destConf, Configuration srcConf, Set<String> excludedKeySet) {
+ public static void mergeConfsWithExclusions(Configuration destConf, Map<String, String> srcConf, Set<String> excludedKeySet) {
Preconditions.checkState(destConf != null, "Destination conf cannot be null");
Preconditions.checkState(srcConf != null, "Source conf cannot be null");
- for (Map.Entry<String, String> entry : srcConf) {
+ for (Map.Entry<String, String> entry : srcConf.entrySet()) {
if (!excludedKeySet.contains(entry.getKey())) {
destConf.set(entry.getKey(), entry.getValue());
}
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/HadoopKeyValuesBasedBaseEdgeConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/HadoopKeyValuesBasedBaseEdgeConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/HadoopKeyValuesBasedBaseEdgeConfigurer.java
index 35d023f..bf41a26 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/HadoopKeyValuesBasedBaseEdgeConfigurer.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/HadoopKeyValuesBasedBaseEdgeConfigurer.java
@@ -18,6 +18,8 @@
package org.apache.tez.runtime.library.conf;
+import javax.annotation.Nullable;
+
import org.apache.hadoop.classification.InterfaceAudience;
@InterfaceAudience.Private
@@ -52,10 +54,12 @@ abstract class HadoopKeyValuesBasedBaseEdgeConfigurer {
/**
* Enable compression for the specific Input / Output / Edge
*
- * @param compressionCodec the codec to be used. null implies using the default
+ * @param enabled whether to enable compression or not
+ * @param compressionCodec the codec to be used if compression is enabled. null implies using
+ * the default
* @return instance of the current builder
*/
- public T enableCompression(String compressionCodec) {
+ public T setCompression(boolean enabled, @Nullable String compressionCodec) {
return (T) this;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfiguration.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfiguration.java
deleted file mode 100644
index fed20ab..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfiguration.java
+++ /dev/null
@@ -1,356 +0,0 @@
-/*
- * *
- * * 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.tez.runtime.library.conf;
-
-import java.io.IOException;
-import java.util.Map;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.tez.common.TezUtils;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.output.OnFileSortedOutput;
-
-
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-public class OnFileSortedOutputConfiguration {
-
- /**
- * Configure parameters which are specific to the Output.
- */
- @InterfaceAudience.Private
- public static interface SpecificConfigurer<T> extends BaseConfigurer<T> {
- /**
- * Set the buffer size to use when sort the output
- *
- * @param sortBufferSize the size of the buffer in MB
- * @return instance of the current builder
- */
- public T setSortBufferSize(int sortBufferSize);
-
- /**
- * Configure the combiner class
- *
- * @param combinerClassName the combiner class name
- * @param combinerConf configuration for the combiner. This can be null
- * @return instance of the current builder
- */
- public T setCombiner(String combinerClassName, Configuration combinerConf);
-
- /**
- * Configure the number of threads to be used by the sorter
- *
- * @param numThreads the number of threads
- * @return instance of the current builder
- */
- public T setSorterNumThreads(int numThreads);
- }
-
- @SuppressWarnings("rawtypes")
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class SpecificBuilder<E extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder> implements
- SpecificConfigurer<SpecificBuilder> {
-
- private final E edgeBuilder;
- private final Builder builder;
-
- SpecificBuilder(E edgeBuilder, Builder builder) {
- this.edgeBuilder = edgeBuilder;
- this.builder = builder;
- }
-
- @Override
- public SpecificBuilder<E> setSortBufferSize(int sortBufferSize) {
- builder.setSortBufferSize(sortBufferSize);
- return this;
- }
-
- @Override
- public SpecificBuilder<E> setCombiner(String combinerClassName, Configuration combinerConf) {
- builder.setCombiner(combinerClassName, combinerConf);
- return this;
- }
-
- @Override
- public SpecificBuilder<E> setSorterNumThreads(int numThreads) {
- builder.setSorterNumThreads(numThreads);
- return this;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(String key, String value) {
- builder.setAdditionalConfiguration(key, value);
- return this;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(Map<String, String> confMap) {
- builder.setAdditionalConfiguration(confMap);
- return this;
- }
-
- @Override
- public SpecificBuilder setFromConfiguration(Configuration conf) {
- builder.setFromConfiguration(conf);
- return this;
- }
-
- public E done() {
- return edgeBuilder;
- }
- }
-
- @InterfaceAudience.Private
- @VisibleForTesting
- Configuration conf;
-
- @InterfaceAudience.Private
- @VisibleForTesting
- OnFileSortedOutputConfiguration() {
- }
-
- private OnFileSortedOutputConfiguration(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Get a byte array representation of the configuration
- * @return a byte array which can be used as the payload
- */
- public byte[] toByteArray() {
- try {
- return TezUtils.createUserPayloadFromConf(conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @InterfaceAudience.Private
- public void fromByteArray(byte[] payload) {
- try {
- this.conf = TezUtils.createConfFromUserPayload(payload);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- public static Builder newBuilder(String keyClass, String valueClass, String partitionerClassName,
- Configuration partitionerConf) {
- return new Builder(keyClass, valueClass, partitionerClassName, partitionerConf);
- }
-
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class Builder implements SpecificConfigurer<Builder> {
-
- private final Configuration conf = new Configuration(false);
-
- /**
- * Create a configuration builder for {@link org.apache.tez.runtime.library.output.OnFileSortedOutput}
- *
- * @param keyClassName the key class name
- * @param valueClassName the value class name
- * @param partitionerClassName the partitioner class name
- * @param partitionerConf configuration for the partitioner. This can be null
- */
- @InterfaceAudience.Private
- Builder(String keyClassName, String valueClassName, String partitionerClassName,
- Configuration partitionerConf) {
- this();
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
- setKeyClassName(keyClassName);
- setValueClassName(valueClassName);
- setPartitioner(partitionerClassName, partitionerConf);
- }
-
- @InterfaceAudience.Private
- Builder() {
- Map<String, String> tezDefaults = ConfigUtils
- .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
- OnFileSortedOutput.getConfigurationKeySet());
- ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
- ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
- }
-
- @InterfaceAudience.Private
- Builder setKeyClassName(String keyClassName) {
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClassName);
- return this;
- }
-
- @InterfaceAudience.Private
- Builder setValueClassName(String valueClassName) {
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
- return this;
- }
-
- @InterfaceAudience.Private
- Builder setPartitioner(String partitionerClassName, Configuration partitionerConf) {
- Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, partitionerClassName);
- if (partitionerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, partitionerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- @Override
- public Builder setSortBufferSize(int sortBufferSize) {
- this.conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, sortBufferSize);
- return this;
- }
-
- @Override
- public Builder setCombiner(String combinerClassName, Configuration combinerConf) {
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS, combinerClassName);
- if (combinerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, combinerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- @Override
- public Builder setSorterNumThreads(int numThreads) {
- this.conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_THREADS, numThreads);
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(String key, String value) {
- Preconditions.checkNotNull(key, "Key cannot be null");
- if (ConfigUtils.doesKeyQualify(key,
- Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
- TezRuntimeConfiguration.getAllowedPrefixes())) {
- if (value == null) {
- this.conf.unset(key);
- } else {
- this.conf.set(key, value);
- }
- }
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(Map<String, String> confMap) {
- Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
- Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- @Override
- public Builder setFromConfiguration(Configuration conf) {
- // Maybe ensure this is the first call ? Otherwise this can end up overriding other parameters
- Preconditions.checkArgument(conf != null, "Configuration cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
- Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- /**
- * Set the key comparator class
- *
- * @param comparatorClassName the key comparator class name
- * @return instance of the current builder
- */
- public Builder setKeyComparatorClass(String comparatorClassName) {
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
- comparatorClassName);
- return this;
- }
-
- public Builder enableCompression(String compressionCodec) {
- this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, true);
- if (compressionCodec != null) {
- this.conf
- .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
- }
- return this;
- }
-
- /**
- * Set serialization class and the relevant comparator to be used for sorting.
- * Providing custom serialization class could change the way, keys needs to be compared in
- * sorting. Providing invalid comparator here could create invalid results.
- *
- * @param serializationClassName
- * @param comparatorClassName
- * @return
- */
- public Builder setKeySerializationClass(String serializationClassName,
- String comparatorClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- Preconditions.checkArgument(comparatorClassName != null,
- "comparator cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- setKeyComparatorClass(comparatorClassName);
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @return
- */
- public Builder setValueSerializationClass(String serializationClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- return this;
- }
-
- /**
- * Create the actual configuration instance.
- *
- * @return an instance of the Configuration
- */
- public OnFileSortedOutputConfiguration build() {
- return new OnFileSortedOutputConfiguration(this.conf);
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfigurer.java
new file mode 100644
index 0000000..d2cd85f
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileSortedOutputConfigurer.java
@@ -0,0 +1,416 @@
+/*
+ * *
+ * * 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.tez.runtime.library.conf;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.Map;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.ConfigUtils;
+import org.apache.tez.runtime.library.output.OnFileSortedOutput;
+
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+/**
+ * Configure {@link org.apache.tez.runtime.library.output.OnFileSortedOutput. </p>
+ *
+ * Values will be picked up from tez-site if not specified, otherwise defaults from
+ * {@link org.apache.tez.runtime.library.api.TezRuntimeConfiguration} will be used.
+ */
+public class OnFileSortedOutputConfigurer {
+
+ /**
+ * Configure parameters which are specific to the Output.
+ */
+ @InterfaceAudience.Private
+ public static interface SpecificConfigurer<T> extends BaseConfigurer<T> {
+ /**
+ * Set the buffer size to use when sort the output
+ *
+ * @param sortBufferSize the size of the buffer in MB
+ * @return instance of the current builder
+ */
+ public T setSortBufferSize(int sortBufferSize);
+
+
+ /**
+ * Configure the combiner class
+ *
+ * @param combinerClassName the combiner class name
+ * @return instance of the current builder
+ */
+ public T setCombiner(String combinerClassName);
+
+ /**
+ * Configure the combiner class and it's associated configuration (specified as key-value
+ * pairs). This method should only be used if the combiner requires some specific configuration.
+ * {@link #setCombiner(String)} is the preferred method for setting a combiner.
+ *
+ * @param combinerClassName the combiner class name
+ * @param combinerConf the combiner configuration. This can be null, and otherwise
+ * is a {@link java.util.Map} of key-value pairs. The keys should
+ * be limited to the ones required by the combiner.
+ * @return instance of the current builder
+ */
+ public T setCombiner(String combinerClassName, @Nullable Map<String, String> combinerConf);
+
+
+
+ /**
+ * Configure the number of threads to be used by the sorter
+ *
+ * @param numThreads the number of threads
+ * @return instance of the current builder
+ */
+ public T setSorterNumThreads(int numThreads);
+ }
+
+ @SuppressWarnings("rawtypes")
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class SpecificBuilder<E extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder> implements
+ SpecificConfigurer<SpecificBuilder> {
+
+ private final E edgeBuilder;
+ private final Builder builder;
+
+ SpecificBuilder(E edgeBuilder, Builder builder) {
+ this.edgeBuilder = edgeBuilder;
+ this.builder = builder;
+ }
+
+ @Override
+ public SpecificBuilder<E> setSortBufferSize(int sortBufferSize) {
+ builder.setSortBufferSize(sortBufferSize);
+ return this;
+ }
+
+ public SpecificBuilder<E> setCombiner(String combinerClassName) {
+ return this.setCombiner(combinerClassName, null);
+ }
+
+ @Override
+ public SpecificBuilder<E> setCombiner(String combinerClassName, Map<String, String> combinerConf) {
+ builder.setCombiner(combinerClassName, combinerConf);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder<E> setSorterNumThreads(int numThreads) {
+ builder.setSorterNumThreads(numThreads);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder setAdditionalConfiguration(String key, String value) {
+ builder.setAdditionalConfiguration(key, value);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder setAdditionalConfiguration(Map<String, String> confMap) {
+ builder.setAdditionalConfiguration(confMap);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder setFromConfiguration(Configuration conf) {
+ builder.setFromConfiguration(conf);
+ return this;
+ }
+
+ public E done() {
+ return edgeBuilder;
+ }
+ }
+
+ @InterfaceAudience.Private
+ @VisibleForTesting
+ Configuration conf;
+
+ @InterfaceAudience.Private
+ @VisibleForTesting
+ OnFileSortedOutputConfigurer() {
+ }
+
+ private OnFileSortedOutputConfigurer(Configuration conf) {
+ this.conf = conf;
+ }
+
+ /**
+ * Get a byte array representation of the configuration
+ * @return a byte array which can be used as the payload
+ */
+ public byte[] toByteArray() {
+ try {
+ return TezUtils.createUserPayloadFromConf(conf);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @InterfaceAudience.Private
+ public void fromByteArray(byte[] payload) {
+ try {
+ this.conf = TezUtils.createConfFromUserPayload(payload);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static Builder newBuilder(String keyClass, String valueClass, String partitionerClassName) {
+ return newBuilder(keyClass, valueClass, partitionerClassName, null);
+ }
+
+ public static Builder newBuilder(String keyClass, String valueClass, String partitionerClassName,
+ Map<String, String> partitionerConf) {
+ return new Builder(keyClass, valueClass, partitionerClassName, partitionerConf);
+ }
+
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class Builder implements SpecificConfigurer<Builder> {
+
+ private final Configuration conf = new Configuration(false);
+
+ /**
+ * Create a configuration builder for {@link org.apache.tez.runtime.library.output.OnFileSortedOutput}
+ *
+ * @param keyClassName the key class name
+ * @param valueClassName the value class name
+ * @param partitionerClassName the partitioner class name
+ * @param partitionerConf the partitioner configuration. This can be null, and is a {@link
+ * java.util.Map} of key-value pairs. The keys should be limited to
+ * the ones required by the partitioner.
+ */
+ @InterfaceAudience.Private
+ Builder(String keyClassName, String valueClassName, String partitionerClassName,
+ @Nullable Map<String, String> partitionerConf) {
+ this();
+ Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
+ Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
+ Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
+ setKeyClassName(keyClassName);
+ setValueClassName(valueClassName);
+ setPartitioner(partitionerClassName, partitionerConf);
+ }
+
+ @InterfaceAudience.Private
+ Builder() {
+ Map<String, String> tezDefaults = ConfigUtils
+ .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
+ OnFileSortedOutput.getConfigurationKeySet());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
+ ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
+ }
+
+ @InterfaceAudience.Private
+ Builder setKeyClassName(String keyClassName) {
+ Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClassName);
+ return this;
+ }
+
+ @InterfaceAudience.Private
+ Builder setValueClassName(String valueClassName) {
+ Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
+ return this;
+ }
+
+ @InterfaceAudience.Private
+ Builder setPartitioner(String partitionerClassName, @Nullable Map<String, String> partitionerConf) {
+ Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, partitionerClassName);
+ if (partitionerConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, partitionerConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ }
+ return this;
+ }
+
+ @Override
+ public Builder setSortBufferSize(int sortBufferSize) {
+ this.conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, sortBufferSize);
+ return this;
+ }
+
+ @Override
+ public Builder setCombiner(String combinerClassName) {
+ return this.setCombiner(combinerClassName, null);
+ }
+
+ @Override
+ public Builder setCombiner(String combinerClassName, Map<String, String> combinerConf) {
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS, combinerClassName);
+ if (combinerConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, combinerConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ }
+ return this;
+ }
+
+ @Override
+ public Builder setSorterNumThreads(int numThreads) {
+ this.conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_THREADS, numThreads);
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(String key, String value) {
+ Preconditions.checkNotNull(key, "Key cannot be null");
+ if (ConfigUtils.doesKeyQualify(key,
+ Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
+ TezRuntimeConfiguration.getAllowedPrefixes())) {
+ if (value == null) {
+ this.conf.unset(key);
+ } else {
+ this.conf.set(key, value);
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(Map<String, String> confMap) {
+ Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
+ Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
+ Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, map);
+ return this;
+ }
+
+ @Override
+ public Builder setFromConfiguration(Configuration conf) {
+ // Maybe ensure this is the first call ? Otherwise this can end up overriding other parameters
+ Preconditions.checkArgument(conf != null, "Configuration cannot be null");
+ Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
+ Lists.newArrayList(OnFileSortedOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, map);
+ return this;
+ }
+
+ /**
+ * Set the key comparator class
+ *
+ * @param comparatorClassName the key comparator class name
+ * @return instance of the current builder
+ */
+ public Builder setKeyComparatorClass(String comparatorClassName) {
+ return this.setKeyComparatorClass(comparatorClassName, null);
+ }
+
+ /**
+ * Set the key comparator class and it's associated configuration. This method should only be
+ * used if the comparator requires some specific configuration, which is typically not the
+ * case. {@link #setKeyComparatorClass(String)} is the preferred method for setting a
+ * comparator.
+ *
+ * @param comparatorClassName the key comparator class name
+ * @param comparatorConf the comparator configuration. This can be null, and is a {@link
+ * java.util.Map} of key-value pairs. The keys should be limited to
+ * the ones required by the comparator.
+ * @return instance of the current builder
+ */
+ public Builder setKeyComparatorClass(String comparatorClassName,
+ @Nullable Map<String, String> comparatorConf) {
+ Preconditions.checkNotNull(comparatorClassName, "Comparator class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ comparatorClassName);
+ if (comparatorConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, comparatorConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ }
+ return this;
+ }
+
+ public Builder setCompression(boolean enabled, @Nullable String compressionCodec) {
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, enabled);
+ if (enabled && compressionCodec != null) {
+ this.conf
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
+ }
+ return this;
+ }
+
+ /**
+ * Set serialization class and the relevant comparator to be used for sorting.
+ * Providing custom serialization class could change the way, keys needs to be compared in
+ * sorting. Providing invalid comparator here could create invalid results.
+ *
+ * @param serializationClassName
+ * @param comparatorClassName
+ * @return
+ */
+ public Builder setKeySerializationClass(String serializationClassName,
+ String comparatorClassName) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ Preconditions.checkArgument(comparatorClassName != null,
+ "comparator cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ setKeyComparatorClass(comparatorClassName, null);
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for values.
+ *
+ * @param serializationClassName
+ * @return
+ */
+ public Builder setValueSerializationClass(String serializationClassName) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ return this;
+ }
+
+ /**
+ * Create the actual configuration instance.
+ *
+ * @return an instance of the Configuration
+ */
+ public OnFileSortedOutputConfigurer build() {
+ return new OnFileSortedOutputConfigurer(this.conf);
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfiguration.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfiguration.java
deleted file mode 100644
index e8e8be1..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfiguration.java
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
- * *
- * * 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.tez.runtime.library.conf;
-
-import java.io.IOException;
-import java.util.Map;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.tez.common.TezUtils;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput;
-
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-public class OnFileUnorderedKVOutputConfiguration {
- /**
- * Configure parameters which are specific to the Output.
- */
- @InterfaceAudience.Private
- public static interface SpecificConfigurer<T> extends BaseConfigurer<T> {
- }
-
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class SpecificBuilder<E extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder> implements
- SpecificConfigurer<SpecificBuilder> {
-
- private final E edgeBuilder;
- private final Builder builder;
-
- SpecificBuilder(E edgeBuilder, Builder builder) {
- this.edgeBuilder = edgeBuilder;
- this.builder = builder;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(String key, String value) {
- builder.setAdditionalConfiguration(key, value);
- return this;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(Map<String, String> confMap) {
- builder.setAdditionalConfiguration(confMap);
- return this;
- }
-
- @Override
- public SpecificBuilder setFromConfiguration(Configuration conf) {
- builder.setFromConfiguration(conf);
- return this;
- }
-
- public E done() {
- return edgeBuilder;
- }
- }
-
- @InterfaceAudience.Private
- @VisibleForTesting
- Configuration conf;
-
- @InterfaceAudience.Private
- @VisibleForTesting
- OnFileUnorderedKVOutputConfiguration() {
- }
-
- private OnFileUnorderedKVOutputConfiguration(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Get a byte array representation of the configuration
- * @return a byte array which can be used as the payload
- */
- public byte[] toByteArray() {
- try {
- return TezUtils.createUserPayloadFromConf(conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @InterfaceAudience.Private
- public void fromByteArray(byte[] payload) {
- try {
- this.conf = TezUtils.createConfFromUserPayload(payload);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- public static Builder newBuilder(String keyClass, String valClass) {
- return new Builder(keyClass, valClass);
- }
-
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class Builder implements SpecificConfigurer<Builder> {
-
- private final Configuration conf = new Configuration(false);
-
- /**
- * Create a configuration builder for {@link org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput}
- *
- * @param keyClassName the key class name
- * @param valueClassName the value class name
- */
- @InterfaceAudience.Private
- Builder(String keyClassName, String valueClassName) {
- this();
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- setKeyClassName(keyClassName);
- setValueClassName(valueClassName);
- }
-
- @InterfaceAudience.Private
- Builder() {
- Map<String, String> tezDefaults = ConfigUtils
- .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
- OnFileUnorderedKVOutput.getConfigurationKeySet());
- ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
- ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
- }
-
- @InterfaceAudience.Private
- Builder setKeyClassName(String keyClassName) {
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClassName);
- return this;
- }
-
- @InterfaceAudience.Private
- Builder setValueClassName(String valueClassName) {
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(String key, String value) {
- Preconditions.checkNotNull(key, "Key cannot be null");
- if (ConfigUtils.doesKeyQualify(key,
- Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
- TezRuntimeConfiguration.getAllowedPrefixes())) {
- if (value == null) {
- this.conf.unset(key);
- } else {
- this.conf.set(key, value);
- }
- }
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(Map<String, String> confMap) {
- Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
- Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- @Override
- public Builder setFromConfiguration(Configuration conf) {
- // Maybe ensure this is the first call ? Otherwise this can end up overriding other parameters
- Preconditions.checkArgument(conf != null, "Configuration cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
- Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for keys.
- *
- * @param serializationClassName
- * @return
- */
- public Builder setKeySerializationClass(String serializationClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @return
- */
- public Builder setValueSerializationClass(String serializationClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- return this;
- }
-
- public Builder enableCompression(String compressionCodec) {
- this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, true);
- if (compressionCodec != null) {
- this.conf
- .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
- }
- return this;
- }
-
- /**
- * Create the actual configuration instance.
- *
- * @return an instance of the Configuration
- */
- public OnFileUnorderedKVOutputConfiguration build() {
- return new OnFileUnorderedKVOutputConfiguration(this.conf);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfigurer.java
new file mode 100644
index 0000000..f84315f
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedKVOutputConfigurer.java
@@ -0,0 +1,257 @@
+/*
+ * *
+ * * 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.tez.runtime.library.conf;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.Map;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.ConfigUtils;
+import org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+/**
+ * Configure {@link org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput. </p>
+ *
+ * Values will be picked up from tez-site if not specified, otherwise defaults from
+ * {@link org.apache.tez.runtime.library.api.TezRuntimeConfiguration} will be used.
+ */
+public class OnFileUnorderedKVOutputConfigurer {
+ /**
+ * Configure parameters which are specific to the Output.
+ */
+ @InterfaceAudience.Private
+ public static interface SpecificConfigurer<T> extends BaseConfigurer<T> {
+ }
+
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class SpecificBuilder<E extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder> implements
+ SpecificConfigurer<SpecificBuilder> {
+
+ private final E edgeBuilder;
+ private final Builder builder;
+
+ SpecificBuilder(E edgeBuilder, Builder builder) {
+ this.edgeBuilder = edgeBuilder;
+ this.builder = builder;
+ }
+
+ @Override
+ public SpecificBuilder setAdditionalConfiguration(String key, String value) {
+ builder.setAdditionalConfiguration(key, value);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder setAdditionalConfiguration(Map<String, String> confMap) {
+ builder.setAdditionalConfiguration(confMap);
+ return this;
+ }
+
+ @Override
+ public SpecificBuilder setFromConfiguration(Configuration conf) {
+ builder.setFromConfiguration(conf);
+ return this;
+ }
+
+ public E done() {
+ return edgeBuilder;
+ }
+ }
+
+ @InterfaceAudience.Private
+ @VisibleForTesting
+ Configuration conf;
+
+ @InterfaceAudience.Private
+ @VisibleForTesting
+ OnFileUnorderedKVOutputConfigurer() {
+ }
+
+ private OnFileUnorderedKVOutputConfigurer(Configuration conf) {
+ this.conf = conf;
+ }
+
+ /**
+ * Get a byte array representation of the configuration
+ * @return a byte array which can be used as the payload
+ */
+ public byte[] toByteArray() {
+ try {
+ return TezUtils.createUserPayloadFromConf(conf);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @InterfaceAudience.Private
+ public void fromByteArray(byte[] payload) {
+ try {
+ this.conf = TezUtils.createConfFromUserPayload(payload);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static Builder newBuilder(String keyClass, String valClass) {
+ return new Builder(keyClass, valClass);
+ }
+
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class Builder implements SpecificConfigurer<Builder> {
+
+ private final Configuration conf = new Configuration(false);
+
+ /**
+ * Create a configuration builder for {@link org.apache.tez.runtime.library.output.OnFileUnorderedKVOutput}
+ *
+ * @param keyClassName the key class name
+ * @param valueClassName the value class name
+ */
+ @InterfaceAudience.Private
+ Builder(String keyClassName, String valueClassName) {
+ this();
+ Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
+ Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
+ setKeyClassName(keyClassName);
+ setValueClassName(valueClassName);
+ }
+
+ @InterfaceAudience.Private
+ Builder() {
+ Map<String, String> tezDefaults = ConfigUtils
+ .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
+ OnFileUnorderedKVOutput.getConfigurationKeySet());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
+ ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
+ }
+
+ @InterfaceAudience.Private
+ Builder setKeyClassName(String keyClassName) {
+ Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClassName);
+ return this;
+ }
+
+ @InterfaceAudience.Private
+ Builder setValueClassName(String valueClassName) {
+ Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(String key, String value) {
+ Preconditions.checkNotNull(key, "Key cannot be null");
+ if (ConfigUtils.doesKeyQualify(key,
+ Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
+ TezRuntimeConfiguration.getAllowedPrefixes())) {
+ if (value == null) {
+ this.conf.unset(key);
+ } else {
+ this.conf.set(key, value);
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(Map<String, String> confMap) {
+ Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
+ Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
+ Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, map);
+ return this;
+ }
+
+ @Override
+ public Builder setFromConfiguration(Configuration conf) {
+ // Maybe ensure this is the first call ? Otherwise this can end up overriding other parameters
+ Preconditions.checkArgument(conf != null, "Configuration cannot be null");
+ Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
+ Lists.newArrayList(OnFileUnorderedKVOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, map);
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for keys.
+ *
+ * @param serializationClassName
+ * @return
+ */
+ public Builder setKeySerializationClass(String serializationClassName) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for values.
+ *
+ * @param serializationClassName
+ * @return
+ */
+ public Builder setValueSerializationClass(String serializationClassName) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ return this;
+ }
+
+ public Builder setCompression(boolean enabled, @Nullable String compressionCodec) {
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, enabled);
+ if (enabled && compressionCodec != null) {
+ this.conf
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
+ }
+ return this;
+ }
+
+ /**
+ * Create the actual configuration instance.
+ *
+ * @return an instance of the Configuration
+ */
+ public OnFileUnorderedKVOutputConfigurer build() {
+ return new OnFileUnorderedKVOutputConfigurer(this.conf);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/675b2372/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedPartitionedKVOutputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedPartitionedKVOutputConfiguration.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedPartitionedKVOutputConfiguration.java
deleted file mode 100644
index 95a3a3c..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OnFileUnorderedPartitionedKVOutputConfiguration.java
+++ /dev/null
@@ -1,289 +0,0 @@
-/*
- * *
- * * 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.tez.runtime.library.conf;
-
-import java.io.IOException;
-import java.util.Map;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.tez.common.TezUtils;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.output.OnFileUnorderedPartitionedKVOutput;
-
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-public class OnFileUnorderedPartitionedKVOutputConfiguration {
- /**
- * Configure parameters which are specific to the Output.
- */
- @InterfaceAudience.Private
- public static interface SpecificConfigurer<T> extends BaseConfigurer<T> {
- /**
- * Set the buffer size to use
- *
- * @param availableBufferSize the size of the buffer in MB
- * @return instance of the current builder
- */
- public T setAvailableBufferSize(int availableBufferSize);
- }
-
- @SuppressWarnings("rawtypes")
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class SpecificBuilder<E extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder> implements
- SpecificConfigurer<SpecificBuilder> {
-
- private final E edgeBuilder;
- private final Builder builder;
-
- SpecificBuilder(E edgeBuilder, Builder builder) {
- this.edgeBuilder = edgeBuilder;
- this.builder = builder;
- }
-
- @Override
- public SpecificBuilder<E> setAvailableBufferSize(int availableBufferSize) {
- builder.setAvailableBufferSize(availableBufferSize);
- return this;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(String key, String value) {
- builder.setAdditionalConfiguration(key, value);
- return this;
- }
-
- @Override
- public SpecificBuilder setAdditionalConfiguration(Map<String, String> confMap) {
- builder.setAdditionalConfiguration(confMap);
- return this;
- }
-
- @Override
- public SpecificBuilder setFromConfiguration(Configuration conf) {
- builder.setFromConfiguration(conf);
- return this;
- }
-
- public E done() {
- return edgeBuilder;
- }
- }
-
- @InterfaceAudience.Private
- @VisibleForTesting
- Configuration conf;
-
- @InterfaceAudience.Private
- @VisibleForTesting
- OnFileUnorderedPartitionedKVOutputConfiguration() {
- }
-
- private OnFileUnorderedPartitionedKVOutputConfiguration(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Get a byte array representation of the configuration
- * @return a byte array which can be used as the payload
- */
- public byte[] toByteArray() {
- try {
- return TezUtils.createUserPayloadFromConf(conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @InterfaceAudience.Private
- public void fromByteArray(byte[] payload) {
- try {
- this.conf = TezUtils.createConfFromUserPayload(payload);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- public static Builder newBuilder(String keyClass, String valClass, String partitionerClassName,
- Configuration partitionerConf) {
- return new Builder(keyClass, valClass, partitionerClassName, partitionerConf);
- }
-
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class Builder implements SpecificConfigurer<Builder> {
-
- private final Configuration conf = new Configuration(false);
-
- /**
- * Create a configuration builder for {@link org.apache.tez.runtime.library.output.OnFileUnorderedPartitionedKVOutput}
- *
- * @param keyClassName the key class name
- * @param valueClassName the value class name
- * @param partitionerClassName the partitioner class name
- * @param partitionerConf configuration for the partitioner. This can be null
- */
- @InterfaceAudience.Private
- Builder(String keyClassName, String valueClassName, String partitionerClassName,
- Configuration partitionerConf) {
- this();
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
- setKeyClassName(keyClassName);
- setValueClassName(valueClassName);
- setPartitioner(partitionerClassName, partitionerConf);
- }
-
- @InterfaceAudience.Private
- Builder() {
- Map<String, String> tezDefaults = ConfigUtils
- .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
- OnFileUnorderedPartitionedKVOutput.getConfigurationKeySet());
- ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
- ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
- }
-
- @InterfaceAudience.Private
- Builder setKeyClassName(String keyClassName) {
- Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClassName);
- return this;
- }
-
- @InterfaceAudience.Private
- Builder setValueClassName(String valueClassName) {
- Preconditions.checkNotNull(valueClassName, "Value class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
- return this;
- }
-
- @InterfaceAudience.Private
- Builder setPartitioner(String partitionerClassName, Configuration partitionerConf) {
- Preconditions.checkNotNull(partitionerClassName, "Partitioner class name cannot be null");
- this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, partitionerClassName);
- if (partitionerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, partitionerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- @Override
- public Builder setAvailableBufferSize(int availableBufferSize) {
- this.conf
- .setInt(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB, availableBufferSize);
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(String key, String value) {
- Preconditions.checkNotNull(key, "Key cannot be null");
- if (ConfigUtils.doesKeyQualify(key,
- Lists.newArrayList(OnFileUnorderedPartitionedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
- TezRuntimeConfiguration.getAllowedPrefixes())) {
- if (value == null) {
- this.conf.unset(key);
- } else {
- this.conf.set(key, value);
- }
- }
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(Map<String, String> confMap) {
- Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
- Lists.newArrayList(OnFileUnorderedPartitionedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- @Override
- public Builder setFromConfiguration(Configuration conf) {
- // Maybe ensure this is the first call ? Otherwise this can end up overriding other parameters
- Preconditions.checkArgument(conf != null, "Configuration cannot be null");
- Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
- Lists.newArrayList(OnFileUnorderedPartitionedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- public Builder enableCompression(String compressionCodec) {
- this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, true);
- if (compressionCodec != null) {
- this.conf
- .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
- }
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for keys.
- *
- * @param serializationClassName
- * @return
- */
- public Builder setKeySerializationClass(String serializationClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @return
- */
- public Builder setValueSerializationClass(String serializationClassName) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- return this;
- }
-
- /**
- * Create the actual configuration instance.
- *
- * @return an instance of the Configuration
- */
- public OnFileUnorderedPartitionedKVOutputConfiguration build() {
- return new OnFileUnorderedPartitionedKVOutputConfiguration(this.conf);
- }
- }
-}