DistCp Guide

Basic Usage

The most common invocation of DistCp is an inter-cluster copy:

bash$ hadoop distcp hdfs://nn1:8020/foo/bar \
hdfs://nn2:8020/bar/foo

This will expand the namespace under /foo/bar on nn1 into a temporary file, partition its contents among a set of map tasks, and start a copy on each NodeManager from nn1 to nn2.

One can also specify multiple source directories on the command line:

bash$ hadoop distcp hdfs://nn1:8020/foo/a \
hdfs://nn1:8020/foo/b \
hdfs://nn2:8020/bar/foo

Or, equivalently, from a file using the -f option:

bash$ hadoop distcp -f hdfs://nn1:8020/srclist \
hdfs://nn2:8020/bar/foo

Where srclist contains

hdfs://nn1:8020/foo/a
hdfs://nn1:8020/foo/b

When copying from multiple sources, DistCp will abort the copy with an error message if two sources collide, but collisions at the destination are resolved per the options specified. By default, files already existing at the destination are skipped (i.e. not replaced by the source file). A count of skipped files is reported at the end of each job, but it may be inaccurate if a copier failed for some subset of its files, but succeeded on a later attempt.

It is important that each NodeManager can reach and communicate with both the source and destination file systems. For HDFS, both the source and destination must be running the same version of the protocol or use a backwards-compatible protocol; see [Copying Between Versions] (#Copying_Between_Versions_of_HDFS).

After a copy, it is recommended that one generates and cross-checks a listing of the source and destination to verify that the copy was truly successful. Since DistCp employs both Map/Reduce and the FileSystem API, issues in or between any of the three could adversely and silently affect the copy. Some have had success running with -update enabled to perform a second pass, but users should be acquainted with its semantics before attempting this.

It’s also worth noting that if another client is still writing to a source file, the copy will likely fail. Attempting to overwrite a file being written at the destination should also fail on HDFS. If a source file is (re)moved before it is copied, the copy will fail with a FileNotFoundException.

Please refer to the detailed Command Line Reference for information on all the options available in DistCp.

Update and Overwrite

-update is used to copy files from source that don’t exist at the target or differ from the target version. -overwrite overwrites target-files that exist at the target.

The Update and Overwrite options warrant special attention since their handling of source-paths varies from the defaults in a very subtle manner. Consider a copy from /source/first/ and /source/second/ to /target/, where the source paths have the following contents:

hdfs://nn1:8020/source/first/1
hdfs://nn1:8020/source/first/2
hdfs://nn1:8020/source/second/10
hdfs://nn1:8020/source/second/20

When DistCp is invoked without -update or -overwrite, the DistCp defaults would create directories first/ and second/, under /target. Thus:

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

would yield the following contents in /target:

hdfs://nn2:8020/target/first/1
hdfs://nn2:8020/target/first/2
hdfs://nn2:8020/target/second/10
hdfs://nn2:8020/target/second/20

When either -update or -overwrite is specified, the contents of the source-directories are copied to target, and not the source directories themselves. Thus:

distcp -update hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

would yield the following contents in /target:

hdfs://nn2:8020/target/1
hdfs://nn2:8020/target/2
hdfs://nn2:8020/target/10
hdfs://nn2:8020/target/20

By extension, if both source folders contained a file with the same name (say, 0), then both sources would map an entry to /target/0 at the destination. Rather than to permit this conflict, DistCp will abort.

Now, consider the following copy operation:

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

With sources/sizes:

hdfs://nn1:8020/source/first/1 32
hdfs://nn1:8020/source/first/2 32
hdfs://nn1:8020/source/second/10 64
hdfs://nn1:8020/source/second/20 32

And destination/sizes:

hdfs://nn2:8020/target/1 32
hdfs://nn2:8020/target/10 32
hdfs://nn2:8020/target/20 64

The result will be:

hdfs://nn2:8020/target/1 32
hdfs://nn2:8020/target/2 32
hdfs://nn2:8020/target/10 64
hdfs://nn2:8020/target/20 32

1 is skipped because the file-length and contents match. 2 is copied because it doesn’t exist at the target. 10 and 20 are overwritten since the contents don’t match the source.

If -update is used, 1 is skipped because the file-length and contents match. 2 is copied because it doesn’t exist at the target. 10 and 20 are overwritten since the contents don’t match the source. However, if -append is additionally used, then only 10 is overwritten (source length less than destination) and 20 is appended with the change in file (if the files match up to the destination’s original length).

If -overwrite is used, 1 is overwritten as well.

Sync

-diff option syncs files from a source cluster to a target cluster with a snapshot diff. It copies, renames and removes files in the snapshot diff list.

-update option must be included when -diff option is in use.

Most cloud providers don’t work well with sync at the moment.

Usage:

hadoop distcp -update -diff <from_snapshot> <to_snapshot> <source> <destination>

Example:

hadoop distcp -update -diff snap1 snap2 /src/ /dst/

The command above applies changes from snapshot snap1 to snap2 (i.e. snapshot diff from snap1 to snap2) in /src/ to /dst/. Obviously, it requires /src/ to have both snapshots snap1 and snap2. But the destination /dst/ must also have a snapshot with the same name as <from_snapshot>, in this case snap1. The destination /dst/ should not have new file operations (create, rename, delete) since snap1. Note that when this command finishes, a new snapshot snap2 will NOT be created at /dst/.

-update is required to use -diff option.

For instance, in /src/, if 1.txt is added and 2.txt is deleted after the creation of snap1 and before creation of snap2, the command above will copy 1.txt from /src/ to /dst/ and delete 2.txt from /dst/.

Sync behavior will be elaborated using experiments below.

# Create source and destination directories
hdfs dfs -mkdir /src/ /dst/
# Allow snapshot on source
hdfs dfsadmin -allowSnapshot /src/
# Create a snapshot (empty one)
hdfs dfs -createSnapshot /src/ snap1
# Allow snapshot on destination
hdfs dfsadmin -allowSnapshot /dst/
# Create a from_snapshot with the same name
hdfs dfs -createSnapshot /dst/ snap1

# Put one text file under /src/
echo "This is the 1st text file." > 1.txt
hdfs dfs -put 1.txt /src/
# Create the second snapshot
hdfs dfs -createSnapshot /src/ snap2

# Put another text file under /src/
echo "This is the 2nd text file." > 2.txt
hdfs dfs -put 2.txt /src/
# Create the third snapshot
hdfs dfs -createSnapshot /src/ snap3

hadoop distcp -update -diff snap1 snap2 /src/ /dst/

hdfs dfs -rm -skipTrash /dst/1.txt

hadoop distcp -update -diff snap1 snap3 /src/ /dst/

hdfs dfs -rm -skipTrash /dst/2.txt
# Create snap2 at destination, it contains 1.txt
hdfs dfs -createSnapshot /dst/ snap2

# Delete 1.txt from source
hdfs dfs -rm -skipTrash /src/1.txt
# Create snap4 at source, it only contains 2.txt
hdfs dfs -createSnapshot /src/ snap4

hadoop distcp -update -diff snap2 snap4 /src/ /dst/

raw Namespace Extended Attribute Preservation

This section only applies to HDFS.

If the target and all of the source pathnames are in the /.reserved/raw hierarchy, then ‘raw’ namespace extended attributes will be preserved. ‘raw’ xattrs are used by the system for internal functions such as encryption meta data. They are only visible to users when accessed through the /.reserved/raw hierarchy.

raw xattrs are preserved based solely on whether /.reserved/raw prefixes are supplied. The -p (preserve, see below) flag does not impact preservation of raw xattrs.

To prevent raw xattrs from being preserved, simply do not use the /.reserved/raw prefix on any of the source and target paths.

If the /.reserved/rawprefix is specified on only a subset of the source and target paths, an error will be displayed and a non-0 exit code returned.

DistCp Driver

The DistCp Driver components are responsible for:

• Parsing the arguments passed to the DistCp command on the command-line, via:

  • OptionsParser, and
  • DistCpOptionsSwitch

• Assembling the command arguments into an appropriate DistCpOptions object, and initializing DistCp. These arguments include:

  • Source-paths
  • Target location
  • Copy options (e.g. whether to update-copy, overwrite, which file-attributes to preserve, etc.)

• Orchestrating the copy operation by:

  • Invoking the copy-listing-generator to create the list of files to be copied.
  • Setting up and launching the Hadoop Map-Reduce Job to carry out the copy.
  • Based on the options, either returning a handle to the Hadoop MR Job immediately, or waiting till completion.

The parser-elements are exercised only from the command-line (or if DistCp::run() is invoked). The DistCp class may also be used programmatically, by constructing the DistCpOptions object, and initializing a DistCp object appropriately.

Copy-listing Generator

The copy-listing-generator classes are responsible for creating the list of files/directories to be copied from source. They examine the contents of the source-paths (files/directories, including wild-cards), and record all paths that need copy into a SequenceFile, for consumption by the DistCp Hadoop Job. The main classes in this module include:

1. CopyListing: The interface that should be implemented by any copy-listing-generator implementation. Also provides the factory method by which the concrete CopyListing implementation is chosen.
2. SimpleCopyListing: An implementation of CopyListing that accepts multiple source paths (files/directories), and recursively lists all the individual files and directories under each, for copy.
3. GlobbedCopyListing: Another implementation of CopyListing that expands wild-cards in the source paths.
4. FileBasedCopyListing: An implementation of CopyListing that reads the source-path list from a specified file.

Based on whether a source-file-list is specified in the DistCpOptions, the source-listing is generated in one of the following ways:

1. If there’s no source-file-list, the GlobbedCopyListing is used. All wild-cards are expanded, and all the expansions are forwarded to the SimpleCopyListing, which in turn constructs the listing (via recursive descent of each path).
2. If a source-file-list is specified, the FileBasedCopyListing is used. Source-paths are read from the specified file, and then forwarded to the GlobbedCopyListing. The listing is then constructed as described above.

One may customize the method by which the copy-listing is constructed by providing a custom implementation of the CopyListing interface. The behaviour of DistCp differs here from the legacy DistCp, in how paths are considered for copy.

One may also customize the filtering of files which shouldn’t be copied by passing the current supported implementation of CopyFilter interface or a new implementation can be written. This can be specified by setting the distcp.filters.class in the DistCpOptions:

1. distcp.filters.class to “RegexCopyFilter”. If you are using this implementation, you will have to pass along “CopyFilter” distcp.filters.file which contains the regex used for filtering. Support regular expressions specified by java.util.regex.Pattern.
2. distcp.filters.class to “RegexpInConfigurationFilter”. If you are using this implementation, you will have to pass along the regex also using distcp.exclude-file-regex parameter in “DistCpOptions”. Support regular expressions specified by java.util.regex.Pattern. This is a more dynamic approach as compared to “RegexCopyFilter”.
3. distcp.filters.class to “TrueCopyFilter”. This is used as a default implementation if none of the above options are specified.

The legacy implementation only lists those paths that must definitely be copied on to target. E.g. if a file already exists at the target (and -overwrite isn’t specified), the file isn’t even considered in the MapReduce Copy Job. Determining this during setup (i.e. before the MapReduce Job) involves file-size and checksum-comparisons that are potentially time-consuming.

The new DistCp postpones such checks until the MapReduce Job, thus reducing setup time. Performance is enhanced further since these checks are parallelized across multiple maps.

InputFormats and MapReduce Components

The InputFormats and MapReduce components are responsible for the actual copy of files and directories from the source to the destination path. The listing-file created during copy-listing generation is consumed at this point, when the copy is carried out. The classes of interest here include:

• UniformSizeInputFormat: This implementation of org.apache.hadoop.mapreduce.InputFormat provides equivalence with Legacy DistCp in balancing load across maps. The aim of the UniformSizeInputFormat is to make each map copy roughly the same number of bytes. Apropos, the listing file is split into groups of paths, such that the sum of file-sizes in each InputSplit is nearly equal to every other map. The splitting isn’t always perfect, but its trivial implementation keeps the setup-time low.

• DynamicInputFormat and DynamicRecordReader: The DynamicInputFormat implements org.apache.hadoop.mapreduce.InputFormat, and is new to DistCp. The listing-file is split into several “chunk-files”, the exact number of chunk-files being a multiple of the number of maps requested for in the Hadoop Job. Each map task is “assigned” one of the chunk-files (by renaming the chunk to the task’s id), before the Job is launched. Paths are read from each chunk using the DynamicRecordReader, and processed in the CopyMapper. After all the paths in a chunk are processed, the current chunk is deleted and a new chunk is acquired. The process continues until no more chunks are available. This “dynamic” approach allows faster map-tasks to consume more paths than slower ones, thus speeding up the DistCp job overall.

• CopyMapper: This class implements the physical file-copy. The input-paths are checked against the input-options (specified in the Job’s Configuration), to determine whether a file needs copy. A file will be copied only if at least one of the following is true:

  • A file with the same name doesn’t exist at target.
  • A file with the same name exists at target, but has a different file size.
  • A file with the same name exists at target, but has a different checksum, and -skipcrccheck isn’t mentioned.
  • A file with the same name exists at target, but -overwrite is specified.
  • A file with the same name exists at target, but differs in block-size and block-size needs to be preserved.

• CopyCommitter: This class is responsible for the commit-phase of the DistCp job, including:

  • Preservation of directory-permissions (if specified in the options)
  • Clean-up of temporary-files, work-directories, etc.

Map sizing

By default, DistCp makes an attempt to size each map comparably so that each copies roughly the same number of bytes. Note that files are the finest level of granularity, so increasing the number of simultaneous copiers (i.e. maps) may not always increase the number of simultaneous copies nor the overall throughput.

The new DistCp also provides a strategy to “dynamically” size maps, allowing faster data-nodes to copy more bytes than slower nodes. Using -strategy dynamic (explained in the Architecture), rather than to assign a fixed set of source-files to each map-task, files are instead split into several sets. The number of sets exceeds the number of maps, usually by a factor of 2-3. Each map picks up and copies all files listed in a chunk. When a chunk is exhausted, a new chunk is acquired and processed, until no more chunks remain.

By not assigning a source-path to a fixed map, faster map-tasks (i.e. data-nodes) are able to consume more chunks, and thus copy more data, than slower nodes. While this distribution isn’t uniform, it is fair with regard to each mapper’s capacity.

The dynamic-strategy is implemented by the DynamicInputFormat. It provides superior performance under most conditions.

Tuning the number of maps to the size of the source and destination clusters, the size of the copy, and the available bandwidth is recommended for long-running and regularly run jobs.

Copying Between Versions of HDFS

For copying between two different major versions of Hadoop (e.g. between 1.X and 2.X), one will usually use WebHdfsFileSystem. Unlike the previous HftpFileSystem, as webhdfs is available for both read and write operations, DistCp can be run on both source and destination cluster. Remote cluster is specified as webhdfs://<namenode_hostname>:<http_port>. When copying between same major versions of Hadoop cluster (e.g. between 2.X and 2.X), use hdfs protocol for better performance.

Secure Copy over the wire with distcp

Use the “swebhdfs://” scheme when webhdfs is secured with SSL. For more information see SSL Configurations for SWebHDFS.

MapReduce and other side-effects

As has been mentioned in the preceding, should a map fail to copy one of its inputs, there will be several side-effects.

• Unless -overwrite is specified, files successfully copied by a previous map on a re-execution will be marked as “skipped”.
• If a map fails mapreduce.map.maxattempts times, the remaining map tasks will be killed (unless -i is set).
• If mapreduce.map.speculative is set to be true, the result of the copy is undefined.

DistCp and Object Stores

DistCp works with Object Stores such as Amazon S3, Azure ABFS and Google GCS.

Prequisites

1. The JAR containing the object store implementation is on the classpath, along with all of its dependencies.
2. Unless the JAR automatically registers its bundled filesystem clients, the configuration may need to be modified to state the class which implements the filesystem schema. All of the ASF’s own object store clients are self-registering.
3. The relevant object store access credentials must be available in the cluster configuration, or be otherwise available in all cluster hosts.

DistCp can be used to upload data

hadoop distcp -direct hdfs://nn1:8020/datasets/set1 s3a://bucket/datasets/set1

To download data

hadoop distcp s3a://bucket/generated/results hdfs://nn1:8020/results

To copy data between object stores

hadoop distcp s3a://bucket/generated/results \
  wasb://updates@example.blob.core.windows.net

And do copy data within an object store

hadoop distcp wasb://updates@example.blob.core.windows.net/current \
  wasb://updates@example.blob.core.windows.net/old

And to use -update to only copy changed files.

hadoop distcp -update -numListstatusThreads 20  \
  s3a://history/2016 \
  hdfs://nn1:8020/history/2016

Because object stores are slow to list files, consider setting the -numListstatusThreads option when performing a -update operation on a large directory tree (the limit is 40 threads).

When DistCp -update is used with object stores, generally only the modification time and length of the individual files are compared, not any checksums if the checksum algorithm between the two stores is different.

• The distcp -update between two object stores with different checksum algorithm compares the modification times of source and target files along with the file size to determine whether to skip the file copy. The behavior is controlled by the property distcp.update.modification.time, which is set to true by default. If the source file is more recently modified than the target file, it is assumed that the content has changed, and the file should be updated. We need to ensure that there is no clock skew between the machines. The fact that most object stores do have valid timestamps for directories is irrelevant; only the file timestamps are compared. However, it is important to have the clock of the client computers close to that of the infrastructure, so that timestamps are consistent between the client/HDFS cluster and that of the object store. Otherwise, changed files may be missed/copied too often.

• distcp.update.modification.time would only be used if either of the two stores don’t have checksum validation resulting in incompatible checksum comparison between the two. Even if the property is set to true, it won’t be used if there is valid checksum comparison between the two stores.

To turn off the modification time check, set this in your core-site.xml

<property>
        <name>distcp.update.modification.time</name>
        <value>false</value>
</property>

Notes

• The -atomic option causes a rename of the temporary data, so significantly increases the time to commit work at the end of the operation. Furthermore, as Object Stores other than (optionally) wasb:// do not offer atomic renames of directories the -atomic operation doesn’t actually deliver what is promised. Avoid.

• The -append option is not supported.

• The -diff and rdiff options are not supported

• CRC checking will not be performed, irrespective of the value of the -skipCrc flag.

• All -p options, including those to preserve permissions, user and group information, attributes checksums and replication are generally ignored. The wasb:// connector will preserve the information, but not enforce the permissions.

• Some object store connectors offer an option for in-memory buffering of output —for example the S3A connector. Using such option while copying large files may trigger some form of out of memory event, be it a heap overflow or a YARN container termination. This is particularly common if the network bandwidth between the cluster and the object store is limited (such as when working with remote object stores). It is best to disable/avoid such options and rely on disk buffering.

• Copy operations within a single object store still take place in the Hadoop cluster —even when the object store implements a more efficient COPY operation internally

  That is, an operation such as

  hadoop distcp s3a://bucket/datasets/set1 s3a://bucket/datasets/set2

  Copies each byte down to the Hadoop worker nodes and back to the bucket. As well as being slow, it means that charges may be incurred.

• The -direct option can be used to write to object store target paths directly, avoiding the potentially very expensive temporary file rename operations that would otherwise occur.