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/19 02:02:09 UTC
[3/6] TEZ-1417. Rename *Configurer to ConfigBuilder/Config. (sseth)
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfigurer.java
deleted file mode 100644
index 2ef4350..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfigurer.java
+++ /dev/null
@@ -1,283 +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 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.dag.api.UserPayload;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.output.UnorderedKVOutput;
-
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-/**
- * Configure {@link org.apache.tez.runtime.library.output.UnorderedKVOutput} </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 UnorderedKVOutputConfigurer {
- /**
- * 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
- UnorderedKVOutputConfigurer() {
- }
-
- private UnorderedKVOutputConfigurer(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Get a UserPayload representation of the Configuration
- * @return a {@link org.apache.tez.dag.api.UserPayload} instance
- */
- public UserPayload toUserPayload() {
- try {
- return TezUtils.createUserPayloadFromConf(conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @InterfaceAudience.Private
- public void fromUserPayload(UserPayload 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.UnorderedKVOutput}
- *
- * @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(),
- UnorderedKVOutput.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(UnorderedKVOutput.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(UnorderedKVOutput.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(UnorderedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for keys.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setKeySerializationClass(String serializationClassName,
- @Nullable Map<String, String> serializerConf) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- if (serializerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setValueSerializationClass(String serializationClassName,
- @Nullable Map<String, String> serializerConf) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- if (serializerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- public Builder setCompression(boolean enabled, @Nullable String compressionCodec,
- @Nullable Map<String, String> codecConf) {
- this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, enabled);
- if (enabled && compressionCodec != null) {
- this.conf
- .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
- }
- if (codecConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, codecConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- /**
- * Create the actual configuration instance.
- *
- * @return an instance of the Configuration
- */
- public UnorderedKVOutputConfigurer build() {
- return new UnorderedKVOutputConfigurer(this.conf);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
new file mode 100644
index 0000000..3216786
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
@@ -0,0 +1,265 @@
+/*
+ * *
+ * * 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.util.Map;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.OutputDescriptor;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.runtime.library.input.UnorderedKVInput;
+import org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput;
+
+/**
+ * Configure payloads for the UnorderedPartitionedKVOutput and UnorderedKVInput pair </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.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class UnorderedPartitionedKVEdgeConfig
+ extends HadoopKeyValuesBasedBaseEdgeConfig {
+
+ private final UnorderedPartitionedKVOutputConfig outputConf;
+ private final UnorderedKVInputConfig inputConf;
+
+ private UnorderedPartitionedKVEdgeConfig(
+ UnorderedPartitionedKVOutputConfig outputConfiguration,
+ UnorderedKVInputConfig inputConfiguration) {
+ this.outputConf = outputConfiguration;
+ this.inputConf = inputConfiguration;
+
+ }
+
+ /**
+ * Create a builder to configure the relevant Input and Output. </p> This method should only be
+ * used when using a custom Partitioner which requires specific Configuration. {@link
+ * #newBuilder(String, String, String)} is the preferred method to crate an instance of the
+ * Builder
+ *
+ * @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.
+ * @return a builder to configure the edge
+ */
+ public static Builder newBuilder(String keyClassName, String valueClassName,
+ String partitionerClassName,
+ @Nullable Map<String, String> partitionerConf) {
+ return new Builder(keyClassName, valueClassName, partitionerClassName, partitionerConf);
+ }
+
+ /**
+ * Create a builder to configure the relevant Input and Output
+ *
+ * @param keyClassName the key class name
+ * @param valueClassName the value class name
+ * @param partitionerClassName the partitioner class name
+ * @return a builder to configure the edge
+ */
+ public static Builder newBuilder(String keyClassName, String valueClassName,
+ String partitionerClassName) {
+ return newBuilder(keyClassName, valueClassName, partitionerClassName, null);
+ }
+
+ @Override
+ public UserPayload getOutputPayload() {
+ return outputConf.toUserPayload();
+ }
+
+ @Override
+ public String getOutputClassName() {
+ return UnorderedPartitionedKVOutput.class.getName();
+ }
+
+ @Override
+ public UserPayload getInputPayload() {
+ return inputConf.toUserPayload();
+ }
+
+ @Override
+ public String getInputClassName() {
+ return UnorderedKVInput.class.getName();
+ }
+
+ /**
+ * This is a convenience method for the typical usage of this edge, and creates an instance of
+ * {@link org.apache.tez.dag.api.EdgeProperty} which is likely to be used. </p>
+ * If custom edge properties are required, the methods to get the relevant payloads should be
+ * used. </p>
+ * * In this case - DataMovementType.SCATTER_GATHER, EdgeProperty.DataSourceType.PERSISTED,
+ * EdgeProperty.SchedulingType.SEQUENTIAL
+ *
+ * @return an {@link org.apache.tez.dag.api.EdgeProperty} instance
+ */
+ public EdgeProperty createDefaultEdgeProperty() {
+ EdgeProperty edgeProperty = new EdgeProperty(EdgeProperty.DataMovementType.SCATTER_GATHER,
+ EdgeProperty.DataSourceType.PERSISTED, EdgeProperty.SchedulingType.SEQUENTIAL,
+ new OutputDescriptor(
+ getOutputClassName()).setUserPayload(getOutputPayload()),
+ new InputDescriptor(
+ getInputClassName()).setUserPayload(getInputPayload()));
+ return edgeProperty;
+ }
+
+ /**
+ * This is a convenience method for creating an Edge descriptor based on the specified
+ * EdgeManagerDescriptor.
+ *
+ * @param edgeManagerDescriptor the custom edge specification
+ * @return an {@link org.apache.tez.dag.api.EdgeProperty} instance
+ */
+ public EdgeProperty createDefaultCustomEdgeProperty(EdgeManagerPluginDescriptor edgeManagerDescriptor) {
+ Preconditions.checkNotNull(edgeManagerDescriptor, "EdgeManagerDescriptor cannot be null");
+ EdgeProperty edgeProperty =
+ new EdgeProperty(edgeManagerDescriptor, EdgeProperty.DataSourceType.PERSISTED,
+ EdgeProperty.SchedulingType.SEQUENTIAL,
+ new OutputDescriptor(getOutputClassName()).setUserPayload(getOutputPayload()),
+ new InputDescriptor(getInputClassName()).setUserPayload(getInputPayload()));
+ return edgeProperty;
+ }
+
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class Builder extends HadoopKeyValuesBasedBaseEdgeConfig.Builder<Builder> {
+
+ private final UnorderedPartitionedKVOutputConfig.Builder outputBuilder =
+ new UnorderedPartitionedKVOutputConfig.Builder();
+ private final UnorderedPartitionedKVOutputConfig.SpecificBuilder<UnorderedPartitionedKVEdgeConfig.Builder>
+ specificOutputBuilder =
+ new UnorderedPartitionedKVOutputConfig.SpecificBuilder<UnorderedPartitionedKVEdgeConfig.Builder>(
+ this, outputBuilder);
+
+ private final UnorderedKVInputConfig.Builder inputBuilder =
+ new UnorderedKVInputConfig.Builder();
+ private final UnorderedKVInputConfig.SpecificBuilder<UnorderedPartitionedKVEdgeConfig.Builder>
+ specificInputBuilder =
+ new UnorderedKVInputConfig.SpecificBuilder<UnorderedPartitionedKVEdgeConfig.Builder>(
+ this, inputBuilder);
+
+ @InterfaceAudience.Private
+ Builder(String keyClassName, String valueClassName, String partitionerClassName,
+ Map<String, String> partitionerConf) {
+ outputBuilder.setKeyClassName(keyClassName);
+ outputBuilder.setValueClassName(valueClassName);
+ outputBuilder.setPartitioner(partitionerClassName, partitionerConf);
+ inputBuilder.setKeyClassName(keyClassName);
+ inputBuilder.setValueClassName(valueClassName);
+ }
+
+ @Override
+ public Builder setCompression(boolean enabled, @Nullable String compressionCodec, @Nullable Map<String, String> codecConf) {
+ outputBuilder.setCompression(enabled, compressionCodec, codecConf);
+ inputBuilder.setCompression(enabled, compressionCodec, codecConf);
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(String key, String value) {
+ outputBuilder.setAdditionalConfiguration(key, value);
+ inputBuilder.setAdditionalConfiguration(key, value);
+ return this;
+ }
+
+ @Override
+ public Builder setAdditionalConfiguration(Map<String, String> confMap) {
+ outputBuilder.setAdditionalConfiguration(confMap);
+ inputBuilder.setAdditionalConfiguration(confMap);
+ return this;
+ }
+
+ @Override
+ public Builder setFromConfiguration(Configuration conf) {
+ outputBuilder.setFromConfiguration(conf);
+ inputBuilder.setFromConfiguration(conf);
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for keys.
+ *
+ * @param serializationClassName
+ * @param serializerConf the serializer 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
+ */
+ public Builder setKeySerializationClass(String serializationClassName,
+ @Nullable Map<String, String> serializerConf) {
+ outputBuilder.setKeySerializationClass(serializationClassName, serializerConf);
+ inputBuilder.setKeySerializationClass(serializationClassName, serializerConf);
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for values.
+ *
+ * @param serializationClassName
+ * @param serializerConf the serializer 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
+ */
+ public Builder setValueSerializationClass(String serializationClassName, @Nullable Map<String, String> serializerConf) {
+ outputBuilder.setValueSerializationClass(serializationClassName, serializerConf);
+ inputBuilder.setValueSerializationClass(serializationClassName, serializerConf);
+ return this;
+ }
+
+ /**
+ * Configure the specific output
+ *
+ * @return a builder to configure the output
+ */
+ public UnorderedPartitionedKVOutputConfig.SpecificBuilder<Builder> configureOutput() {
+ return specificOutputBuilder;
+ }
+
+ /**
+ * Configure the specific input
+ * @return a builder to configure the input
+ */
+ public UnorderedKVInputConfig.SpecificBuilder<Builder> configureInput() {
+ return specificInputBuilder;
+ }
+
+ /**
+ * Build and return an instance of the configuration
+ * @return an instance of the acatual configuration
+ */
+ public UnorderedPartitionedKVEdgeConfig build() {
+ return new UnorderedPartitionedKVEdgeConfig(outputBuilder.build(), inputBuilder.build());
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfigurer.java
deleted file mode 100644
index e56a927..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfigurer.java
+++ /dev/null
@@ -1,264 +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 javax.annotation.Nullable;
-
-import java.util.Map;
-
-import com.google.common.base.Preconditions;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
-import org.apache.tez.dag.api.EdgeProperty;
-import org.apache.tez.dag.api.InputDescriptor;
-import org.apache.tez.dag.api.OutputDescriptor;
-import org.apache.tez.dag.api.UserPayload;
-import org.apache.tez.runtime.library.input.UnorderedKVInput;
-import org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput;
-
-/**
- * Configure payloads for the UnorderedPartitionedKVOutput and UnorderedKVInput pair </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.
- */
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-public class UnorderedPartitionedKVEdgeConfigurer extends HadoopKeyValuesBasedBaseEdgeConfigurer {
-
- private final UnorderedPartitionedKVOutputConfigurer outputConf;
- private final UnorderedKVInputConfigurer inputConf;
-
- private UnorderedPartitionedKVEdgeConfigurer(
- UnorderedPartitionedKVOutputConfigurer outputConfiguration,
- UnorderedKVInputConfigurer inputConfiguration) {
- this.outputConf = outputConfiguration;
- this.inputConf = inputConfiguration;
-
- }
-
- /**
- * Create a builder to configure the relevant Input and Output. </p> This method should only be
- * used when using a custom Partitioner which requires specific Configuration. {@link
- * #newBuilder(String, String, String)} is the preferred method to crate an instance of the
- * Builder
- *
- * @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.
- * @return a builder to configure the edge
- */
- public static Builder newBuilder(String keyClassName, String valueClassName,
- String partitionerClassName,
- @Nullable Map<String, String> partitionerConf) {
- return new Builder(keyClassName, valueClassName, partitionerClassName, partitionerConf);
- }
-
- /**
- * Create a builder to configure the relevant Input and Output
- *
- * @param keyClassName the key class name
- * @param valueClassName the value class name
- * @param partitionerClassName the partitioner class name
- * @return a builder to configure the edge
- */
- public static Builder newBuilder(String keyClassName, String valueClassName,
- String partitionerClassName) {
- return newBuilder(keyClassName, valueClassName, partitionerClassName, null);
- }
-
- @Override
- public UserPayload getOutputPayload() {
- return outputConf.toUserPayload();
- }
-
- @Override
- public String getOutputClassName() {
- return UnorderedPartitionedKVOutput.class.getName();
- }
-
- @Override
- public UserPayload getInputPayload() {
- return inputConf.toUserPayload();
- }
-
- @Override
- public String getInputClassName() {
- return UnorderedKVInput.class.getName();
- }
-
- /**
- * This is a convenience method for the typical usage of this edge, and creates an instance of
- * {@link org.apache.tez.dag.api.EdgeProperty} which is likely to be used. </p>
- * If custom edge properties are required, the methods to get the relevant payloads should be
- * used. </p>
- * * In this case - DataMovementType.SCATTER_GATHER, EdgeProperty.DataSourceType.PERSISTED,
- * EdgeProperty.SchedulingType.SEQUENTIAL
- *
- * @return an {@link org.apache.tez.dag.api.EdgeProperty} instance
- */
- public EdgeProperty createDefaultEdgeProperty() {
- EdgeProperty edgeProperty = new EdgeProperty(EdgeProperty.DataMovementType.SCATTER_GATHER,
- EdgeProperty.DataSourceType.PERSISTED, EdgeProperty.SchedulingType.SEQUENTIAL,
- new OutputDescriptor(
- getOutputClassName()).setUserPayload(getOutputPayload()),
- new InputDescriptor(
- getInputClassName()).setUserPayload(getInputPayload()));
- return edgeProperty;
- }
-
- /**
- * This is a convenience method for creating an Edge descriptor based on the specified
- * EdgeManagerDescriptor.
- *
- * @param edgeManagerDescriptor the custom edge specification
- * @return an {@link org.apache.tez.dag.api.EdgeProperty} instance
- */
- public EdgeProperty createDefaultCustomEdgeProperty(EdgeManagerPluginDescriptor edgeManagerDescriptor) {
- Preconditions.checkNotNull(edgeManagerDescriptor, "EdgeManagerDescriptor cannot be null");
- EdgeProperty edgeProperty =
- new EdgeProperty(edgeManagerDescriptor, EdgeProperty.DataSourceType.PERSISTED,
- EdgeProperty.SchedulingType.SEQUENTIAL,
- new OutputDescriptor(getOutputClassName()).setUserPayload(getOutputPayload()),
- new InputDescriptor(getInputClassName()).setUserPayload(getInputPayload()));
- return edgeProperty;
- }
-
- @InterfaceAudience.Public
- @InterfaceStability.Evolving
- public static class Builder extends HadoopKeyValuesBasedBaseEdgeConfigurer.Builder<Builder> {
-
- private final UnorderedPartitionedKVOutputConfigurer.Builder outputBuilder =
- new UnorderedPartitionedKVOutputConfigurer.Builder();
- private final UnorderedPartitionedKVOutputConfigurer.SpecificBuilder<UnorderedPartitionedKVEdgeConfigurer.Builder>
- specificOutputBuilder =
- new UnorderedPartitionedKVOutputConfigurer.SpecificBuilder<UnorderedPartitionedKVEdgeConfigurer.Builder>(
- this, outputBuilder);
-
- private final UnorderedKVInputConfigurer.Builder inputBuilder =
- new UnorderedKVInputConfigurer.Builder();
- private final UnorderedKVInputConfigurer.SpecificBuilder<UnorderedPartitionedKVEdgeConfigurer.Builder>
- specificInputBuilder =
- new UnorderedKVInputConfigurer.SpecificBuilder<UnorderedPartitionedKVEdgeConfigurer.Builder>(
- this, inputBuilder);
-
- @InterfaceAudience.Private
- Builder(String keyClassName, String valueClassName, String partitionerClassName,
- Map<String, String> partitionerConf) {
- outputBuilder.setKeyClassName(keyClassName);
- outputBuilder.setValueClassName(valueClassName);
- outputBuilder.setPartitioner(partitionerClassName, partitionerConf);
- inputBuilder.setKeyClassName(keyClassName);
- inputBuilder.setValueClassName(valueClassName);
- }
-
- @Override
- public Builder setCompression(boolean enabled, @Nullable String compressionCodec, @Nullable Map<String, String> codecConf) {
- outputBuilder.setCompression(enabled, compressionCodec, codecConf);
- inputBuilder.setCompression(enabled, compressionCodec, codecConf);
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(String key, String value) {
- outputBuilder.setAdditionalConfiguration(key, value);
- inputBuilder.setAdditionalConfiguration(key, value);
- return this;
- }
-
- @Override
- public Builder setAdditionalConfiguration(Map<String, String> confMap) {
- outputBuilder.setAdditionalConfiguration(confMap);
- inputBuilder.setAdditionalConfiguration(confMap);
- return this;
- }
-
- @Override
- public Builder setFromConfiguration(Configuration conf) {
- outputBuilder.setFromConfiguration(conf);
- inputBuilder.setFromConfiguration(conf);
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for keys.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setKeySerializationClass(String serializationClassName,
- @Nullable Map<String, String> serializerConf) {
- outputBuilder.setKeySerializationClass(serializationClassName, serializerConf);
- inputBuilder.setKeySerializationClass(serializationClassName, serializerConf);
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setValueSerializationClass(String serializationClassName, @Nullable Map<String, String> serializerConf) {
- outputBuilder.setValueSerializationClass(serializationClassName, serializerConf);
- inputBuilder.setValueSerializationClass(serializationClassName, serializerConf);
- return this;
- }
-
- /**
- * Configure the specific output
- *
- * @return a builder to configure the output
- */
- public UnorderedPartitionedKVOutputConfigurer.SpecificBuilder<Builder> configureOutput() {
- return specificOutputBuilder;
- }
-
- /**
- * Configure the specific input
- * @return a builder to configure the input
- */
- public UnorderedKVInputConfigurer.SpecificBuilder<Builder> configureInput() {
- return specificInputBuilder;
- }
-
- /**
- * Build and return an instance of the configuration
- * @return an instance of the acatual configuration
- */
- public UnorderedPartitionedKVEdgeConfigurer build() {
- return new UnorderedPartitionedKVEdgeConfigurer(outputBuilder.build(), inputBuilder.build());
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
new file mode 100644
index 0000000..fbadc58
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
@@ -0,0 +1,327 @@
+/*
+ * *
+ * * 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.dag.api.UserPayload;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.ConfigUtils;
+import org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+/**
+ * Configure {@link org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput} </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 UnorderedPartitionedKVOutputConfig {
+ /**
+ * Configure parameters which are specific to the Output.
+ */
+ @InterfaceAudience.Private
+ public static interface SpecificConfigBuilder<T> extends BaseConfigBuilder<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 HadoopKeyValuesBasedBaseEdgeConfig.Builder> implements
+ SpecificConfigBuilder<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
+ UnorderedPartitionedKVOutputConfig() {
+ }
+
+ private UnorderedPartitionedKVOutputConfig(Configuration conf) {
+ this.conf = conf;
+ }
+
+ /**
+ * Get a UserPayload representation of the Configuration
+ * @return a {@link org.apache.tez.dag.api.UserPayload} instance
+ */
+ public UserPayload toUserPayload() {
+ try {
+ return TezUtils.createUserPayloadFromConf(conf);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @InterfaceAudience.Private
+ public void fromUserPayload(UserPayload payload) {
+ try {
+ this.conf = TezUtils.createConfFromUserPayload(payload);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ public static Builder newBuilder(String keyClass, String valClass, String partitionerClassName) {
+ return newBuilder(keyClass, valClass, partitionerClassName, null);
+ }
+
+ public static Builder newBuilder(String keyClass, String valClass, String partitionerClassName,
+ Map<String, String> partitionerConf) {
+ return new Builder(keyClass, valClass, partitionerClassName, partitionerConf);
+ }
+
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class Builder implements SpecificConfigBuilder<Builder> {
+
+ private final Configuration conf = new Configuration(false);
+
+ /**
+ * Create a configuration builder for {@link org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput}
+ *
+ * @param keyClassName the key class name
+ * @param valueClassName the value class name
+ * @param partitionerClassName the partitioner class name
+ * @param partitionerConf configuration for the partitioner specified as a map of key-value
+ * pairs. This can be null
+ */
+ @InterfaceAudience.Private
+ Builder(String keyClassName, String valueClassName, String partitionerClassName,
+ 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(),
+ UnorderedPartitionedKVOutput.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, 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 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(UnorderedPartitionedKVOutput.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(UnorderedPartitionedKVOutput.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(UnorderedPartitionedKVOutput.getConfigurationKeySet(),
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, map);
+ return this;
+ }
+
+ public Builder setCompression(boolean enabled, @Nullable String compressionCodec,
+ @Nullable Map<String, String> codecConf) {
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, enabled);
+ if (enabled && compressionCodec != null) {
+ this.conf
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
+ }
+ if (codecConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, codecConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ }
+ return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for keys.
+ *
+ * @param serializationClassName
+ * @param serializerConf the serializer 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
+ */
+ public Builder setKeySerializationClass(String serializationClassName,
+ @Nullable Map<String, String> serializerConf) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ if (serializerConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ } return this;
+ }
+
+ /**
+ * Set serialization class responsible for providing serializer/deserializer for values.
+ *
+ * @param serializationClassName
+ * @param serializerConf the serializer 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
+ */
+ public Builder setValueSerializationClass(String serializationClassName,
+ @Nullable Map<String, String> serializerConf) {
+ Preconditions.checkArgument(serializationClassName != null,
+ "serializationClassName cannot be null");
+ this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
+ + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ if (serializerConf != null) {
+ // Merging the confs for now. Change to be specific in the future.
+ ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
+ }
+ return this;
+ }
+
+ /**
+ * Create the actual configuration instance.
+ *
+ * @return an instance of the Configuration
+ */
+ public UnorderedPartitionedKVOutputConfig build() {
+ return new UnorderedPartitionedKVOutputConfig(this.conf);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfigurer.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfigurer.java
deleted file mode 100644
index 9417fbd..0000000
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfigurer.java
+++ /dev/null
@@ -1,327 +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 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.dag.api.UserPayload;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput;
-
-@InterfaceAudience.Public
-@InterfaceStability.Evolving
-/**
- * Configure {@link org.apache.tez.runtime.library.output.UnorderedPartitionedKVOutput} </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 UnorderedPartitionedKVOutputConfigurer {
- /**
- * 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
- UnorderedPartitionedKVOutputConfigurer() {
- }
-
- private UnorderedPartitionedKVOutputConfigurer(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Get a UserPayload representation of the Configuration
- * @return a {@link org.apache.tez.dag.api.UserPayload} instance
- */
- public UserPayload toUserPayload() {
- try {
- return TezUtils.createUserPayloadFromConf(conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @InterfaceAudience.Private
- public void fromUserPayload(UserPayload payload) {
- try {
- this.conf = TezUtils.createConfFromUserPayload(payload);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
-
- public static Builder newBuilder(String keyClass, String valClass, String partitionerClassName) {
- return newBuilder(keyClass, valClass, partitionerClassName, null);
- }
-
- public static Builder newBuilder(String keyClass, String valClass, String partitionerClassName,
- Map<String, String> 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.UnorderedPartitionedKVOutput}
- *
- * @param keyClassName the key class name
- * @param valueClassName the value class name
- * @param partitionerClassName the partitioner class name
- * @param partitionerConf configuration for the partitioner specified as a map of key-value
- * pairs. This can be null
- */
- @InterfaceAudience.Private
- Builder(String keyClassName, String valueClassName, String partitionerClassName,
- 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(),
- UnorderedPartitionedKVOutput.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, 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 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(UnorderedPartitionedKVOutput.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(UnorderedPartitionedKVOutput.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(UnorderedPartitionedKVOutput.getConfigurationKeySet(),
- TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
- ConfigUtils.addConfigMapToConfiguration(this.conf, map);
- return this;
- }
-
- public Builder setCompression(boolean enabled, @Nullable String compressionCodec,
- @Nullable Map<String, String> codecConf) {
- this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, enabled);
- if (enabled && compressionCodec != null) {
- this.conf
- .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
- }
- if (codecConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, codecConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for keys.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setKeySerializationClass(String serializationClassName,
- @Nullable Map<String, String> serializerConf) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- if (serializerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- } return this;
- }
-
- /**
- * Set serialization class responsible for providing serializer/deserializer for values.
- *
- * @param serializationClassName
- * @param serializerConf the serializer 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
- */
- public Builder setValueSerializationClass(String serializationClassName,
- @Nullable Map<String, String> serializerConf) {
- Preconditions.checkArgument(serializationClassName != null,
- "serializationClassName cannot be null");
- this.conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, serializationClassName + ","
- + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
- if (serializerConf != null) {
- // Merging the confs for now. Change to be specific in the future.
- ConfigUtils.mergeConfsWithExclusions(this.conf, serializerConf,
- TezRuntimeConfiguration.getRuntimeConfigKeySet());
- }
- return this;
- }
-
- /**
- * Create the actual configuration instance.
- *
- * @return an instance of the Configuration
- */
- public UnorderedPartitionedKVOutputConfigurer build() {
- return new UnorderedPartitionedKVOutputConfigurer(this.conf);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestShuffleVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestShuffleVertexManager.java b/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestShuffleVertexManager.java
index 970cd9d..4290cd7 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestShuffleVertexManager.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestShuffleVertexManager.java
@@ -101,8 +101,8 @@ public class TestShuffleVertexManager {
when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(4);
//Check via setters
- ShuffleVertexManager.ShuffleVertexManagerConfigurer configurer = ShuffleVertexManager
- .createConfigurer(null);
+ ShuffleVertexManager.ShuffleVertexManagerConfigBuilder configurer = ShuffleVertexManager
+ .createConfigBuilder(null);
VertexManagerPluginDescriptor pluginDesc = configurer.setAutoReduceParallelism(true)
.setDesiredTaskInputSize(1000l)
.setMinTaskParallelism(10).setSlowStartMaxSrcCompletionFraction(0.5f).build();
@@ -119,7 +119,7 @@ public class TestShuffleVertexManager {
Assert.assertTrue(manager.slowStartMinSrcCompletionFraction == 0.25f);
Assert.assertTrue(manager.slowStartMaxSrcCompletionFraction == 0.5f);
- configurer = ShuffleVertexManager.createConfigurer(null);
+ configurer = ShuffleVertexManager.createConfigBuilder(null);
pluginDesc = configurer.setAutoReduceParallelism(false).build();
when(mockContext.getUserPayload()).thenReturn(pluginDesc.getUserPayload());
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfigurer.java
deleted file mode 100644
index f0a6bf8..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfigurer.java
+++ /dev/null
@@ -1,190 +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 static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.Maps;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestOnFileSortedOutputConfigurer {
-
- @Test
- public void testNullParams() {
- try {
- OrderedPartitionedKVOutputConfigurer.newBuilder(
- null, "VALUE", "PARTITIONER", null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
-
- try {
- OrderedPartitionedKVOutputConfigurer.newBuilder(
- "KEY", null, "PARTITIONER", null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
-
- try {
- OrderedPartitionedKVOutputConfigurer.newBuilder(
- "KEY", "VALUE", null, null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
- }
-
- @Test
- public void testSetters() {
- Configuration fromConf = new Configuration(false);
- fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
- fromConf.set("fs.shouldExist", "fs");
- Map<String, String> additionalConf = new HashMap<String, String>();
- additionalConf.put("test.key.2", "key2");
- additionalConf.put("io.shouldExist", "io");
- additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "TestInternalSorter");
- OrderedPartitionedKVOutputConfigurer.Builder builder =
- OrderedPartitionedKVOutputConfigurer.newBuilder("KEY", "VALUE", "PARTITIONER", null)
- .setKeyComparatorClass("KEY_COMPARATOR", null)
- .setCompression(true, "CustomCodec", null)
- .setSortBufferSize(2048)
- .setAdditionalConfiguration("test.key.1", "key1")
- .setAdditionalConfiguration("file.shouldExist", "file")
- .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- String.valueOf(false))
- .setAdditionalConfiguration(additionalConf)
- .setFromConfiguration(fromConf);
-
- OrderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- OrderedPartitionedKVOutputConfigurer rebuilt = new OrderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Verify programmatic API usage
- assertEquals(2048, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
- assertEquals("CustomCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
- false));
- assertEquals("KEY_COMPARATOR", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
-
- // Verify additional configs
- assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
- assertEquals("TestInternalSorter",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, ""));
- assertEquals("io", conf.get("io.shouldExist"));
- assertEquals("file", conf.get("file.shouldExist"));
- assertEquals("fs", conf.get("fs.shouldExist"));
- assertNull(conf.get("test.conf.key.1"));
- assertNull(conf.get("test.key.1"));
- assertNull(conf.get("test.key.2"));
- }
-
- @Test
- public void testDefaultConfigsUsed() {
- OrderedPartitionedKVOutputConfigurer.Builder builder =
- OrderedPartitionedKVOutputConfigurer.newBuilder("KEY", "VALUE", "PARTITIONER", null);
- OrderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- OrderedPartitionedKVOutputConfigurer rebuilt = new OrderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-
- // Property present
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- // Verify whatever was configured
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
- }
-
- @Test
- public void testPartitionerConfigs() {
- Map<String, String> partitionerConf = Maps.newHashMap();
- partitionerConf.put("partitioner.test.key", "PARTITIONERKEY");
- partitionerConf
- .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
- OrderedPartitionedKVOutputConfigurer.Builder builder =
- OrderedPartitionedKVOutputConfigurer.newBuilder("KEY", "VALUE", "PARTITIONER", partitionerConf);
-
- OrderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- OrderedPartitionedKVOutputConfigurer rebuilt = new OrderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Default Output property should not be overridden based on partitioner config
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- assertEquals("PARTITIONERKEY", conf.get("partitioner.test.key"));
- }
-
- @Test
- public void testCombinerConfigs() {
- Map<String, String> combinerConf = Maps.newHashMap();
- combinerConf.put("combiner.test.key", "COMBINERKEY");
- combinerConf
- .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
- OrderedPartitionedKVOutputConfigurer.Builder builder =
- OrderedPartitionedKVOutputConfigurer.newBuilder("KEY", "VALUE", "PARTITIONER", null)
- .setCombiner("COMBINER", combinerConf);
-
- OrderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- OrderedPartitionedKVOutputConfigurer rebuilt = new OrderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Default Output property should not be overridden based on partitioner config
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- assertEquals("COMBINERKEY", conf.get("combiner.test.key"));
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfigurer.java
deleted file mode 100644
index bd182ec..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfigurer.java
+++ /dev/null
@@ -1,129 +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 static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestOnFileUnorderedKVOutputConfigurer {
-
- @Test
- public void testNullParams() {
- try {
- UnorderedKVOutputConfigurer.newBuilder(
- null, "VALUE");
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
-
- try {
- UnorderedKVOutputConfigurer.newBuilder(
- "KEY", null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
- }
-
- @Test
- public void testSetters() {
- Configuration fromConf = new Configuration(false);
- fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
- fromConf.set("io.shouldExist", "io");
- Map<String, String> additionalConf = new HashMap<String, String>();
- additionalConf.put("test.key.2", "key2");
- additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
- additionalConf.put("file.shouldExist", "file");
- UnorderedKVOutputConfigurer.Builder builder =
- UnorderedKVOutputConfigurer.newBuilder("KEY", "VALUE")
- .setCompression(true, "CustomCodec", null)
- .setAdditionalConfiguration("fs.shouldExist", "fs")
- .setAdditionalConfiguration("test.key.1", "key1")
- .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- String.valueOf(false))
- .setAdditionalConfiguration(additionalConf)
- .setFromConfiguration(fromConf);
-
- UnorderedKVOutputConfigurer configuration = builder.build();
-
- UnorderedKVOutputConfigurer rebuilt =
- new UnorderedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Verify programmatic API usage
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("CustomCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
- false));
-
- // Verify additional configs
- assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
- assertEquals("io", conf.get("io.shouldExist"));
- assertEquals("file", conf.get("file.shouldExist"));
- assertEquals("fs", conf.get("fs.shouldExist"));
- assertNull(conf.get("test.conf.key.1"));
- assertNull(conf.get("test.key.1"));
- assertNull(conf.get("test.key.2"));
- }
-
- @Test
- public void testDefaultConfigsUsed() {
- UnorderedKVOutputConfigurer.Builder builder =
- UnorderedKVOutputConfigurer
- .newBuilder("KEY", "VALUE");
- UnorderedKVOutputConfigurer configuration = builder.build();
-
- UnorderedKVOutputConfigurer rebuilt =
- new UnorderedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-
- // Default property present.
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- // Verify whatever was configured
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutputConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutputConfigurer.java
deleted file mode 100644
index 0ce96eb..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutputConfigurer.java
+++ /dev/null
@@ -1,177 +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 static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.Maps;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestOnFileUnorderedPartitionedKVOutputConfigurer {
-
- @Test
- public void testNullParams() {
- try {
- UnorderedPartitionedKVOutputConfigurer.newBuilder(
- null, "VALUE", "PARTITIONER", null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
-
- try {
- UnorderedPartitionedKVOutputConfigurer.newBuilder(
- "KEY", null, "PARTITIONER", null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
-
- try {
- UnorderedPartitionedKVOutputConfigurer.newBuilder(
- "KEY", "VALUE", null, null);
- fail("Expecting a null parameter list to fail");
- } catch (NullPointerException npe) {
- assertTrue(npe.getMessage().contains("cannot be null"));
- }
- }
-
- @Test
- public void testSetters() {
- Configuration fromConf = new Configuration(false);
- fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
- fromConf.set("io.shouldExist", "io");
- Map<String, String> additionalConf = new HashMap<String, String>();
- additionalConf.put("test.key.2", "key2");
- additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
- additionalConf.put("file.shouldExist", "file");
- UnorderedPartitionedKVOutputConfigurer.Builder builder =
- UnorderedPartitionedKVOutputConfigurer.newBuilder("KEY", "VALUE", "PARTITIONER",
- null)
- .setCompression(true, "CustomCodec", null)
- .setAvailableBufferSize(1111)
- .setAdditionalConfiguration("fs.shouldExist", "fs")
- .setAdditionalConfiguration("test.key.1", "key1")
- .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- String.valueOf(false))
- .setAdditionalConfiguration(additionalConf)
- .setFromConfiguration(fromConf);
-
- UnorderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- UnorderedPartitionedKVOutputConfigurer rebuilt =
- new UnorderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Verify programmatic API usage
- assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB, 0));
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
- assertEquals("CustomCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
- false));
-
- // Verify additional configs
- assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
- assertEquals(2222,
- conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, 0));
- assertEquals("io", conf.get("io.shouldExist"));
- assertEquals("file", conf.get("file.shouldExist"));
- assertEquals("fs", conf.get("fs.shouldExist"));
- assertNull(conf.get("test.conf.key.1"));
- assertNull(conf.get("test.key.1"));
- assertNull(conf.get("test.key.2"));
- }
-
- @Test
- public void testDefaultConfigsUsed() {
- UnorderedPartitionedKVOutputConfigurer.Builder builder =
- UnorderedPartitionedKVOutputConfigurer
- .newBuilder("KEY", "VALUE", "PARTITIONER", null)
- .setKeySerializationClass("SerClass1", null)
- .setValueSerializationClass("SerClass2", null);
- UnorderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- UnorderedPartitionedKVOutputConfigurer rebuilt =
- new UnorderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-
- // Default property present.
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- // Verify whatever was configured
- assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
- assertTrue(conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).startsWith("SerClass2," +
- "SerClass1"));
- //for unordered paritioned kv output, comparator is not populated
- assertNull(conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
- }
-
- @Test
- public void testPartitionerConfigs() {
- Map<String, String> partitionerConf = Maps.newHashMap();
- partitionerConf.put("partitioner.test.key", "PARTITIONERKEY");
- partitionerConf
- .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
- UnorderedPartitionedKVOutputConfigurer.Builder builder =
- UnorderedPartitionedKVOutputConfigurer
- .newBuilder("KEY", "VALUE", "PARTITIONER", partitionerConf);
-
- UnorderedPartitionedKVOutputConfigurer configuration = builder.build();
-
- UnorderedPartitionedKVOutputConfigurer rebuilt =
- new UnorderedPartitionedKVOutputConfigurer();
- rebuilt.fromUserPayload(configuration.toUserPayload());
-
- Configuration conf = rebuilt.conf;
-
- // Default Output property should not be overridden based on partitioner config
- assertEquals("TestCodec",
- conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
- assertEquals("PARTITIONERKEY", conf.get("partitioner.test.key"));
- }
-}