You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by so...@apache.org on 2020/05/29 11:37:57 UTC
[hadoop] branch branch-3.1 updated: HADOOP-14698. Make
copyFromLocals -t option available for put as well. Contributed by Andras
Bokor.
This is an automated email from the ASF dual-hosted git repository.
sodonnell pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 1f674e2 HADOOP-14698. Make copyFromLocals -t option available for put as well. Contributed by Andras Bokor.
1f674e2 is described below
commit 1f674e26ba056ae2bffd54ddda267302e5d2519b
Author: S O'Donnell <so...@cloudera.com>
AuthorDate: Fri May 29 10:32:37 2020 +0100
HADOOP-14698. Make copyFromLocals -t option available for put as well. Contributed by Andras Bokor.
---
.../org/apache/hadoop/fs/shell/CopyCommands.java | 108 ++++++++-------------
.../org/apache/hadoop/fs/shell/MoveCommands.java | 20 +++-
.../src/site/markdown/FileSystemShell.md | 4 +-
.../java/org/apache/hadoop/fs/shell/TestMove.java | 7 ++
.../hadoop-common/src/test/resources/testConf.xml | 66 ++++---------
5 files changed, 82 insertions(+), 123 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
index c408a4f..180f98c 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java
@@ -232,26 +232,35 @@ class CopyCommands {
* Copy local files to a remote filesystem
*/
public static class Put extends CommandWithDestination {
+ private ThreadPoolExecutor executor = null;
+ private int numThreads = 1;
+
+ private static final int MAX_THREADS =
+ Runtime.getRuntime().availableProcessors() * 2;
+
public static final String NAME = "put";
public static final String USAGE =
- "[-f] [-p] [-l] [-d] <localsrc> ... <dst>";
+ "[-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>";
public static final String DESCRIPTION =
- "Copy files from the local file system " +
- "into fs. Copying fails if the file already " +
- "exists, unless the -f flag is given.\n" +
- "Flags:\n" +
- " -p : Preserves access and modification times, ownership and the mode.\n" +
- " -f : Overwrites the destination if it already exists.\n" +
- " -l : Allow DataNode to lazily persist the file to disk. Forces\n" +
- " replication factor of 1. This flag will result in reduced\n" +
- " durability. Use with care.\n" +
+ "Copy files from the local file system " +
+ "into fs. Copying fails if the file already " +
+ "exists, unless the -f flag is given.\n" +
+ "Flags:\n" +
+ " -p : Preserves timestamps, ownership and the mode.\n" +
+ " -f : Overwrites the destination if it already exists.\n" +
+ " -t <thread count> : Number of threads to be used, default is 1.\n" +
+ " -l : Allow DataNode to lazily persist the file to disk. Forces" +
+ " replication factor of 1. This flag will result in reduced" +
+ " durability. Use with care.\n" +
" -d : Skip creation of temporary file(<dst>._COPYING_).\n";
@Override
protected void processOptions(LinkedList<String> args) throws IOException {
CommandFormat cf =
new CommandFormat(1, Integer.MAX_VALUE, "f", "p", "l", "d");
+ cf.addOptionWithValue("t");
cf.parse(args);
+ setNumberThreads(cf.getOptValue("t"));
setOverwrite(cf.getOpt("f"));
setPreserve(cf.getOpt("p"));
setLazyPersist(cf.getOpt("l"));
@@ -281,32 +290,22 @@ class CopyCommands {
copyStreamToTarget(System.in, getTargetPath(args.get(0)));
return;
}
- super.processArguments(args);
- }
- }
- public static class CopyFromLocal extends Put {
- private ThreadPoolExecutor executor = null;
- private int numThreads = 1;
+ executor = new ThreadPoolExecutor(numThreads, numThreads, 1,
+ TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024),
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ super.processArguments(args);
- private static final int MAX_THREADS =
- Runtime.getRuntime().availableProcessors() * 2;
- public static final String NAME = "copyFromLocal";
- public static final String USAGE =
- "[-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>";
- public static final String DESCRIPTION =
- "Copy files from the local file system " +
- "into fs. Copying fails if the file already " +
- "exists, unless the -f flag is given.\n" +
- "Flags:\n" +
- " -p : Preserves access and modification times, ownership and the" +
- " mode.\n" +
- " -f : Overwrites the destination if it already exists.\n" +
- " -t <thread count> : Number of threads to be used, default is 1.\n" +
- " -l : Allow DataNode to lazily persist the file to disk. Forces" +
- " replication factor of 1. This flag will result in reduced" +
- " durability. Use with care.\n" +
- " -d : Skip creation of temporary file(<dst>._COPYING_).\n";
+ // issue the command and then wait for it to finish
+ executor.shutdown();
+ try {
+ executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ displayError(e);
+ Thread.currentThread().interrupt();
+ }
+ }
private void setNumberThreads(String numberThreadsString) {
if (numberThreadsString == null) {
@@ -323,22 +322,6 @@ class CopyCommands {
}
}
- @Override
- protected void processOptions(LinkedList<String> args) throws IOException {
- CommandFormat cf =
- new CommandFormat(1, Integer.MAX_VALUE, "f", "p", "l", "d");
- cf.addOptionWithValue("t");
- cf.parse(args);
- setNumberThreads(cf.getOptValue("t"));
- setOverwrite(cf.getOpt("f"));
- setPreserve(cf.getOpt("p"));
- setLazyPersist(cf.getOpt("l"));
- setDirectWrite(cf.getOpt("d"));
- getRemoteDestination(args);
- // should have a -r option
- setRecursive(true);
- }
-
private void copyFile(PathData src, PathData target) throws IOException {
if (isPathRecursable(src)) {
throw new PathIsDirectoryException(src.toString());
@@ -365,25 +348,6 @@ class CopyCommands {
executor.submit(task);
}
- @Override
- protected void processArguments(LinkedList<PathData> args)
- throws IOException {
- executor = new ThreadPoolExecutor(numThreads, numThreads, 1,
- TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024),
- new ThreadPoolExecutor.CallerRunsPolicy());
- super.processArguments(args);
-
- // issue the command and then wait for it to finish
- executor.shutdown();
- try {
- executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
- } catch (InterruptedException e) {
- executor.shutdownNow();
- displayError(e);
- Thread.currentThread().interrupt();
- }
- }
-
@VisibleForTesting
public int getNumThreads() {
return numThreads;
@@ -394,6 +358,12 @@ class CopyCommands {
return executor;
}
}
+
+ public static class CopyFromLocal extends Put {
+ public static final String NAME = "copyFromLocal";
+ public static final String USAGE = Put.USAGE;
+ public static final String DESCRIPTION = "Identical to the -put command.";
+ }
public static class CopyToLocal extends Get {
public static final String NAME = "copyToLocal";
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/MoveCommands.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/MoveCommands.java
index 5ef4277..c20293e 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/MoveCommands.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/MoveCommands.java
@@ -25,7 +25,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.PathIOException;
import org.apache.hadoop.fs.PathExistsException;
-import org.apache.hadoop.fs.shell.CopyCommands.Put;
+import org.apache.hadoop.fs.shell.CopyCommands.CopyFromLocal;
/** Various commands for moving files */
@InterfaceAudience.Private
@@ -41,12 +41,22 @@ class MoveCommands {
/**
* Move local files to a remote filesystem
*/
- public static class MoveFromLocal extends Put {
+ public static class MoveFromLocal extends CopyFromLocal {
public static final String NAME = "moveFromLocal";
- public static final String USAGE = "<localsrc> ... <dst>";
+ public static final String USAGE =
+ "[-f] [-p] [-l] [-d] <localsrc> ... <dst>";
public static final String DESCRIPTION =
- "Same as -put, except that the source is " +
- "deleted after it's copied.";
+ "Same as -put, except that the source is " +
+ "deleted after it's copied\n" +
+ "and -t option has not yet implemented.";
+
+ @Override
+ protected void processOptions(LinkedList<String> args) throws IOException {
+ if(args.contains("-t")) {
+ throw new CommandFormat.UnknownOptionException("-t");
+ }
+ super.processOptions(args);
+ }
@Override
protected void processPath(PathData src, PathData target) throws IOException {
diff --git a/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md b/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
index f4a37ea..6099bd2 100644
--- a/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
+++ b/hadoop-common-project/hadoop-common/src/site/markdown/FileSystemShell.md
@@ -522,7 +522,7 @@ Returns 0 on success and -1 on error.
put
---
-Usage: `hadoop fs -put [-f] [-p] [-l] [-d] [ - | <localsrc1> .. ]. <dst>`
+Usage: `hadoop fs -put [-f] [-p] [-l] [-d] [-t <thread count>] [ - | <localsrc1> .. ]. <dst>`
Copy single src, or multiple srcs from local file system to the destination file system.
Also reads input from stdin and writes to destination file system if the source is set to "-"
@@ -534,6 +534,8 @@ Options:
* `-p` : Preserves access and modification times, ownership and the permissions.
(assuming the permissions can be propagated across filesystems)
* `-f` : Overwrites the destination if it already exists.
+* `-t <thread count>` : Number of threads to be used, default is 1. Useful
+ when uploading a directory containing more than 1 file.
* `-l` : Allow DataNode to lazily persist the file to disk, Forces a replication
factor of 1. This flag will result in reduced durability. Use with care.
* `-d` : Skip creation of temporary file with the suffix `._COPYING_`.
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestMove.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestMove.java
index 94930e5..e2a124e 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestMove.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestMove.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FilterFileSystem;
import org.apache.hadoop.fs.PathExistsException;
+import org.apache.hadoop.fs.shell.CommandFormat.UnknownOptionException;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -93,6 +94,12 @@ public class TestMove {
assertTrue("Rename should have failed with path exists exception",
cmd.error instanceof PathExistsException);
}
+
+ @Test(expected = UnknownOptionException.class)
+ public void testMoveFromLocalDoesNotAllowTOption() {
+ new MoveCommands.MoveFromLocal().run("-t", "2",
+ null, null);
+ }
static class MockFileSystem extends FilterFileSystem {
Configuration conf;
diff --git a/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml b/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
index 29a88fc..2464c28 100644
--- a/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
+++ b/hadoop-common-project/hadoop-common/src/test/resources/testConf.xml
@@ -496,7 +496,10 @@
<comparators>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^-put \[-f\] \[-p\] \[-l\] \[-d\] <localsrc> \.\.\. <dst> :( )*</expected-output>
+ <comparator>
+ <type>RegexpComparator</type>
+ <expected-output>^-put \[-f\] \[-p\] \[-l\] \[-d\] \[-t <thread count>\] <localsrc> \.\.\. <dst> :\s*</expected-output>
+ </comparator>
</comparator>
<comparator>
<type>RegexpComparator</type>
@@ -512,15 +515,19 @@
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-p Preserves access and modification times, ownership and the mode.( )*</expected-output>
+ <expected-output>^\s*-p Preserves timestamps, ownership and the mode.( )*</expected-output>
+ </comparator>
+ <comparator>
+ <type>RegexpComparator</type>
+ <expected-output>^\s*-f Overwrites the destination if it already exists.( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-f Overwrites the destination if it already exists.( )*</expected-output>
+ <expected-output>^\s*-t <thread count> Number of threads to be used, default is 1.( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-l Allow DataNode to lazily persist the file to disk. Forces( )*</expected-output>
+ <expected-output>^\s*-l Allow DataNode to lazily persist the file to disk. Forces( )*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
@@ -532,7 +539,7 @@
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*-d Skip creation of temporary file\(<dst>\._COPYING_\).( )*</expected-output>
+ <expected-output>^\s*-d Skip creation of temporary file\(<dst>\._COPYING_\).( )*</expected-output>
</comparator>
</comparators>
</test>
@@ -551,47 +558,7 @@
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^\s*Copy files from the local file system into fs.( )*Copying fails if the file already( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*exists, unless the -f flag is given.( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*Flags:( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*-p Preserves access and modification times, ownership and the( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*mode.( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*-f Overwrites the destination if it already exists.( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*-t <thread count> Number of threads to be used, default is 1.( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*-l Allow DataNode to lazily persist the file to disk. Forces( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*replication factor of 1. This flag will result in reduced( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*durability. Use with care.( )*</expected-output>
- </comparator>
- <comparator>
- <type>RegexpComparator</type>
- <expected-output>^\s*-d Skip creation of temporary file\(<dst>\._COPYING_\).( )*</expected-output>
+ <expected-output>^\s*Identical to the -put command\.\s*</expected-output>
</comparator>
</comparators>
</test>
@@ -606,11 +573,14 @@
<comparators>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^-moveFromLocal <localsrc> \.\.\. <dst> :\s*</expected-output>
+ <expected-output>^-moveFromLocal \[-f\] \[-p\] \[-l\] \[-d\] <localsrc> \.\.\. <dst> :\s*</expected-output>
</comparator>
<comparator>
<type>RegexpComparator</type>
- <expected-output>^( |\t)*Same as -put, except that the source is deleted after it's copied.</expected-output>
+ <expected-output>^( |\t)*Same as -put, except that the source is deleted after it's copied</expected-output>
+ </comparator><comparator>
+ <type>RegexpComparator</type>
+ <expected-output>^\s* and -t option has not yet implemented.</expected-output>
</comparator>
</comparators>
</test>
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org