HADOOP-19307 Add option to add parent directory of source directories to target directories

Author: AuthurWang2009

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

@@ -19,9 +19,11 @@
 package org.apache.hadoop.tools;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Random;
 
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
+import org.apache.hadoop.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -203,6 +205,7 @@ public Job createAndSubmitJob() throws Exception {
         job = createJob();
       }
       prepareFileListing(job);
+      addSourcePathsToConf(job);
       job.submit();
       submitted = true;
     } finally {
@@ -221,6 +224,8 @@ public Job createAndSubmitJob() throws Exception {
     return job;
   }
 
+
+
   /**
    * Wait for the given job to complete.
    * @param job the given mapreduce job that has already been submitted
@@ -233,6 +238,27 @@ public void waitForJobCompletion(Job job) throws Exception {
     }
   }
 
+  private void addSourcePathsToConf(Job job){
+    List<Path> sourcePaths = context.getSourcePaths();
+
+    if(sourcePaths == null || sourcePaths.size() == 0){
+      return;
+    }
+
+    StringBuilder sb = new StringBuilder();
+    for(Path source : sourcePaths){
+      sb.append(StringUtils.escapeString(source.toString()));
+      sb.append(",");
+    }
+    if(sb.length() > 0){
+      sb.deleteCharAt(sb.length() -1);
+    }
+
+
+    job.getConfiguration().set(DistCpConstants.CONF_LABEL_SOURCE_PATHS,
+        sb.toString());
+  }
+
   /**
    * Set targetPathExists in both inputOptions and job config,
    * for the benefit of CopyCommitter

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

@@ -64,7 +64,10 @@ private DistCpConstants() {
   public static final String CONF_LABEL_LISTSTATUS_THREADS = "distcp.liststatus.threads";
   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_SOURCE_PATHS = "distcp.source.paths";
   public static final String CONF_LABEL_COPY_STRATEGY = "distcp.copy.strategy";
+  public static final String CONF_LABEL_PRESERVE_PARENT_DIRS = "distcp.preserve.parent.dirs";
   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

@@ -155,6 +155,10 @@ public void setPreserveRawXattrs(boolean preserveRawXattrs) {
     this.preserveRawXattrs = preserveRawXattrs;
   }
 
+  public boolean shouldPreserveParentDirs(){
+    return options.shouldPreserveParentDirs();
+  }
+
   public Path getAtomicWorkPath() {
     return options.getAtomicWorkPath();
   }

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

@@ -56,6 +56,10 @@ public enum DistCpOptionSwitch {
           "preservation is independent of the -p flag. " +
           "Refer to the DistCp documentation for more details.")),
 
+  PRESERVE_PARENT_DIRS(DistCpConstants.CONF_LABEL_PRESERVE_PARENT_DIRS,
+      new Option("preserveParentDirs", false, "Preserve source directories," +
+          " not the contents of the source-directories")),
+
   /**
    * Update target location by copying only files that are missing
    * in the target. This can be used to periodically sync two folders

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

@@ -149,6 +149,9 @@ public final class DistCpOptions {
   /** File attributes that need to be preserved. */
   private final EnumSet<FileAttribute> preserveStatus;
 
+  /** Source directories that need to be preserved. */
+  private final boolean preserveParentDirs;
+
   // Size of chunk in number of blocks when splitting large file into chunks
   // to copy in parallel. Default is 0 and file are not splitted.
   private final int blocksPerChunk;
@@ -216,6 +219,8 @@ private DistCpOptions(Builder builder) {
 
     this.preserveStatus = builder.preserveStatus;
 
+    this.preserveParentDirs = builder.preserveParentDirs;
+
     this.blocksPerChunk = builder.blocksPerChunk;
 
     this.copyBufferSize = builder.copyBufferSize;
@@ -336,6 +341,15 @@ public boolean shouldPreserve(FileAttribute attribute) {
     return preserveStatus.contains(attribute);
   }
 
+  /**
+   * Checks if the source directories should be preserved or not.
+   *
+   * @return true - True if source dirs should be preserved
+   */
+  public boolean shouldPreserveParentDirs(){
+    return preserveParentDirs;
+  }
+
   public int getBlocksPerChunk() {
     return blocksPerChunk;
   }
@@ -394,6 +408,8 @@ public void appendToConf(Configuration conf) {
       DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.FILTERS,
           filtersFile);
     }
+    DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.PRESERVE_PARENT_DIRS,
+        String.valueOf(preserveParentDirs));
     DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.BLOCKS_PER_CHUNK,
         String.valueOf(blocksPerChunk));
     DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.COPY_BUFFER_SIZE,
@@ -451,6 +467,7 @@ public String toString() {
         ", verboseLog=" + verboseLog +
         ", directWrite=" + directWrite +
         ", useiterator=" + useIterator +
+        ", preserveParentDirs=" + preserveParentDirs +
         '}';
   }
 
@@ -495,6 +512,8 @@ public static class Builder {
     private EnumSet<FileAttribute> preserveStatus =
         EnumSet.noneOf(FileAttribute.class);
 
+    private boolean preserveParentDirs = false;
+
     private int blocksPerChunk = 0;
 
     private int copyBufferSize =
@@ -740,6 +759,11 @@ public Builder preserve(FileAttribute attribute) {
       return this;
     }
 
+    public Builder withPreserveParentDirs(boolean newPreserveDirs){
+      this.preserveParentDirs = newPreserveDirs;
+      return this;
+    }
+
     public Builder withBlocksPerChunk(int newBlocksPerChunk) {
       this.blocksPerChunk = newBlocksPerChunk;
       return this;

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

@@ -465,6 +465,8 @@ private Path computeSourceRootPath(FileStatus sourceStatus,
       } else {
         return sourceStatus.getPath().getParent();
       }
+    } else if (context.shouldPreserveParentDirs()) {
+      return sourceStatus.getPath().getParent();
     } else {
       boolean specialHandling =
           (context.getSourcePaths().size() == 1 &&

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

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.tools.mapred;
 
+import org.apache.hadoop.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,6 +47,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.LinkedList;
 import java.util.List;
@@ -71,6 +73,9 @@ public class CopyCommitter extends FileOutputCommitter {
   private final TaskAttemptContext taskAttemptContext;
   private boolean syncFolder = false;
   private boolean overwrite = false;
+
+  private boolean preserveParentDirs = false;
+
   private boolean targetPathExists = true;
   private boolean ignoreFailures = false;
   private boolean skipCrc = false;
@@ -104,6 +109,7 @@ public void commitJob(JobContext jobContext) throws IOException {
         DistCpConstants.CONF_LABEL_TARGET_PATH_EXISTS, true);
     ignoreFailures = conf.getBoolean(
         DistCpOptionSwitch.IGNORE_FAILURES.getConfigLabel(), false);
+    preserveParentDirs = conf.getBoolean(CONF_LABEL_PRESERVE_PARENT_DIRS, false);
 
     if (blocksPerChunk > 0) {
       concatFileChunks(conf);
@@ -451,8 +457,8 @@ private void deleteMissing(Configuration conf) throws IOException {
           srcAvailable = sourceReader.next(srcRelPath, srcFileStatus);
         }
         Path targetEntry = trgtFileStatus.getPath();
-        LOG.debug("Comparing {} and {}",
-            srcFileStatus.getPath(), targetEntry);
+        LOG.debug("Comparing {} and {}. Source REL PATH: {}, target REL PATH: {}",
+            srcFileStatus.getPath(), targetEntry, srcRelPath, trgtRelPath);
 
         if (srcAvailable && trgtRelPath.equals(srcRelPath)) continue;
 
@@ -552,7 +558,19 @@ private Path listTargetFiles(final Configuration conf,
     Path targetFinalPath = new Path(
         conf.get(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH));
     List<Path> targets = new ArrayList<>(1);
-    targets.add(targetFinalPath);
+    boolean preserveParentDirs = conf.getBoolean(CONF_LABEL_PRESERVE_PARENT_DIRS, false);
+    if(preserveParentDirs){
+      String sourcePathsStr = conf.get(CONF_LABEL_SOURCE_PATHS);
+      String[] sourcePaths = StringUtils.split(sourcePathsStr);
+      LOG.info("sourcePaths: {}", Arrays.toString(sourcePaths));
+      for(String sourcePath : sourcePaths){
+        Path source = new Path(sourcePath);
+        targets.add(new Path(targetFinalPath, source.getName()));
+      }
+    }else{
+      targets.add(targetFinalPath);
+    }
+
     Path resultNonePath = Path.getPathWithoutSchemeAndAuthority(targetFinalPath)
         .toString().startsWith(DistCpConstants.HDFS_RESERVED_RAW_DIRECTORY_NAME)
         ? DistCpConstants.RAW_NONE_PATH
@@ -571,9 +589,11 @@ private Path listTargetFiles(final Configuration conf,
     DistCpOptions options = new DistCpOptions.Builder(targets, resultNonePath)
         .withOverwrite(overwrite)
         .withSyncFolder(syncFolder)
+        .withPreserveParentDirs(preserveParentDirs)
         .withNumListstatusThreads(threads)
         .withUseIterator(useIterator)
         .build();
+    LOG.info("Source directories: {}, target directories: {}", targets, resultNonePath);
     DistCpContext distCpContext = new DistCpContext(options);
     distCpContext.setTargetPathExists(targetPathExists);
 

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

@@ -289,7 +289,7 @@ public void testToString() {
         "atomicWorkPath=null, logPath=null, sourceFileListing=abc, " +
         "sourcePaths=null, targetPath=xyz, filtersFile='null', " +
         "blocksPerChunk=0, copyBufferSize=8192, verboseLog=false, " +
-        "directWrite=false, useiterator=false}";
+        "directWrite=false, useiterator=false, preserveParentDirs=false}";
     String optionString = option.toString();
     Assert.assertEquals(val, optionString);
     Assert.assertNotSame(DistCpOptionSwitch.ATOMIC_COMMIT.toString(),

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

@@ -345,12 +345,12 @@ public void testDeleteMissingInDestination() {
 
     try {
       addEntries(listFile, "srcdir");
-      createFiles("srcdir/file1", "dstdir/file1", "dstdir/file2");
+      createFiles("srcdir/file1", "srcdir/dir1/file1", "dstdir/file1", "dstdir/file2");
 
       Path target = new Path(root + "/dstdir");
       runTest(listFile, target, false, true, true, false);
 
-      checkResult(target, 1, "file1");
+      checkResult(target, 2, "file1","dir1/file1");
     } catch (IOException e) {
       LOG.error("Exception encountered while running distcp", e);
       Assert.fail("distcp failure");
@@ -359,6 +359,32 @@ public void testDeleteMissingInDestination() {
       TestDistCpUtils.delete(fs, "target/tmp1");
     }
   }
+
+
+  @Test
+  public void testUpdateAndDeleteMissingInDestination() {
+
+    try {
+      addEntries(listFile, "/srcdir/dir1","/srcdir/dir2");
+
+      mkdirs(root + "/srcdir/dir1");
+      mkdirs(root + "/srcdir/dir2");
+      mkdirs(root + "/srcdir/dir3");
+
+      createFiles("srcdir/dir1/file1", "srcdir/dir2/file1", "srcdir/dir3/file1");
+      createFiles("dstdir/dir1/file1", "dstdir/dir1/file2", "dstdir/dir3/file1");
+
+      Path target = new Path(root + "/dstdir");
+      runTest(listFile, target, true, true, true, false, true);
+
+      checkResult(target, 3, "dir1/file1","dir2/file1","dir3/file1");
+    } catch (IOException e) {
+      LOG.error("Exception encountered while running distcp", e);
+      Assert.fail("distcp failure");
+    } finally {
+      TestDistCpUtils.delete(fs, root);
+    }
+  }
 
   @Test(timeout=100000)
   public void testOverwrite() {
@@ -559,11 +585,18 @@ private void runTest(Path listFile, Path target, boolean targetExists,
   private void runTest(Path listFile, Path target, boolean targetExists, 
       boolean sync, boolean delete,
       boolean overwrite) throws IOException {
+    runTest(listFile, target, targetExists, sync, delete, overwrite, false);
+  }
+
+  private void runTest(Path listFile, Path target, boolean targetExists,
+                       boolean sync, boolean delete,
+                       boolean overwrite, boolean preserveDirs) throws IOException {
     final DistCpOptions options = new DistCpOptions.Builder(listFile, target)
         .withSyncFolder(sync)
         .withDeleteMissing(delete)
         .withOverwrite(overwrite)
         .withNumListstatusThreads(numListstatusThreads)
+        .withPreserveParentDirs(preserveDirs)
         .build();
     try {
       final DistCp distCp = new DistCp(getConf(), options);