You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by sp...@apache.org on 2017/02/27 16:14:02 UTC
hive git commit: HIVE-15881: Use hive.exec.input.listing.max.threads
variable name instead of mapred.dfsclient.parallelism.max (Sergio Pena,
reviewed by Yongzhi Che, Thomas Poepping, Illya Yalovvy, Sahil Takiar)
Repository: hive
Updated Branches:
refs/heads/master 13a1978a6 -> 7f1c29ebe
HIVE-15881: Use hive.exec.input.listing.max.threads variable name instead of mapred.dfsclient.parallelism.max (Sergio Pena, reviewed by Yongzhi Che, Thomas Poepping, Illya Yalovvy, Sahil Takiar)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/7f1c29eb
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/7f1c29eb
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/7f1c29eb
Branch: refs/heads/master
Commit: 7f1c29ebe436c0c1ef42caf5f7a697481107d4fb
Parents: 13a1978
Author: Sergio Pena <se...@cloudera.com>
Authored: Mon Feb 27 10:12:28 2017 -0600
Committer: Sergio Pena <se...@cloudera.com>
Committed: Mon Feb 27 10:13:47 2017 -0600
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 4 +
.../apache/hadoop/hive/ql/exec/Utilities.java | 64 +++++--
.../hadoop/hive/ql/exec/TestUtilities.java | 185 +++++++++++++++----
3 files changed, 196 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/7f1c29eb/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index f0c129b..9c721ed 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -3322,6 +3322,10 @@ public class HiveConf extends Configuration {
"Timeout for Running Query in seconds. A nonpositive value means infinite. " +
"If the query timeout is also set by thrift API call, the smaller one will be taken."),
+
+ HIVE_EXEC_INPUT_LISTING_MAX_THREADS("hive.exec.input.listing.max.threads", 0, new SizeValidator(0L, true, 1024L, true),
+ "Maximum number of threads that Hive uses to list file information from file systems (recommended > 1 for blobstore)."),
+
/* BLOBSTORE section */
HIVE_BLOBSTORE_SUPPORTED_SCHEMES("hive.blobstore.supported.schemes", "s3,s3a,s3n",
http://git-wip-us.apache.org/repos/asf/hive/blob/7f1c29eb/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index 3484493..906b4db 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -228,6 +228,9 @@ public final class Utilities {
public static String MAPNAME = "Map ";
public static String REDUCENAME = "Reducer ";
+ @Deprecated
+ protected static String DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX = "mapred.dfsclient.parallelism.max";
+
/**
* ReduceField:
* KEY: record key
@@ -2069,6 +2072,41 @@ public final class Utilities {
private static final Object INPUT_SUMMARY_LOCK = new Object();
/**
+ * Returns the maximum number of executors required to get file information from several input locations.
+ * It checks whether HIVE_EXEC_INPUT_LISTING_MAX_THREADS or DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX are > 1
+ *
+ * @param conf Configuration object to get the maximum number of threads.
+ * @param inputLocationListSize Number of input locations required to process.
+ * @return The maximum number of executors to use.
+ */
+ @VisibleForTesting
+ static int getMaxExecutorsForInputListing(final Configuration conf, int inputLocationListSize) {
+ if (inputLocationListSize < 1) return 0;
+
+ int maxExecutors = 1;
+
+ if (inputLocationListSize > 1) {
+ int listingMaxThreads = HiveConf.getIntVar(conf, ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS);
+
+ // DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX must be removed on next Hive version (probably on 3.0).
+ // If HIVE_EXEC_INPUT_LISTING_MAX_THREADS is not set, then we check of the deprecated configuration.
+ if (listingMaxThreads <= 0) {
+ listingMaxThreads = conf.getInt(DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, 0);
+ if (listingMaxThreads > 0) {
+ LOG.warn("Deprecated configuration is used: " + DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX +
+ ". Please use " + ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname);
+ }
+ }
+
+ if (listingMaxThreads > 1) {
+ maxExecutors = Math.min(inputLocationListSize, listingMaxThreads);
+ }
+ }
+
+ return maxExecutors;
+ }
+
+ /**
* Calculate the total size of input files.
*
* @param ctx
@@ -2117,9 +2155,9 @@ public final class Utilities {
final Map<String, ContentSummary> resultMap = new ConcurrentHashMap<String, ContentSummary>();
ArrayList<Future<?>> results = new ArrayList<Future<?>>();
final ExecutorService executor;
- int maxThreads = ctx.getConf().getInt("mapred.dfsclient.parallelism.max", 0);
- if (pathNeedProcess.size() > 1 && maxThreads > 1) {
- int numExecutors = Math.min(pathNeedProcess.size(), maxThreads);
+
+ int numExecutors = getMaxExecutorsForInputListing(ctx.getConf(), pathNeedProcess.size());
+ if (numExecutors > 1) {
LOG.info("Using " + numExecutors + " threads for getContentSummary");
executor = Executors.newFixedThreadPool(numExecutors,
new ThreadFactoryBuilder().setDaemon(true)
@@ -2978,19 +3016,6 @@ public final class Utilities {
public static List<Path> getInputPaths(JobConf job, MapWork work, Path hiveScratchDir,
Context ctx, boolean skipDummy) throws Exception {
- int numThreads = job.getInt("mapred.dfsclient.parallelism.max", 0);
- ExecutorService pool = null;
- if (numThreads > 1) {
- pool = Executors.newFixedThreadPool(numThreads,
- new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Get-Input-Paths-%d").build());
- }
- return getInputPaths(job, work, hiveScratchDir, ctx, skipDummy, pool);
- }
-
- @VisibleForTesting
- static List<Path> getInputPaths(JobConf job, MapWork work, Path hiveScratchDir,
- Context ctx, boolean skipDummy, ExecutorService pool) throws Exception {
-
Set<Path> pathsProcessed = new HashSet<Path>();
List<Path> pathsToAdd = new LinkedList<Path>();
// AliasToWork contains all the aliases
@@ -3043,6 +3068,13 @@ public final class Utilities {
}
}
+ ExecutorService pool = null;
+ int numExecutors = getMaxExecutorsForInputListing(job, pathsToAdd.size());
+ if (numExecutors > 1) {
+ pool = Executors.newFixedThreadPool(numExecutors,
+ new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Get-Input-Paths-%d").build());
+ }
+
List<Path> finalPathsToAdd = new LinkedList<>();
List<Future<Path>> futures = new LinkedList<>();
for (final Path path : pathsToAdd) {
http://git-wip-us.apache.org/repos/asf/hive/blob/7f1c29eb/ql/src/test/org/apache/hadoop/hive/ql/exec/TestUtilities.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/TestUtilities.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/TestUtilities.java
index 5a9d83c..650f169 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/exec/TestUtilities.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/TestUtilities.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.ql.exec;
+import static org.apache.hadoop.hive.ql.exec.Utilities.DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.apache.hadoop.hive.ql.exec.Utilities.getFileExtension;
@@ -34,7 +35,6 @@ import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ContentSummary;
@@ -301,60 +301,142 @@ public class TestUtilities {
}
/**
- * Test for {@link Utilities#getInputPaths(JobConf, MapWork, Path, Context, boolean, ExecutorService)} with a single
- * threaded {@link ExecutorService}.
+ * Check that calling {@link Utilities#getMaxExecutorsForInputListing(JobConf, int)}
+ * returns the maximum number of executors to use based on the number of input locations.
*/
@Test
- public void testGetInputPathsWithPool() throws Exception {
- ExecutorService pool = Executors.newSingleThreadExecutor();
+ public void testGetMaxExecutorsForInputListing() {
+ Configuration conf = new Configuration();
+
+ final int ZERO_EXECUTORS = 0;
+ final int ONE_EXECUTOR = 1;
+ final int TWO_EXECUTORS = 2;
+
+ final int ZERO_THREADS = 0;
+ final int ONE_THREAD = 1;
+ final int TWO_THREADS = 2;
+
+ final int ZERO_LOCATIONS = 0;
+ final int ONE_LOCATION = 1;
+ final int TWO_LOCATIONS = 2;
+ final int THREE_LOCATIONS = 3;
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, ONE_THREAD);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, TWO_THREADS);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+
+ /*
+ * The following tests will verify the deprecation variable is still usable.
+ */
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, ZERO_THREADS);
+ conf.setInt(DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, ZERO_THREADS);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, ZERO_THREADS);
+ conf.setInt(DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, ONE_THREAD);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, ZERO_THREADS);
+ conf.setInt(DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, TWO_THREADS);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+
+ // Check that HIVE_EXEC_INPUT_LISTING_MAX_THREADS has priority overr DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX
+
+ conf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, TWO_THREADS);
+ conf.setInt(DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, ONE_THREAD);
+
+ assertEquals(ZERO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, ZERO_LOCATIONS));
+ assertEquals(ONE_EXECUTOR, Utilities.getMaxExecutorsForInputListing(conf, ONE_LOCATION));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, TWO_LOCATIONS));
+ assertEquals(TWO_EXECUTORS, Utilities.getMaxExecutorsForInputListing(conf, THREE_LOCATIONS));
+ }
+
+ /**
+ * Test for {@link Utilities#getInputPaths(JobConf, MapWork, Path, Context, boolean)} with a single
+ * threaded.
+ */
+ @Test
+ public void testGetInputPathsWithASingleThread() throws Exception {
+ final int NUM_PARTITIONS = 5;
JobConf jobConf = new JobConf();
+
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 1);
+ runTestGetInputPaths(jobConf, NUM_PARTITIONS);
+ }
+
+ /**
+ * Test for {@link Utilities#getInputPaths(JobConf, MapWork, Path, Context, boolean)} with multiple
+ * threads.
+ */
+ @Test
+ public void testGetInputPathsWithMultipleThreads() throws Exception {
+ final int NUM_PARTITIONS = 5;
+
+ JobConf jobConf = new JobConf();
+
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 2);
+ runTestGetInputPaths(jobConf, NUM_PARTITIONS);
+ }
+
+ private void runTestGetInputPaths(JobConf jobConf, int numOfPartitions) throws Exception {
MapWork mapWork = new MapWork();
Path scratchDir = new Path(HiveConf.getVar(jobConf, HiveConf.ConfVars.LOCALSCRATCHDIR));
+ LinkedHashMap<Path, ArrayList<String>> pathToAliasTable = new LinkedHashMap<>();
+
String testTableName = "testTable";
- String testPartitionName = "testPartition";
Path testTablePath = new Path(testTableName);
- Path testPartitionPath = new Path(testTablePath, testPartitionName);
- Path testFileTablePath = new Path(testTablePath, "test.txt");
- Path testFilePartitionPath = new Path(testPartitionPath, "test.txt");
+ Path[] testPartitionsPaths = new Path[numOfPartitions];
+ for (int i=0; i<numOfPartitions; i++) {
+ String testPartitionName = "p=" + 1;
+ testPartitionsPaths[i] = new Path(testTablePath, "p=" + i);
- TableDesc mockTableDesc = mock(TableDesc.class);
+ pathToAliasTable.put(testPartitionsPaths[i], Lists.newArrayList(testPartitionName));
- when(mockTableDesc.isNonNative()).thenReturn(false);
- when(mockTableDesc.getProperties()).thenReturn(new Properties());
+ mapWork.getAliasToWork().put(testPartitionName, (Operator<?>) mock(Operator.class));
+ }
- LinkedHashMap<Path, ArrayList<String>> pathToAliasTable = new LinkedHashMap<>();
- pathToAliasTable.put(testTablePath, Lists.newArrayList(testTableName));
mapWork.setPathToAliases(pathToAliasTable);
- mapWork.getAliasToWork().put(testTableName, (Operator<?>) mock(Operator.class));
-
FileSystem fs = FileSystem.getLocal(jobConf);
try {
fs.mkdirs(testTablePath);
- fs.create(testFileTablePath).close();
-
- // Run a test with an un-partitioned table with a single file as the input
- List<Path> tableInputPaths = Utilities.getInputPaths(jobConf, mapWork, scratchDir, mock(Context.class), false,
- pool);
- assertEquals(tableInputPaths.size(), 1);
- assertEquals(tableInputPaths.get(0), testTablePath);
-
- LinkedHashMap<Path, ArrayList<String>> pathToAliasPartition = new LinkedHashMap<>();
- pathToAliasPartition.put(testPartitionPath, Lists.newArrayList(testTableName));
- mapWork.setPathToAliases(pathToAliasPartition);
-
- fs.delete(testFileTablePath, false);
- fs.mkdirs(testPartitionPath);
- fs.create(testFilePartitionPath).close();
-
- // Run a test with a partitioned table with a single partition and a single file as the input
- List<Path> tablePartitionInputPaths = Utilities.getInputPaths(jobConf, mapWork, scratchDir, mock(Context.class),
- false, pool);
- assertEquals(tablePartitionInputPaths.size(), 1);
- assertEquals(tablePartitionInputPaths.get(0), testPartitionPath);
+
+ for (int i=0; i<numOfPartitions; i++) {
+ fs.mkdirs(testPartitionsPaths[i]);
+ fs.create(new Path(testPartitionsPaths[i], "test1.txt")).close();
+ }
+
+ List<Path> inputPaths = Utilities.getInputPaths(jobConf, mapWork, scratchDir, mock(Context.class), false);
+ assertEquals(inputPaths.size(), numOfPartitions);
+ for (int i=0; i<numOfPartitions; i++) {
+ assertEquals(inputPaths.get(i), testPartitionsPaths[i]);
+ }
} finally {
if (fs.exists(testTablePath)) {
fs.delete(testTablePath, true);
@@ -370,7 +452,7 @@ public class TestUtilities {
JobConf jobConf = new JobConf();
Properties properties = new Properties();
- jobConf.setInt("mapred.dfsclient.parallelism.max", 0);
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 0);
ContentSummary summary = runTestGetInputSummary(jobConf, properties, NUM_PARTITIONS, BYTES_PER_FILE, HiveInputFormat.class);
assertEquals(NUM_PARTITIONS * BYTES_PER_FILE, summary.getLength());
assertEquals(NUM_PARTITIONS, summary.getFileCount());
@@ -385,11 +467,19 @@ public class TestUtilities {
JobConf jobConf = new JobConf();
Properties properties = new Properties();
- jobConf.setInt("mapred.dfsclient.parallelism.max", 2);
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 2);
ContentSummary summary = runTestGetInputSummary(jobConf, properties, NUM_PARTITIONS, BYTES_PER_FILE, HiveInputFormat.class);
assertEquals(NUM_PARTITIONS * BYTES_PER_FILE, summary.getLength());
assertEquals(NUM_PARTITIONS, summary.getFileCount());
assertEquals(NUM_PARTITIONS, summary.getDirectoryCount());
+
+ // Test deprecated mapred.dfsclient.parallelism.max
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 0);
+ jobConf.setInt(Utilities.DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, 2);
+ summary = runTestGetInputSummary(jobConf, properties, NUM_PARTITIONS, BYTES_PER_FILE, HiveInputFormat.class);
+ assertEquals(NUM_PARTITIONS * BYTES_PER_FILE, summary.getLength());
+ assertEquals(NUM_PARTITIONS, summary.getFileCount());
+ assertEquals(NUM_PARTITIONS, summary.getDirectoryCount());
}
@Test
@@ -401,7 +491,7 @@ public class TestUtilities {
JobConf jobConf = new JobConf();
Properties properties = new Properties();
- jobConf.setInt("mapred.dfsclient.parallelism.max", 2);
+ jobConf.setInt(Utilities.DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, 2);
properties.setProperty(hive_metastoreConstants.META_TABLE_STORAGE, InputEstimatorTestClass.class.getName());
InputEstimatorTestClass.setEstimation(new InputEstimator.Estimation(NUM_OF_ROWS, BYTES_PER_FILE));
@@ -411,6 +501,19 @@ public class TestUtilities {
assertEquals(NUM_PARTITIONS * BYTES_PER_FILE, summary.getLength());
assertEquals(NUM_PARTITIONS * -1, summary.getFileCount()); // Current getInputSummary() returns -1 for each file found
assertEquals(NUM_PARTITIONS * -1, summary.getDirectoryCount()); // Current getInputSummary() returns -1 for each file found
+
+ // Test deprecated mapred.dfsclient.parallelism.max
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 0);
+ jobConf.setInt(HiveConf.ConfVars.HIVE_EXEC_INPUT_LISTING_MAX_THREADS.varname, 2);
+
+ properties.setProperty(hive_metastoreConstants.META_TABLE_STORAGE, InputEstimatorTestClass.class.getName());
+ InputEstimatorTestClass.setEstimation(new InputEstimator.Estimation(NUM_OF_ROWS, BYTES_PER_FILE));
+
+ /* Let's write more bytes to the files to test that Estimator is actually working returning the file size not from the filesystem */
+ summary = runTestGetInputSummary(jobConf, properties, NUM_PARTITIONS, BYTES_PER_FILE * 2, HiveInputFormat.class);
+ assertEquals(NUM_PARTITIONS * BYTES_PER_FILE, summary.getLength());
+ assertEquals(NUM_PARTITIONS * -1, summary.getFileCount()); // Current getInputSummary() returns -1 for each file found
+ assertEquals(NUM_PARTITIONS * -1, summary.getDirectoryCount()); // Current getInputSummary() returns -1 for each file found
}
static class ContentSummaryInputFormatTestClass extends FileInputFormat implements ContentSummaryInputFormat {
@@ -439,7 +542,7 @@ public class TestUtilities {
JobConf jobConf = new JobConf();
Properties properties = new Properties();
- jobConf.setInt("mapred.dfsclient.parallelism.max", 2);
+ jobConf.setInt(Utilities.DEPRECATED_MAPRED_DFSCLIENT_PARALLELISM_MAX, 2);
ContentSummaryInputFormatTestClass.setContentSummary(
new ContentSummary.Builder().length(BYTES_PER_FILE).fileCount(2).directoryCount(1).build());