HDFS-17582. Distcp support fastcopy

Author: LiuGuH

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/FastCopy.java

@@ -101,6 +101,9 @@ public class FastCopy {
 
   private final DFSClient srcDFSClient;
   private final DistributedFileSystem dstFs;
+  private long chunkOffset = 0;
+  private long chunkLength = Long.MAX_VALUE;
+
 
   public FastCopy(Configuration conf, Path sourcePath, Path dstPath, boolean overwrite) throws IOException {
     this.conf = conf;
@@ -134,6 +137,12 @@ public FastCopy(Configuration conf, Path sourcePath, Path dstPath, boolean overw
     this.copyBlockExecutor = HadoopExecutors.newFixedThreadPool(this.copyBlockExecutorPoolSize);
   }
 
+  public FastCopy(Configuration conf, Path sourcePath, Path dstPath, boolean overwrite, long chunkOffset, long chunkLength) throws IOException {
+     this(conf,sourcePath,dstPath,overwrite);
+     this.chunkOffset = chunkOffset;
+     this.chunkLength = chunkLength;
+  }
+
   private class CopyBlockCrossNamespace implements Runnable {
     private final ExtendedBlock source;
     private final ExtendedBlock target;
@@ -466,7 +475,7 @@ public void copyFile() throws IOException {
 
     LOG.info("Start to copy {} to {}.", src, dst);
     try {
-      LocatedBlocks blocks = srcDFSClient.getLocatedBlocks(src, 0, srcFileStatus.getLen());
+      LocatedBlocks blocks = srcDFSClient.getLocatedBlocks(src, chunkOffset, chunkLength);
       List<LocatedBlock> blocksList = blocks.getLocatedBlocks();
       LOG.debug("FastCopy : Block locations retrieved for {} : {}.", src, blocksList);
 

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java

@@ -65,6 +65,7 @@ private DistCpConstants() {
   public static final String CONF_LABEL_MAX_MAPS = "distcp.max.maps";
   public static final String CONF_LABEL_SOURCE_LISTING = "distcp.source.listing";
   public static final String CONF_LABEL_COPY_STRATEGY = "distcp.copy.strategy";
+  public static final String CONF_LABEL_USE_FAST_COPY = "distcp.use.fast.copy";
   public static final String CONF_LABEL_SKIP_CRC = "distcp.skip.crc";
   public static final String CONF_LABEL_OVERWRITE = "distcp.copy.overwrite";
   public static final String CONF_LABEL_APPEND = "distcp.copy.append";

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpContext.java

@@ -99,6 +99,10 @@ public boolean shouldSkipCRC() {
     return options.shouldSkipCRC();
   }
 
+  public boolean shouldUseFastCopy() {
+    return options.shouldSkipCRC();
+  }
+
   public boolean shouldBlock() {
     return options.shouldBlock();
   }

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java

@@ -153,6 +153,12 @@ public enum DistCpOptionSwitch {
       new Option("strategy", true, "Copy strategy to use. Default is " +
           "dividing work based on file sizes")),
 
+  /**
+   * Copy file uses fastCopy.
+   */
+  USE_FASTCOPY(DistCpConstants.CONF_LABEL_USE_FAST_COPY,
+      new Option("fastcopy", false, "Copy file uses fastCopy.")),
+
   /**
    * Skip CRC checks between source and target, when determining what
    * files need to be copied.

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java

@@ -161,6 +161,7 @@ public final class DistCpOptions {
   private final boolean directWrite;
 
   private final boolean useIterator;
+  private final boolean useFastCopy;
 
   private final boolean updateRoot;
 
@@ -230,6 +231,7 @@ private DistCpOptions(Builder builder) {
     this.directWrite = builder.directWrite;
 
     this.useIterator = builder.useIterator;
+    this.useFastCopy = builder.useFastCopy;
 
     this.updateRoot = builder.updateRoot;
   }
@@ -290,6 +292,10 @@ public boolean shouldSkipCRC() {
     return skipCRC;
   }
 
+  public boolean shouldUseFastCopy() {
+    return useFastCopy;
+  }
+
   public boolean shouldBlock() {
     return blocking;
   }
@@ -406,6 +412,8 @@ public void appendToConf(Configuration conf) {
         String.valueOf(useRdiff));
     DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.SKIP_CRC,
         String.valueOf(skipCRC));
+    DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.USE_FASTCOPY,
+        String.valueOf(useFastCopy));
     if (mapBandwidth > 0) {
       DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.BANDWIDTH,
           String.valueOf(mapBandwidth));
@@ -532,6 +540,8 @@ public static class Builder {
 
     private boolean updateRoot = false;
 
+    private boolean useFastCopy = false;
+
     public Builder(List<Path> sourcePaths, Path targetPath) {
       Preconditions.checkArgument(sourcePaths != null && !sourcePaths.isEmpty(),
           "Source paths should not be null or empty!");
@@ -816,6 +826,11 @@ public Builder withUpdateRoot(boolean updateRootAttrs) {
       this.updateRoot = updateRootAttrs;
       return this;
     }
+
+    public Builder withUseFastCopy(boolean useFastCopy) {
+      this.useFastCopy = useFastCopy;
+      return this;
+    }
   }
 
 }

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java

@@ -118,6 +118,8 @@ public static DistCpOptions parse(String[] args)
             command.hasOption(DistCpOptionSwitch.DIRECT_WRITE.getSwitch()))
         .withUseIterator(
             command.hasOption(DistCpOptionSwitch.USE_ITERATOR.getSwitch()))
+        .withUseFastCopy(
+            command.hasOption(DistCpOptionSwitch.USE_FASTCOPY.getSwitch()))
         .withUpdateRoot(
             command.hasOption(DistCpOptionSwitch.UPDATE_ROOT.getSwitch()));
 

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java

@@ -97,6 +97,7 @@ public enum ChecksumComparison {
   private boolean verboseLog = false;
   private boolean directWrite = false;
   private boolean useModTimeToUpdate;
+  private boolean useFastCopy = false;
   private EnumSet<FileAttribute> preserve = EnumSet.noneOf(FileAttribute.class);
 
   private FileSystem targetFS = null;
@@ -129,6 +130,7 @@ public void setup(Context context) throws IOException, InterruptedException {
     useModTimeToUpdate =
         conf.getBoolean(DistCpConstants.CONF_LABEL_UPDATE_MOD_TIME,
             CONF_LABEL_UPDATE_MOD_TIME_DEFAULT);
+    useFastCopy = conf.getBoolean(DistCpOptionSwitch.USE_FASTCOPY.getConfigLabel(), false);
 
     targetWorkPath = new Path(conf.get(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH));
     Path targetFinalPath = new Path(conf.get(
@@ -272,9 +274,14 @@ private void copyFileWithRetry(String description,
       throws IOException, InterruptedException {
     long bytesCopied;
     try {
-      bytesCopied = (Long) new RetriableFileCopyCommand(skipCrc, description,
-          action, directWrite).execute(sourceFileStatus, target, context,
-              fileAttributes, sourceStatus);
+      if (!useFastCopy) {
+        bytesCopied =
+            (Long) new RetriableFileCopyCommand(skipCrc, description, action, directWrite).execute(
+                sourceFileStatus, target, context, fileAttributes, sourceStatus);
+      } else {
+        bytesCopied = (Long) new RetriableFileFastCopyCommand(skipCrc, description, action,
+            directWrite).execute(sourceFileStatus, target, context, fileAttributes, sourceStatus);
+      }
     } catch (Exception e) {
       context.setStatus("Copy Failure: " + sourceFileStatus.getPath());
       throw new IOException("File copy failed: " + sourceFileStatus.getPath() +

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java

@@ -66,7 +66,7 @@ public class RetriableFileCopyCommand extends RetriableCommand {
   private static Logger LOG = LoggerFactory.getLogger(RetriableFileCopyCommand.class);
   private boolean skipCrc = false;
   private boolean directWrite = false;
-  private FileAction action;
+  protected FileAction action;
 
   /**
    * Constructor, taking a description of the action.
@@ -192,7 +192,7 @@ private ChecksumOpt getChecksumOpt(EnumSet<FileAttribute> fileAttributes,
   }
 
   @SuppressWarnings("checkstyle:parameternumber")
-  private long copyToFile(Path targetPath, FileSystem targetFS,
+  protected long copyToFile(Path targetPath, FileSystem targetFS,
       CopyListingFileStatus source, long sourceOffset, Mapper.Context context,
       EnumSet<FileAttribute> fileAttributes, final FileChecksum sourceChecksum,
       FileStatus sourceStatus)

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileFastCopyCommand.java

@@ -0,0 +1,62 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.tools.mapred;
+
+import java.io.IOException;
+import java.util.EnumSet;
+
+import org.apache.hadoop.fs.FileChecksum;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.FastCopy;
+import org.apache.hadoop.mapreduce.Mapper.Context;
+import org.apache.hadoop.tools.CopyListingFileStatus;
+import org.apache.hadoop.tools.DistCpOptions.FileAttribute;
+import org.apache.hadoop.tools.mapred.CopyMapper.FileAction;
+
+public class RetriableFileFastCopyCommand extends RetriableFileCopyCommand {
+  public RetriableFileFastCopyCommand(String description, FileAction action) {
+    super(description, action);
+  }
+
+  public RetriableFileFastCopyCommand(boolean skipCrc, String description, FileAction action) {
+    super(skipCrc, description, action);
+  }
+
+  public RetriableFileFastCopyCommand(boolean skipCrc, String description, FileAction action,
+      boolean directWrite) {
+    super(skipCrc, description, action, directWrite);
+  }
+
+  @Override
+  protected long copyToFile(Path targetPath, FileSystem targetFS, CopyListingFileStatus source,
+      long sourceOffset, Context context, EnumSet<FileAttribute> fileAttributes,
+      FileChecksum sourceChecksum, FileStatus sourceStatus) throws IOException {
+    FastCopy fastCopy = new FastCopy(context.getConfiguration(), source.getPath(), targetPath,
+        action == FileAction.OVERWRITE || action == FileAction.APPEND, source.getChunkOffset(),
+        source.getChunkLength());
+    fastCopy.copyFile();
+
+    if (action == FileAction.APPEND) {
+      return source.getLen() - sourceOffset;
+    }
+    return source.getSizeToCopy();
+  }
+}

hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpOptions.java

@@ -134,6 +134,18 @@ public void testSetSkipCRC() {
     Assert.assertTrue(options.shouldSkipCRC());
   }
 
+  @Test
+  public void testUseFastCopy() {
+    final DistCpOptions.Builder builder = new DistCpOptions.Builder(
+        Collections.singletonList(new Path("hdfs://localhost:8020/source")),
+        new Path("hdfs://localhost:8020/target/"));
+    Assert.assertFalse(builder.build().shouldUseFastCopy());
+
+    final DistCpOptions options = builder.withUseFastCopy(true)
+        .build();
+    Assert.assertTrue(options.shouldUseFastCopy());
+  }
+
   @Test
   public void testSetAtomicCommit() {
     final DistCpOptions.Builder builder = new DistCpOptions.Builder(