HADOOP-13600. S3a rename() to copy files in a directory in parallel #157

sahilTakiar · 2016-11-10T19:54:05Z

Creates two separate TransferManagers - one for file uploads and one for file copies; this way they can be both configured separately; users may want to set the # of parallel uploads to a lower value than the # of parallel copies because uploads require actual I/O while copies do not
Main modifications are to the innerRename method
Instead of renaming a directory file by file, use the copy TranferManager to upload them in parallel
Copies are submitted to the TransferManager and then tracked using the returned Copy object
Deletes are handled via a BlockingQueue - once each copy is complete it adds a key to the queue, once MAX_ENTRIES_TO_DELETE keys need to be deleted, then the removeKeys method is invoked
A separate thread is spawned to read from the delete queue and issue the delete requests
A ProgressListener is used to track when a copy has been completed, once a copy finishes, the key to delete is added to the delete queue
Some other re-factoring to make the above possible

steveloughran · 2016-11-28T15:04:36Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+          break;
+        case TRANSFER_COMPLETED_EVENT:
+          try {
+            innerDelete(srcStatus, false);


if we add some threads for low priority delete/create calls, then this can be done in the background. It'd also change how failures are handed: delete won't fail; no need to catch IOEs or AmazonClientExceptions. However, I'd pool the delete operations to avoid throttling problems

thodemoor · 2016-12-12T22:23:55Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+          "s3a-upload-shared");
+
+      copyThreadPoolExecutor = BlockingThreadPoolExecutorService.newInstance(
+          maxThreads,


COPY is server-side (no data transfer) and is thus generally much less resource-intensive and much quicker than PUT (the smaller your bandwidth to S3, the bigger the difference becomes). So I think the maxThreads for the copyThreadpool could be (much) higher than for uploadThreadpool and should thus be configurable separately.

How about the core pool size? I've updated both for now. Wasn't sure what reasonable defaults would be, so I just put down by best guess.

thodemoor · 2016-12-16T15:31:57Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

+  public static final int UPLOAD_DEFAULT_MAX_THREADS = 10;
+
+  // the maximum number of threads to allow in the pool used by TransferManager for copies
+  public static final String COPY_MAX_THREADS = "fs.s3a.threads.max";


fs.s3a.copy.threads.max?

Whoops, fixed. Leaving the parameters for the upload TransferManager the same since changing the name would be a backwards incompatible change.

steveloughran · 2017-09-09T17:45:47Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

-  public static final int DEFAULT_MAX_THREADS = 10;
+  // the maximum number of threads to allow in the pool used by TransferManager for uploads
+  public static final String UPLOAD_MAX_THREADS = "fs.s3a.threads.max";
+  public static final int DEFAULT_UPLOAD_MAX_THREADS = 10;


not going to rename constants because it will break anyone using them. Class is tagged {@InterfaceAudience.Public/evolving}

steveloughran · 2017-09-09T17:46:55Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -269,7 +273,8 @@ public StorageStatistics provide() {
      }
      useListV1 = (listVersion == 1);

-      initTransferManager();
+      this.uploads = createUploadTransferManager(conf);
+      this.copies = createCopyTransferManager(conf);


I'd actually like this to be on demand. That way, if not needed, no overhead of creating it. init times of s3a are already high

Added a separate class called LazyTransferManager that lazily initializes a TransferManager.

steveloughran · 2017-09-09T17:47:44Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+    return uploads;
+  }
+
+  private TransferManager createCopyTransferManager(Configuration conf) {


this is almost the same as the previous one apart from the conf fields and names; should be a common method

Re-factored. Most of this was moved into LazyTransferManager. I abstracted the determination of # of threads into a common method.

steveloughran · 2017-09-09T17:48:39Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+
+      try {
+        // Launch a thread that will read from the deleteQueue and batch delete any files that have already been copied
+        deleteFuture = new FutureTask<>(new Callable<Void>() {


If we target java 8 only, this can be a proper lambda-expression :)

Changed to lambda expression.

steveloughran · 2017-09-09T17:51:13Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -143,9 +150,11 @@
  private Listing listing;
  private long partSize;
  private boolean enableMultiObjectsDelete;
-  private TransferManager transfers;
+  private TransferManager uploads;


leave the name of this alone unless really, really needed because (a) we may do other things with it in future, (b) reduces the size of the diff hence compatibility with other patches.

steveloughran · 2017-09-09T17:53:47Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -2167,7 +2280,34 @@ public String getCanonicalServiceName() {
   * @throws IOException Other IO problems
   */
  private void copyFile(String srcKey, String dstKey, long size)
-      throws IOException, InterruptedIOException, AmazonClientException {
+      throws IOException, AmazonClientException {


leave as before so as to make clear it may be interrupted, unless its mentioned in the javadocs

ajfabbri · 2017-09-12T19:12:03Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/CopyContext.java

@@ -0,0 +1,34 @@
+package org.apache.hadoop.fs.s3a;


Will need apache license text at top.

Whoops, always forget those. Fixed.

ajfabbri · 2017-09-12T19:12:31Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/LazyTransferManager.java

@@ -0,0 +1,63 @@
+package org.apache.hadoop.fs.s3a;


Ditto here.. new files need the apache license text at top

ajfabbri · 2017-09-12T19:23:40Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+              Optional<DeleteObjectsRequest.KeyVersion> key = deleteQueue.take();
+
+              // The thread runs until is is given an EOF message (an Optional#empty())
+              if (key.isPresent()) {


Use of Optional will make this harder to backport to Java 7 branches, no? Looks like you are using it for an EOF marker only? You could also create a static final KeyVersion instance and use a reference equality comparison to detect that it is the special EOF version?

I'm making the leap to Java 8 with the committer and retry logic. Doesn't mean we should jump to using Optional just because we can. Put differently: if you are going to just use it as a fancy null, it's overkill. Key benefit is that you and use map & foreach, but there we are crippled by java's checked exceptions stopping us throwing IOEs

I removed the usage of Optional. Using a private static final DeleteObjectsRequest.KeyVersion END_OF_KEYS_TO_DELETE = new DeleteObjectsRequest.KeyVersion(null, null); as the EOF instead.

ajfabbri · 2017-09-12T19:31:51Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+        if (!deleteFutureComplete) {
+          if (deleteFuture != null && !deleteFuture.isDone() && !deleteFuture.isCancelled()) {
+            deleteFuture.cancel(true);
+          }


Wondering if we can move parallel rename logic to a new class.. S3AFileSystem is getting too big.

mmm. I mostly concur, handing in either the S3aFS or the (expanded) WriteOperationsHelper

Moved all the parallel rename logic into a dedicated calls called ParallelDirectoryRenamer

steveloughran · 2017-09-13T13:28:34Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/CopyContext.java

+    return srcKey;
+  }
+
+  String getDstKey() {


no, go on, use "destKey". We can afford the letter e's

steveloughran · 2017-09-13T13:31:02Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

      int totalTasks = intOption(conf,
          MAX_TOTAL_TASKS, DEFAULT_MAX_TOTAL_TASKS, 1);
      long keepAliveTime = longOption(conf, KEEPALIVE_TIME,
          DEFAULT_KEEPALIVE_TIME, 0);
+


ideally, keep superfluous edits to a min. I know, we all abuse that...but its good to try

steveloughran · 2017-09-13T13:31:51Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -303,7 +296,37 @@ public StorageStatistics provide() {
    } catch (AmazonClientException e) {
      throw translateException("initializing ", new Path(name), e);
    }
+  }
+
+  private int getMaxThreads(Configuration conf, String maxThreadsKey, int defaultMaxThreads) {


move to S3AUtils

steveloughran · 2017-09-13T13:32:59Z

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

+    return maxThreads;
+  }
+
+  private LazyTransferManager createLazyUploadTransferManager(Configuration conf) {


this logic could be moved into LazyUploadTransferManager itself, maybe just as a static method

steveloughran · 2017-09-13T13:37:36Z

In HADOOP-13786 I'm wrapping every single s3 client call with retry policy, then expanding the inconsistent client to generate more faults (initially throttle, later connection setup/response parsing). I'd really like this work to actually await that, as without it this code isn't going to be resilient to large copies where you are much more likely to hit parallel IO. And we need to make sure there's a good failure policy set up there

sahilTakiar · 2017-09-14T04:01:11Z

Updates:

Moved the parallel rename logic into a dedicated class called ParallelDirectoryRenamer
A few other bug fixes, the core logic remains the same

@steveloughran your last comment on HADOOP-13786 suggested you may move the retry logic out into a separate patch? Are you planning to do that? If not, do you think this patch requires waiting for all the work in HADOOP-13786 to be completed?

If there are concerns with retry behavior, we could also set the default value of the copy thread pool to be 1, that way this feature is essentially off by default.

Also what do you mean by "isn't going to be resilient to large copies where you are much more likely to hit parallel IO"? What parallel IO are you referring to?

steveloughran · 2017-09-14T09:26:38Z

I don't see why this needs to be blocked waiting for all of the '13786 patch to go in
but the rename/throttling stuff: yes.
And yes, I do think I can pull out the retry logic once we've got it stable (the committer code is stressing it). But it will be java 8...whoever wants it in branch-2 gets to convert the closure around every s3 call to an anonymous Callable

Parallel execution: we've got another transfer manager issuing COPY requests, so more HTTPS requests to an S3 bucket/shard. The more requests to a single shard, the likelier you are to hit failures. Now, I think the xfer manager does handle the 503 replies which come back, but the rest of the FS client doesn't

sahilTakiar · 2017-09-14T14:42:56Z

@steveloughran ok that makes sense. Thanks for the explanation. Let me know if you need any help with pulling the retry logic out.

This is reincarnation of PR apache#157 It was not possible to merge/rebase that branch, so I had to create another one and reapply the changes. Author: Boris Shkolnik <boryas@apache.org> Reviewers: Xinyu Liu <xinyu@apache.org> Closes apache#178 from sborya/unitTestHappyPath4

steveloughran reviewed Nov 28, 2016

View reviewed changes

sahilTakiar force-pushed the HADOOP-13600 branch 2 times, most recently from 9d2b4e7 to e616400 Compare November 30, 2016 05:40

thodemoor reviewed Dec 12, 2016

View reviewed changes

sahilTakiar force-pushed the HADOOP-13600 branch from e616400 to 7fbfef3 Compare December 15, 2016 18:42

thodemoor reviewed Dec 16, 2016

View reviewed changes

sahilTakiar force-pushed the HADOOP-13600 branch from 7fbfef3 to 343b31f Compare December 16, 2016 22:55

asfgit force-pushed the trunk branch from 25ac447 to 4d1fac5 Compare April 15, 2017 19:10

sahilTakiar force-pushed the HADOOP-13600 branch from 343b31f to 5d116d7 Compare September 8, 2017 20:44

steveloughran reviewed Sep 9, 2017

View reviewed changes

sahilTakiar force-pushed the HADOOP-13600 branch from 5d116d7 to 3f6075e Compare September 12, 2017 17:38

ajfabbri reviewed Sep 12, 2017

View reviewed changes

steveloughran reviewed Sep 13, 2017

View reviewed changes

sahilTakiar force-pushed the HADOOP-13600 branch from 3f6075e to 6597735 Compare September 14, 2017 03:51

HADOOP-13600. S3a rename() to copy files in a directory in parallel

6597735

aajisaka closed this Jul 26, 2019

This was referenced Aug 30, 2021

HDFS-16188. RBF: Router to support resolving monitored namenodes with DNS #3346

Merged

HDFS-16199. Resolve log placeholders in NamenodeBeanMetrics #3362

Merged

HDFS-16213. Flaky test TestFsDatasetImpl#testDnRestartWithHardLink #3386

Merged

hadoop-yetus mentioned this pull request Sep 14, 2021

YARN-10915. AbstractCSQueue: Simplify complex logic in methods: deriveCapacityFromAbsoluteConfigurations and updateEffectiveResources #3418

Merged

4 tasks

HADOOP-13600. S3a rename() to copy files in a directory in parallel #157

HADOOP-13600. S3a rename() to copy files in a directory in parallel #157

Uh oh!

Conversation

sahilTakiar commented Nov 10, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

steveloughran commented Sep 13, 2017

Uh oh!

sahilTakiar commented Sep 14, 2017

sahilTakiar commented Nov 10, 2016 •

edited

Loading

steveloughran commented Sep 14, 2017 •

edited

Loading