You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2014/07/24 19:42:09 UTC
[2/4] TEZ-1137. Move TezJobConfig to runtime-library and rename to
TezRuntimeConfiguration (bikas)
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledMergedInputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledMergedInputConfiguration.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledMergedInputConfiguration.java
index 92fba8a..1a5a6ed 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledMergedInputConfiguration.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledMergedInputConfiguration.java
@@ -26,11 +26,12 @@ 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.tez.common.TezJobConfig;
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.input.ShuffledMergedInput;
import org.apache.tez.runtime.library.input.ShuffledMergedInputLegacy;
@@ -275,23 +276,23 @@ public class ShuffledMergedInputConfiguration {
@InterfaceAudience.Private
Builder() {
Map<String, String> tezDefaults = ConfigUtils
- .extractConfigurationMap(TezJobConfig.getTezRuntimeConfigDefaults(),
+ .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
ShuffledMergedInput.getConfigurationKeySet());
ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
- ConfigUtils.addConfigMapToConfiguration(this.conf, TezJobConfig.getOtherConfigDefaults());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
}
@InterfaceAudience.Private
Builder setKeyClassName(String keyClassName) {
Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- this.conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, keyClassName);
+ 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(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
return this;
}
@@ -303,48 +304,48 @@ public class ShuffledMergedInputConfiguration {
@Override
public Builder setShuffleBufferFraction(float shuffleBufferFraction) {
this.conf
- .setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, shuffleBufferFraction);
+ .setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, shuffleBufferFraction);
return this;
}
@Override
public Builder setPostMergeBufferFraction(float postMergeBufferFraction) {
- this.conf.setFloat(TezJobConfig.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, postMergeBufferFraction);
+ this.conf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, postMergeBufferFraction);
return this;
}
@Override
public Builder setMaxSingleMemorySegmentFraction(float maxSingleSegmentFraction) {
- this.conf.setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
+ this.conf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
maxSingleSegmentFraction);
return this;
}
@Override
public Builder enableMemToMemMerger() {
- this.conf.setBoolean(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM, true);
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM, true);
return this;
}
@Override
public Builder setMergeFraction(float mergeFraction) {
- this.conf.setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, mergeFraction);
+ this.conf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, mergeFraction);
return this;
}
@Override
public Builder enableEncryptedTransfer() {
- this.conf.setBoolean(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL, true);
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL, true);
return this;
}
@Override
public Builder setCombiner(String combinerClassName, Configuration combinerConf) {
- this.conf.set(TezJobConfig.TEZ_RUNTIME_COMBINER_CLASS, combinerClassName);
+ 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,
- TezJobConfig.getRuntimeConfigKeySet());
+ TezRuntimeConfiguration.getRuntimeConfigKeySet());
}
return this;
}
@@ -356,7 +357,7 @@ public class ShuffledMergedInputConfiguration {
* @return instance of the current builder
*/
public Builder setKeyComparatorClass(String comparatorClassName) {
- this.conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
comparatorClassName);
return this;
}
@@ -366,8 +367,8 @@ public class ShuffledMergedInputConfiguration {
Preconditions.checkNotNull(key, "Key cannot be null");
if (ConfigUtils.doesKeyQualify(key,
Lists.newArrayList(ShuffledMergedInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()),
- TezJobConfig.getAllowedPrefixes())) {
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
+ TezRuntimeConfiguration.getAllowedPrefixes())) {
if (value == null) {
this.conf.unset(key);
} else {
@@ -382,7 +383,7 @@ public class ShuffledMergedInputConfiguration {
Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
Lists.newArrayList(ShuffledMergedInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()), TezJobConfig.getAllowedPrefixes());
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
ConfigUtils.addConfigMapToConfiguration(this.conf, map);
return this;
}
@@ -393,16 +394,16 @@ public class ShuffledMergedInputConfiguration {
Preconditions.checkArgument(conf != null, "Configuration cannot be null");
Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
Lists.newArrayList(ShuffledMergedInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()), TezJobConfig.getAllowedPrefixes());
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
ConfigUtils.addConfigMapToConfiguration(this.conf, map);
return this;
}
public Builder enableCompression(String compressionCodec) {
- this.conf.setBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS, true);
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, true);
if (compressionCodec != null) {
this.conf
- .set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
}
return this;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledUnorderedKVInputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledUnorderedKVInputConfiguration.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledUnorderedKVInputConfiguration.java
index 0a1e7d3..8452b36 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledUnorderedKVInputConfiguration.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/ShuffledUnorderedKVInputConfiguration.java
@@ -26,11 +26,12 @@ 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.tez.common.TezJobConfig;
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.input.ShuffledUnorderedKVInput;
@@ -208,49 +209,49 @@ public class ShuffledUnorderedKVInputConfiguration {
@InterfaceAudience.Private
Builder() {
Map<String, String> tezDefaults = ConfigUtils
- .extractConfigurationMap(TezJobConfig.getTezRuntimeConfigDefaults(),
+ .extractConfigurationMap(TezRuntimeConfiguration.getTezRuntimeConfigDefaults(),
ShuffledUnorderedKVInput.getConfigurationKeySet());
ConfigUtils.addConfigMapToConfiguration(this.conf, tezDefaults);
- ConfigUtils.addConfigMapToConfiguration(this.conf, TezJobConfig.getOtherConfigDefaults());
+ ConfigUtils.addConfigMapToConfiguration(this.conf, TezRuntimeConfiguration.getOtherConfigDefaults());
}
@InterfaceAudience.Private
Builder setKeyClassName(String keyClassName) {
Preconditions.checkNotNull(keyClassName, "Key class name cannot be null");
- this.conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, keyClassName);
+ 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(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
+ this.conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valueClassName);
return this;
}
@Override
public Builder setShuffleBufferFraction(float shuffleBufferFraction) {
this.conf
- .setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, shuffleBufferFraction);
+ .setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, shuffleBufferFraction);
return this;
}
@Override
public Builder setMaxSingleMemorySegmentFraction(float maxSingleSegmentFraction) {
- this.conf.setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
+ this.conf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
maxSingleSegmentFraction);
return this;
}
@Override
public Builder setMergeFraction(float mergeFraction) {
- this.conf.setFloat(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, mergeFraction);
+ this.conf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, mergeFraction);
return this;
}
@Override
public Builder enableEncryptedTransfer() {
- this.conf.setBoolean(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL, true);
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL, true);
return this;
}
@@ -259,8 +260,8 @@ public class ShuffledUnorderedKVInputConfiguration {
Preconditions.checkNotNull(key, "Key cannot be null");
if (ConfigUtils.doesKeyQualify(key,
Lists.newArrayList(ShuffledUnorderedKVInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()),
- TezJobConfig.getAllowedPrefixes())) {
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()),
+ TezRuntimeConfiguration.getAllowedPrefixes())) {
if (value == null) {
this.conf.unset(key);
} else {
@@ -275,7 +276,7 @@ public class ShuffledUnorderedKVInputConfiguration {
Preconditions.checkNotNull(confMap, "ConfMap cannot be null");
Map<String, String> map = ConfigUtils.extractConfigurationMap(confMap,
Lists.newArrayList(ShuffledUnorderedKVInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()), TezJobConfig.getAllowedPrefixes());
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
ConfigUtils.addConfigMapToConfiguration(this.conf, map);
return this;
}
@@ -286,16 +287,16 @@ public class ShuffledUnorderedKVInputConfiguration {
Preconditions.checkArgument(conf != null, "Configuration cannot be null");
Map<String, String> map = ConfigUtils.extractConfigurationMap(conf,
Lists.newArrayList(ShuffledUnorderedKVInput.getConfigurationKeySet(),
- TezJobConfig.getRuntimeAdditionalConfigKeySet()), TezJobConfig.getAllowedPrefixes());
+ TezRuntimeConfiguration.getRuntimeAdditionalConfigKeySet()), TezRuntimeConfiguration.getAllowedPrefixes());
ConfigUtils.addConfigMapToConfiguration(this.conf, map);
return this;
}
public Builder enableCompression(String compressionCodec) {
- this.conf.setBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS, true);
+ this.conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, true);
if (compressionCodec != null) {
this.conf
- .set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, compressionCodec);
}
return this;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledMergedInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledMergedInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledMergedInput.java
index e53383e..6c50b93 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledMergedInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledMergedInput.java
@@ -18,7 +18,11 @@
package org.apache.tez.runtime.library.input;
import java.io.IOException;
-import java.util.*;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -28,14 +32,15 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.RawComparator;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.AbstractLogicalInput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.ConfigUtils;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.ValuesIterator;
@@ -288,39 +293,39 @@ public class ShuffledMergedInput extends AbstractLogicalInput {
private static final Set<String> confKeys = new HashSet<String>();
static {
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMBINER_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTERS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUP_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUPS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_INPUT_BUFFER_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUP_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUPS_MAX_KEYS);
}
// TODO Maybe add helper methods to extract keys
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledUnorderedKVInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledUnorderedKVInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledUnorderedKVInput.java
index d445398..e2d73e6 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledUnorderedKVInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/ShuffledUnorderedKVInput.java
@@ -31,14 +31,15 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.AbstractLogicalInput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.ConfigUtils;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.readers.ShuffledUnorderedKVReader;
@@ -108,17 +109,17 @@ public class ShuffledUnorderedKVInput extends AbstractLogicalInput {
codec = null;
}
- boolean ifileReadAhead = conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT);
+ boolean ifileReadAhead = conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT);
int ifileReadAheadLength = 0;
int ifileBufferSize = 0;
if (ifileReadAhead) {
- ifileReadAheadLength = conf.getInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT);
+ ifileReadAheadLength = conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT);
}
ifileBufferSize = conf.getInt("io.file.buffer.size",
- TezJobConfig.TEZ_RUNTIME_IFILE_BUFFER_SIZE_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_BUFFER_SIZE_DEFAULT);
this.inputManager = new SimpleFetchedInputAllocator(getContext().getUniqueIdentifier(), conf,
getContext().getTotalMemoryAvailableToTask(),
@@ -215,32 +216,32 @@ public class ShuffledUnorderedKVInput extends AbstractLogicalInput {
private static final Set<String> confKeys = new HashSet<String>();
static {
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTERS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUP_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUPS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_INPUT_BUFFER_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUP_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUPS_MAX_KEYS);
}
// TODO Maybe add helper methods to extract keys
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
index c1f5ddb..240d2d6 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.protobuf.ByteString;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -36,16 +37,17 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.AbstractLogicalOutput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.api.events.VertexManagerEvent;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.api.KeyValuesWriter;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.sort.impl.ExternalSorter;
import org.apache.tez.runtime.library.common.sort.impl.PipelinedSorter;
@@ -86,8 +88,8 @@ public class OnFileSortedOutput extends AbstractLogicalOutput {
getContext().getTotalMemoryAvailableToTask()), memoryUpdateCallbackHandler);
sendEmptyPartitionDetails = this.conf.getBoolean(
- TezJobConfig.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED,
- TezJobConfig.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED,
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT);
return Collections.emptyList();
}
@@ -95,8 +97,8 @@ public class OnFileSortedOutput extends AbstractLogicalOutput {
public synchronized void start() throws Exception {
if (!isStarted.get()) {
memoryUpdateCallbackHandler.validateUpdateReceived();
- if (this.conf.getInt(TezJobConfig.TEZ_RUNTIME_SORT_THREADS,
- TezJobConfig.TEZ_RUNTIME_SORT_THREADS_DEFAULT) > 1) {
+ if (this.conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_THREADS,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SORT_THREADS_DEFAULT) > 1) {
sorter = new PipelinedSorter(getContext(), conf, getNumPhysicalOutputs(),
memoryUpdateCallbackHandler.getMemoryAssigned());
} else {
@@ -201,28 +203,28 @@ public class OnFileSortedOutput extends AbstractLogicalOutput {
private static final Set<String> confKeys = new HashSet<String>();
static {
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SORT_SPILL_PERCENT);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_SORT_MB);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_SORT_THREADS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMBINER_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTERS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUP_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUPS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_INTERNAL_SORTER_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_SPILL_PERCENT);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_THREADS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUP_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUPS_MAX_KEYS);
}
// TODO Maybe add helper methods to extract keys
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedKVOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedKVOutput.java
index 8cd47c1..e9956e4 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedKVOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedKVOutput.java
@@ -32,14 +32,15 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.runtime.api.AbstractLogicalOutput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.events.DataMovementEvent;
import org.apache.tez.runtime.library.api.KeyValuesWriter;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.broadcast.output.FileBasedKVWriter;
import org.apache.tez.runtime.library.shuffle.common.ShuffleUtils;
import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads.DataMovementEventPayloadProto;
@@ -75,11 +76,11 @@ public class OnFileUnorderedKVOutput extends AbstractLogicalOutput {
getContext().requestInitialMemory(0l, null); // mandatory call
this.dataViaEventsEnabled = conf.getBoolean(
- TezJobConfig.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_ENABLED,
- TezJobConfig.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_ENABLED_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_ENABLED,
+ TezRuntimeConfiguration.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_ENABLED_DEFAULT);
this.dataViaEventsMaxSize = conf.getInt(
- TezJobConfig.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_MAX_SIZE,
- TezJobConfig.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_MAX_SIZE_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_MAX_SIZE,
+ TezRuntimeConfiguration.TEZ_RUNTIME_BROADCAST_DATA_VIA_EVENTS_MAX_SIZE_DEFAULT);
LOG.info(this.getClass().getSimpleName() + " running with params -> "
+ "dataViaEventsEnabled: " + dataViaEventsEnabled
@@ -167,18 +168,18 @@ public class OnFileUnorderedKVOutput extends AbstractLogicalOutput {
private static final Set<String> confKeys = new HashSet<String>();
static {
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTERS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUP_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUPS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUP_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUPS_MAX_KEYS);
}
// TODO Maybe add helper methods to extract keys
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedPartitionedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedPartitionedKVOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedPartitionedKVOutput.java
index 5f7ac48..f666924 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedPartitionedKVOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileUnorderedPartitionedKVOutput.java
@@ -25,18 +25,20 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Preconditions;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.AbstractLogicalOutput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.LogicalOutput;
import org.apache.tez.runtime.api.TezOutputContext;
import org.apache.tez.runtime.api.Writer;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.writers.UnorderedPartitionedKVWriter;
@@ -112,22 +114,22 @@ public class OnFileUnorderedPartitionedKVOutput extends AbstractLogicalOutput {
private static final Set<String> confKeys = new HashSet<String>();
static {
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTERS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUP_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_NAME_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COUNTER_GROUPS_MAX_KEY);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezJobConfig.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
+ confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUP_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_NAME_MAX_KEYS);
+ confKeys.add(TezConfiguration.TEZ_AM_COUNTERS_GROUPS_MAX_KEYS);
}
// TODO Maybe add helper methods to extract keys
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java
index d19323b..38859cc 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java
@@ -28,7 +28,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.common.resources.InitialMemoryAllocator;
import org.apache.tez.runtime.common.resources.InitialMemoryRequestContext;
import org.apache.tez.runtime.library.input.ShuffledMergedInput;
@@ -57,9 +57,6 @@ public class WeightedScalingMemoryDistributor implements InitialMemoryAllocator
private static final Log LOG = LogFactory.getLog(WeightedScalingMemoryDistributor.class);
- @VisibleForTesting
- static final double DEFAULT_RESERVE_FRACTION = 0.25d;
-
static final double MAX_ADDITIONAL_RESERVATION_FRACTION_PER_IO = 0.3d;
static final double RESERVATION_FRACTION_PER_IO = 0.025d;
@@ -196,7 +193,7 @@ public class WeightedScalingMemoryDistributor implements InitialMemoryAllocator
}
private void populateTypeScaleMap() {
- String[] ratios = conf.getStrings(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_WEIGHTED_RATIOS);
+ String[] ratios = conf.getStrings(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_WEIGHTED_RATIOS);
int numExpectedValues = RequestType.values().length;
if (ratios == null) {
LOG.info("No ratio specified. Falling back to Linear scaling");
@@ -233,10 +230,10 @@ public class WeightedScalingMemoryDistributor implements InitialMemoryAllocator
private double computeReservedFraction(int numTotalRequests) {
double reserveFractionPerIo = conf.getDouble(
- TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO,
+ TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO,
RESERVATION_FRACTION_PER_IO);
double maxAdditionalReserveFraction = conf.getDouble(
- TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_MAX,
+ TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_MAX,
MAX_ADDITIONAL_RESERVATION_FRACTION_PER_IO);
Preconditions.checkArgument(maxAdditionalReserveFraction >= 0f
&& maxAdditionalReserveFraction <= 1f);
@@ -247,8 +244,8 @@ public class WeightedScalingMemoryDistributor implements InitialMemoryAllocator
+ maxAdditionalReserveFraction);
}
- double initialReserveFraction = conf.getDouble(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_RESERVE_FRACTION,
- DEFAULT_RESERVE_FRACTION);
+ double initialReserveFraction = conf.getDouble(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_RESERVE_FRACTION,
+ TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_RESERVE_FRACTION_DEFAULT);
double additionalReserveFraction = Math.min(maxAdditionalReserveFraction, numTotalRequests
* reserveFractionPerIo);
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
index c0ac632..e9bdccb 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
@@ -38,7 +38,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.security.ssl.SSLFactory;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.shuffle.impl.ShuffleHeader;
import org.apache.tez.runtime.library.shuffle.common.FetchedInput.Type;
@@ -59,8 +59,8 @@ public class Fetcher implements Callable<FetchResult> {
// Configurable fields.
private CompressionCodec codec;
- private boolean ifileReadAhead = TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT;
- private int ifileReadAheadLength = TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT;
+ private boolean ifileReadAhead = TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT;
+ private int ifileReadAheadLength = TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT;
private final SecretKey shuffleSecret;
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/ShuffleUtils.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/ShuffleUtils.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/ShuffleUtils.java
index 87af557..c7b1dde 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/ShuffleUtils.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/ShuffleUtils.java
@@ -37,9 +37,9 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.security.token.Token;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.security.JobTokenIdentifier;
import org.apache.tez.common.security.JobTokenSecretManager;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.sort.impl.IFile;
import org.apache.tez.runtime.library.shuffle.common.HttpConnection.HttpConnectionParams;
@@ -185,24 +185,24 @@ public class ShuffleUtils {
HttpConnectionParamsBuilder builder = new HttpConnectionParamsBuilder();
int connectionTimeout =
- conf.getInt(TezJobConfig.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_STALLED_COPY_TIMEOUT_DEFAULT);
+ conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_STALLED_COPY_TIMEOUT_DEFAULT);
int readTimeout =
- conf.getInt(TezJobConfig.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT_DEFAULT);
+ conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT_DEFAULT);
int bufferSize =
- conf.getInt(TezJobConfig.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE_DEFAULT);
+ conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE_DEFAULT);
boolean keepAlive =
- conf.getBoolean(TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED_DEFAULT);
+ conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED_DEFAULT);
int keepAliveMaxConnections =
conf.getInt(
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS_DEFAULT);
if (keepAlive) {
System.setProperty("sun.net.http.errorstream.enableBuffering", "true");
System.setProperty("http.maxConnections",
@@ -214,8 +214,8 @@ public class ShuffleUtils {
.setBufferSize(bufferSize)
.setKeepAlive(keepAlive, keepAliveMaxConnections);
- boolean sslShuffle = conf.getBoolean(TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL_DEFAULT);
+ boolean sslShuffle = conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL_DEFAULT);
builder.setSSL(sslShuffle, conf);
return builder.build();
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/ShuffleManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/ShuffleManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/ShuffleManager.java
index 16930da..2a0ae0e 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/ShuffleManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/ShuffleManager.java
@@ -43,7 +43,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
@@ -52,6 +51,7 @@ import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.TezInputContext;
import org.apache.tez.runtime.api.events.InputReadErrorEvent;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.InputIdentifier;
import org.apache.tez.runtime.library.common.TezRuntimeUtils;
@@ -166,8 +166,8 @@ public class ShuffleManager implements FetcherCallback {
int maxConfiguredFetchers =
conf.getInt(
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
this.numFetchers = Math.min(maxConfiguredFetchers, numInputs);
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/SimpleFetchedInputAllocator.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/SimpleFetchedInputAllocator.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/SimpleFetchedInputAllocator.java
index 44e20c6..4b17bc7 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/SimpleFetchedInputAllocator.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/impl/SimpleFetchedInputAllocator.java
@@ -25,9 +25,9 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.LocalDirAllocator;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.dag.api.TezUncheckedException;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.task.local.output.TezTaskOutputFiles;
@@ -74,11 +74,11 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
// Setup configuration
final float maxInMemCopyUse = conf.getFloat(
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT_DEFAULT);
if (maxInMemCopyUse > 1.0 || maxInMemCopyUse < 0.0) {
throw new IllegalArgumentException("Invalid value for "
- + TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT + ": "
+ + TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT + ": "
+ maxInMemCopyUse);
}
@@ -94,12 +94,12 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
LOG.info("RequestedMem=" + memReq + ", Allocated: " + this.memoryLimit);
final float singleShuffleMemoryLimitPercent = conf.getFloat(
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT_DEFAULT);
if (singleShuffleMemoryLimitPercent <= 0.0f
|| singleShuffleMemoryLimitPercent > 1.0f) {
throw new IllegalArgumentException("Invalid value for "
- + TezJobConfig.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT + ": "
+ + TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT + ": "
+ singleShuffleMemoryLimitPercent);
}
@@ -112,11 +112,11 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
@Private
public static long getInitialMemoryReq(Configuration conf, long maxAvailableTaskMemory) {
final float maxInMemCopyUse = conf.getFloat(
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT,
- TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT_DEFAULT);
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT,
+ TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT_DEFAULT);
if (maxInMemCopyUse > 1.0 || maxInMemCopyUse < 0.0) {
throw new IllegalArgumentException("Invalid value for "
- + TezJobConfig.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT + ": "
+ + TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT + ": "
+ maxInMemCopyUse);
}
long memReq = (long) (conf.getLong(Constants.TEZ_RUNTIME_TASK_MEMORY,
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/common/resources/TestWeightedScalingMemoryDistributor.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/common/resources/TestWeightedScalingMemoryDistributor.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/common/resources/TestWeightedScalingMemoryDistributor.java
index 0de527f..eefcb0b 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/common/resources/TestWeightedScalingMemoryDistributor.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/common/resources/TestWeightedScalingMemoryDistributor.java
@@ -23,9 +23,9 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.dag.api.InputDescriptor;
import org.apache.tez.dag.api.OutputDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.LogicalInput;
import org.apache.tez.runtime.api.LogicalOutput;
import org.apache.tez.runtime.api.MemoryUpdateCallback;
@@ -44,19 +44,20 @@ public class TestWeightedScalingMemoryDistributor extends TestMemoryDistributor
@Override
public void setup() {
- conf.setBoolean(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ENABLED, true);
- conf.set(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ALLOCATOR_CLASS,
+ conf.setBoolean(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ENABLED, true);
+ conf.set(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ALLOCATOR_CLASS,
WeightedScalingMemoryDistributor.class.getName());
- conf.setDouble(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_RESERVE_FRACTION, 0.3d);
- conf.setDouble(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO, 0.0d);
+ conf.setDouble(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_RESERVE_FRACTION, 0.3d);
+ conf.setDouble(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO, 0.0d);
}
@Test(timeout = 5000)
public void testSimpleWeightedScaling() {
Configuration conf = new Configuration(this.conf);
- conf.setStrings(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_WEIGHTED_RATIOS,
+ conf.setStrings(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_WEIGHTED_RATIOS,
generateWeightStrings(1, 2, 3, 1, 1));
- System.err.println(Joiner.on(",").join(conf.getStringCollection(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_WEIGHTED_RATIOS)));
+ System.err.println(Joiner.on(",").join(conf.getStringCollection(
+ TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_WEIGHTED_RATIOS)));
MemoryDistributor dist = new MemoryDistributor(2, 2, conf);
@@ -100,10 +101,10 @@ public class TestWeightedScalingMemoryDistributor extends TestMemoryDistributor
@Test(timeout = 5000)
public void testAdditionalReserveFractionWeightedScaling() {
Configuration conf = new Configuration(this.conf);
- conf.setStrings(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_WEIGHTED_RATIOS,
+ conf.setStrings(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_WEIGHTED_RATIOS,
generateWeightStrings(2, 3, 6, 1, 1));
- conf.setDouble(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO, 0.025d);
- conf.setDouble(TezJobConfig.TEZ_RUNTIME_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_MAX, 0.2d);
+ conf.setDouble(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_PER_IO, 0.025d);
+ conf.setDouble(TezConfiguration.TEZ_TASK_SCALE_TASK_MEMORY_ADDITIONAL_RESERVATION_FRACTION_MAX, 0.2d);
MemoryDistributor dist = new MemoryDistributor(2, 2, conf);
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/api/TestTezRuntimeConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/api/TestTezRuntimeConfiguration.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/api/TestTezRuntimeConfiguration.java
new file mode 100644
index 0000000..4c0f6b9
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/api/TestTezRuntimeConfiguration.java
@@ -0,0 +1,55 @@
+/*
+ * *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF licenses this file
+ * * to you under the Apache License, Version 2.0 (the
+ * * "License"); you may not use this file except in compliance
+ * * with the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.tez.runtime.library.api;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.lang.reflect.Field;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestTezRuntimeConfiguration {
+
+
+ @Test
+ public void testKeySet() throws IllegalAccessException {
+ Class<?> c = TezRuntimeConfiguration.class;
+ Set<String> expectedKeys = new HashSet<String>();
+ for (Field f : c.getFields()) {
+ if (!f.getName().endsWith("DEFAULT") && f.getType() == String.class) {
+ expectedKeys.add((String) f.get(null));
+ }
+ }
+
+ Set<String> actualKeySet = TezRuntimeConfiguration.getRuntimeConfigKeySet();
+ for (String key : actualKeySet) {
+ if (!expectedKeys.remove(key)) {
+ fail("Found unexpected key: " + key + " in key set");
+ }
+ }
+ assertTrue("Missing keys in key set: " + expectedKeys, expectedKeys.size() == 0);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java
index 9b37c78..ad99d29 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java
@@ -58,7 +58,6 @@ import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.util.DiskChecker.DiskErrorException;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
@@ -68,6 +67,7 @@ import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.TezOutputContext;
import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.library.api.Partitioner;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.sort.impl.IFile;
import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
import org.apache.tez.runtime.library.common.sort.impl.TezSpillRecord;
@@ -633,16 +633,16 @@ public class TestUnorderedPartitionedKVWriter {
Class<? extends Partitioner> partitionerClass) {
Configuration conf = new Configuration(false);
conf.setStrings(TezRuntimeFrameworkConfigs.LOCAL_DIRS, outputContext.getWorkDirs());
- conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, keyClass.getName());
- conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, valClass.getName());
- conf.set(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS, partitionerClass.getName());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, keyClass.getName());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, valClass.getName());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, partitionerClass.getName());
if (maxSingleBufferSizeBytes >= 0) {
- conf.setInt(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES,
+ conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES,
maxSingleBufferSizeBytes);
}
- conf.setBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS, shouldCompress);
+ conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS, shouldCompress);
if (shouldCompress) {
- conf.set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC,
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC,
DefaultCodec.class.getName());
}
return conf;
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfiguration.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfiguration.java
index 2b039a4..c45f597 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfiguration.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileSortedOutputConfiguration.java
@@ -29,7 +29,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.junit.Test;
public class TestOnFileSortedOutputConfiguration {
@@ -65,12 +65,12 @@ public class TestOnFileSortedOutputConfiguration {
public void testSetters() {
Configuration fromConf = new Configuration(false);
fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+ 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(TezJobConfig.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "TestInternalSorter");
+ additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "TestInternalSorter");
OnFileSortedOutputConfiguration.Builder builder =
OnFileSortedOutputConfiguration.newBuilder("KEY", "VALUE", "PARTITIONER", null)
.setKeyComparatorClass("KEY_COMPARATOR")
@@ -78,7 +78,7 @@ public class TestOnFileSortedOutputConfiguration {
.setSortBufferSize(2048)
.setAdditionalConfiguration("test.key.1", "key1")
.setAdditionalConfiguration("file.shouldExist", "file")
- .setAdditionalConfiguration(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
+ .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
String.valueOf(false))
.setAdditionalConfiguration(additionalConf)
.setFromConfiguration(fromConf);
@@ -93,23 +93,23 @@ public class TestOnFileSortedOutputConfiguration {
Configuration conf = rebuilt.conf;
// Verify programmatic API usage
- assertEquals(2048, conf.getInt(TezJobConfig.TEZ_RUNTIME_IO_SORT_MB, 0));
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+ 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(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS,
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
false));
- assertEquals("KEY_COMPARATOR", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+ assertEquals("KEY_COMPARATOR", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
// Verify additional configs
- assertEquals(false, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+ 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(TezJobConfig.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, ""));
+ 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"));
@@ -130,17 +130,17 @@ public class TestOnFileSortedOutputConfiguration {
Configuration conf = rebuilt.conf;
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
// Property present
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
// Verify whatever was configured
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+ 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
@@ -148,7 +148,7 @@ public class TestOnFileSortedOutputConfiguration {
Configuration partitionerConf = new Configuration(false);
partitionerConf.set("partitioner.test.key", "PARTITIONERKEY");
partitionerConf
- .set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
OnFileSortedOutputConfiguration.Builder builder =
OnFileSortedOutputConfiguration.newBuilder("KEY", "VALUE", "PARTITIONER", partitionerConf);
@@ -162,7 +162,7 @@ public class TestOnFileSortedOutputConfiguration {
// Default Output property should not be overridden based on partitioner config
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
assertEquals("PARTITIONERKEY", conf.get("partitioner.test.key"));
}
@@ -172,7 +172,7 @@ public class TestOnFileSortedOutputConfiguration {
Configuration combinerConf = new Configuration(false);
combinerConf.set("combiner.test.key", "COMBINERKEY");
combinerConf
- .set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
OnFileSortedOutputConfiguration.Builder builder =
OnFileSortedOutputConfiguration.newBuilder("KEY", "VALUE", "PARTITIONER", null)
.setCombiner("COMBINER", combinerConf);
@@ -187,7 +187,7 @@ public class TestOnFileSortedOutputConfiguration {
// Default Output property should not be overridden based on partitioner config
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
assertEquals("COMBINERKEY", conf.get("combiner.test.key"));
}
http://git-wip-us.apache.org/repos/asf/tez/blob/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfiguration.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfiguration.java
index 3adc003..8826a5e 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfiguration.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedKVOutputConfiguration.java
@@ -29,7 +29,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.junit.Test;
public class TestOnFileUnorderedKVOutputConfiguration {
@@ -57,18 +57,18 @@ public class TestOnFileUnorderedKVOutputConfiguration {
public void testSetters() {
Configuration fromConf = new Configuration(false);
fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+ 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(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
+ additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
additionalConf.put("file.shouldExist", "file");
OnFileUnorderedKVOutputConfiguration.Builder builder =
OnFileUnorderedKVOutputConfiguration.newBuilder("KEY", "VALUE")
.enableCompression("CustomCodec")
.setAdditionalConfiguration("fs.shouldExist", "fs")
.setAdditionalConfiguration("test.key.1", "key1")
- .setAdditionalConfiguration(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
+ .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
String.valueOf(false))
.setAdditionalConfiguration(additionalConf)
.setFromConfiguration(fromConf);
@@ -84,18 +84,18 @@ public class TestOnFileUnorderedKVOutputConfiguration {
Configuration conf = rebuilt.conf;
// Verify programmatic API usage
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
+ assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+ assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
assertEquals("CustomCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS,
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
false));
// Verify additional configs
- assertEquals(false, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+ 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"));
@@ -118,15 +118,15 @@ public class TestOnFileUnorderedKVOutputConfiguration {
Configuration conf = rebuilt.conf;
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
// Default property present.
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
// Verify whatever was configured
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
+ 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/2050f50e/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutput.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutput.java
index 89d6f02..262d778 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutput.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOnFileUnorderedPartitionedKVOutput.java
@@ -29,7 +29,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.common.TezJobConfig;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.junit.Test;
public class TestOnFileUnorderedPartitionedKVOutput {
@@ -65,11 +65,11 @@ public class TestOnFileUnorderedPartitionedKVOutput {
public void testSetters() {
Configuration fromConf = new Configuration(false);
fromConf.set("test.conf.key.1", "confkey1");
- fromConf.setInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+ 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(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
+ additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
additionalConf.put("file.shouldExist", "file");
OnFileUnorderedPartitionedKVOutputConfiguration.Builder builder =
OnFileUnorderedPartitionedKVOutputConfiguration.newBuilder("KEY", "VALUE", "PARTITIONER",
@@ -78,7 +78,7 @@ public class TestOnFileUnorderedPartitionedKVOutput {
.setAvailableBufferSize(1111)
.setAdditionalConfiguration("fs.shouldExist", "fs")
.setAdditionalConfiguration("test.key.1", "key1")
- .setAdditionalConfiguration(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
+ .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
String.valueOf(false))
.setAdditionalConfiguration(additionalConf)
.setFromConfiguration(fromConf);
@@ -94,22 +94,22 @@ public class TestOnFileUnorderedPartitionedKVOutput {
Configuration conf = rebuilt.conf;
// Verify programmatic API usage
- assertEquals(1111, conf.getInt(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB, 0));
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+ 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(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_COMPRESS,
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
false));
// Verify additional configs
- assertEquals(false, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
- assertEquals(1111, conf.getInt(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+ 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(TezJobConfig.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, 0));
+ 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"));
@@ -132,17 +132,17 @@ public class TestOnFileUnorderedPartitionedKVOutput {
Configuration conf = rebuilt.conf;
- assertEquals(true, conf.getBoolean(TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD,
- TezJobConfig.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+ assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+ TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
// Default property present.
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
// Verify whatever was configured
- assertEquals("KEY", conf.get(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, ""));
- assertEquals("VALUE", conf.get(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, ""));
- assertEquals("PARTITIONER", conf.get(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+ 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
@@ -150,7 +150,7 @@ public class TestOnFileUnorderedPartitionedKVOutput {
Configuration partitionerConf = new Configuration(false);
partitionerConf.set("partitioner.test.key", "PARTITIONERKEY");
partitionerConf
- .set(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+ .set(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
OnFileUnorderedPartitionedKVOutputConfiguration.Builder builder =
OnFileUnorderedPartitionedKVOutputConfiguration
.newBuilder("KEY", "VALUE", "PARTITIONER", partitionerConf);
@@ -166,7 +166,7 @@ public class TestOnFileUnorderedPartitionedKVOutput {
// Default Output property should not be overridden based on partitioner config
assertEquals("TestCodec",
- conf.get(TezJobConfig.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
assertEquals("PARTITIONERKEY", conf.get("partitioner.test.key"));
}