DistCp — Distributed Copy for Cross-Cluster and Cloud Transfers

Why it matters

Cluster migrations are surprisingly common. Companies move from on-premise to cloud, from one cloud region to another, or from one Hadoop distribution to another. Every one of these involves copying a lot of data, and every one has DistCp somewhere in the runbook. The efficiency of the copy is a direct driver of migration timeline and therefore cost.

Cross-cluster disaster recovery is another common use. Nightly DistCp jobs replicate critical directories from the primary to a DR cluster, giving you a warm standby that can take over if the primary fails. Incremental DistCp using snapshots keeps the replication fast even for petabyte-scale directories.

The architecture

DistCp is a map-only MapReduce job with a custom input format, and almost every behaviour that surprises operators falls out of that one fact. There is no reducer and no shuffle, because nothing needs to be aggregated: the job is a fleet of independent tasks that each open a source stream and write a destination stream. Job success means every map attempt succeeded, and the tool has no view of global progress beyond the counters those maps report.

Before any map starts, the driver process on the submitting host builds a copy listing - a SequenceFile whose keys are destination-relative paths and whose values carry each source file's length, block size, permissions, ownership and extended attributes. That file is the job's input. UniformSizeInputFormat, the default, slices the listing into splits with roughly equal byte totals. DynamicInputFormat, selected with -strategy dynamic, instead publishes many small chunk files that maps pull from as they finish.

Two consequences are worth internalising. First, -m is a request rather than a guarantee: it caps the number of splits the input format produces, and the split count then caps the map count, so asking for 400 maps on a listing of twelve files gets you twelve maps. Second, DistCp disables map speculative execution, because two attempts writing the same destination path is destructive rather than merely wasteful - a slow map is a slow map, and you cannot race around it. (See speculative execution for how that mechanism works where it is safe.)

Building the copy listing - the phase that runs on one thread

The listing walk is the phase nobody budgets for. It runs in the client JVM before the job is submitted, and by default it recurses the source tree with a single thread issuing listStatus and getFileStatus calls against the source NameNode. On a warehouse directory holding tens of millions of paths this is not a warm-up, it is a substantial fraction of the wall clock - during which zero bytes have moved and the job does not yet exist in YARN at all. Operators watching the ResourceManager see nothing and conclude the submission failed.

-numListstatusThreads parallelises that walk, up to a hard ceiling of 40 threads, against a default of one. Raising it is usually the highest-leverage single change on a large tree, with the obvious caveat that you are now pointing forty concurrent RPC streams at a NameNode which is also serving production traffic, so it interacts directly with handler counts and the RPC call queue. -useiterator solves a different problem: it consumes the listing API iteratively so the client streams directory entries rather than materialising an entire directory in memory, which is what keeps the driver from dying on a directory with a million children.

The listing is also where the small-files tax gets levied twice. Every path costs a NameNode round trip during the walk, an entry in the listing SequenceFile, and later a file open plus a file create in whichever map draws it. Copying ten million 4 KB files moves 40 GB and routinely takes longer than copying forty 1 TB files that move a thousand times more data, because the bottleneck is metadata operations per second and not bandwidth. If that is the shape of your source tree, fix it upstream - see the HDFS small files problem - rather than tuning the copy.

Splits, stragglers, and copying one file with many maps

The uniform-size and dynamic strategies balance work between files, and the tradeoff between them is developed in the DistCp architecture article. Neither helps with the case that actually stalls migrations: a single file so large that no assignment of whole files to maps can balance the job. A 3 TB file is one indivisible unit of work under both strategies, so it pins one map for as long as one stream over one link takes, while the other 199 maps drain their splits and sit idle.

-blocksperchunk is the escape hatch. Given a value of, say, 128, DistCp splits any file larger than that many HDFS blocks into chunks of 128 blocks, treats each chunk as an independent listing entry so separate maps copy them in parallel, and then stitches the chunks back into a single file at the destination using the filesystem's concat operation. That final step is the constraint: the destination filesystem has to support concat, which HDFS does and object stores generally do not, and the option is only meaningful in combination with -update or -overwrite. With a 128 MB block size, 128 blocks per chunk turns that 3 TB file into roughly 190 chunks - enough parallelism that it finishes with the fleet instead of an hour behind it.

-update, -overwrite, and what actually gets compared

These two flags do more than choose a conflict policy - they change the meaning of the destination path. Plain distcp /data/a hdfs://dst/b creates /b/a when /b already exists, copying the source directory into the target. With -update or -overwrite, the contents of /data/a are copied directly into /b. Adding the flag to a previously validated command on the assumption that it only affects skipping is a reliable way to produce a destination tree one level off from the one you tested.

The comparison itself is the part most people get wrong, usually by reasoning from rsync. DistCp does not consult modification time, ever. Under -update, a destination file is skipped when its length matches the source and - where block size preservation is in effect - its block size matches too; if both match, checksums are then compared as well, unless -skipcrccheck was passed. So a file rewritten in place with identical content but a fresh timestamp is correctly skipped, and a file rewritten with different content at the same length is caught by the checksum and missed entirely if you turned that check off. -overwrite skips the comparison and recopies unconditionally, which is right for a small tree you want to be certain about and wrong for a nightly job.

-append is the narrow optimisation for the log-file shape: where the destination file is a strict prefix of a longer source file, copy only the tail instead of the whole thing. It requires -update and is incompatible with skipping the CRC check, for the obvious reason that the prefix claim has to be verified before anything can be appended to it.

-delete and the commit phase nobody profiles

-delete removes destination paths that no longer exist at the source, which is what turns DistCp from an additive copy into a genuine mirror. It is only accepted alongside -update or -overwrite, and - this is the part that surprises people - it does not run in the maps. The deletion pass happens in the job's committer, after every map has finished, in a single process that diffs a listing of the destination against the source listing and issues deletes one at a time.

On a destination holding millions of paths that pass can take longer than the copy did, and it emits no map progress, so the job sits at one hundred percent maps complete for an uncomfortably long time while operators wonder whether it has hung. The same committer also applies preserved attributes to directories, which the maps cannot do reliably because directory contents are still arriving while they run. And if trash is enabled on the destination, deleted files land in .Trash rather than disappearing - a genuine safety net on the first run of a mirror, and a slow-motion quota problem if the mirror deletes heavily every night.

Checksum mismatch - the cause and the escape hatches

The failure looks like a copy that streams perfectly and then fails per file at verification. The cause is that an HDFS file checksum is not a checksum of the file's bytes. Classically it is a digest over the concatenated CRC32C chunk checksums, grouped per block, which makes it a function of the bytes and of the layout - the bytes-per-checksum setting and the block size. Copy a file written with a 128 MB block size into a cluster configured for 256 MB and the destination is byte-identical while the two checksums differ by construction.

There are two honest fixes and one that is not a fix. Preserving block size, via the b attribute in -p, keeps the layout identical so the classic checksums stay comparable; this is why -pb appears in cross-cluster runbooks written by people who do not otherwise care about block size. The composite CRC mode, selected through dfs.checksum.combine.mode, computes a checksum independent of block and chunk boundaries so files with different layouts compare correctly - it has to be available on both ends, and it is treated as an architectural choice in DistCp architecture. -skipcrccheck is the third option and it repairs nothing: it removes the check. That is legitimately correct where the checksums cannot match even in principle, as when copying into or out of an encryption zone, and it is a way to hide real corruption everywhere else.

Object storage destinations - the commit is the expensive part

Pointing DistCp at s3a:// or an equivalent connector changes the cost model without changing the command line. The write path is fine: the connector buffers and issues multipart uploads, and throughput is a function of how many maps run and how much upload bandwidth each container gets. The commit is where the time goes. A job of this shape traditionally writes each task's output to a temporary path and renames it into place on success, and a rename on an object store is not a metadata operation - it is a server-side copy of every byte followed by a delete. On a 100 TB transfer that is a second full copy of the data, charged and timed like the first.

-direct is the flag that removes it: maps write straight to the final destination path and no rename occurs. You give up the intermediate-state guarantee, since a reader watching the prefix during the run sees partial results, and that is precisely the trade - correct for a bulk load into a staging prefix, wrong for a path that consumers read live. The other object-store tax is listing itself: the connector emulates directories over a flat key space, so a recursive listing of a large prefix is many paginated API calls, and both the source walk and any -delete pass pay it.

A realistic invocation

A cross-cluster mirror of a warehouse directory, with the listing parallelised, the giant-file tail split, and the WAN budget respected:

hadoop distcp \
  -Ddfs.checksum.combine.mode=COMPOSITE_CRC \
  -Dmapreduce.map.memory.mb=4096 \
  -update -delete \
  -pbugp \
  -strategy dynamic \
  -blocksperchunk 128 \
  -numListstatusThreads 40 \
  -bandwidth 20 \
  -m 200 \
  -filters /home/etl/distcp-excludes.txt \
  -log hdfs://dst-nn:8020/tmp/distcp-logs/orders-20260705 \
  hdfs://src-nn:8020/warehouse/orders \
  hdfs://dst-nn:8020/warehouse/orders

-filters points at a file of one regular expression per line; any source path matching any of them is dropped from the listing, which is how you keep staging directories, _temporary leftovers and per-job scratch out of a mirror without restructuring the source. The output worth reading afterwards is DistCp's own counter group, which reports files copied, files skipped as already current, files failed, bytes copied and bytes expected. Bytes copied far below bytes expected on an -update run is normal and is the entire point of the flag. A non-zero failure count on a run that exited successfully means -i was in effect and the job ignored errors - which is why -log matters, because that directory is the only artefact that records per-file outcomes on a run which touched a million paths.

Failure modes on a multi-day copy

Delegation tokens expire. On a Kerberised cluster the job runs on tokens obtained at submission, not on your ticket-granting ticket. Those tokens are renewable on an interval - commonly renewed daily against a maximum lifetime of about a week - and a copy still running past the maximum dies on credentials the ResourceManager can no longer refresh. A bootstrap transfer of hundreds of terabytes over a constrained link genuinely reaches that boundary. The related sharp edge is the remote end: the ResourceManager attempts to renew tokens for every HDFS service the job names, so if the destination NameNode sits in another realm or is otherwise unreachable from the RM, submission fails on token renewal rather than on anything to do with copying. The configuration property for excluding specific NameNodes from renewal exists for exactly that case.

There is no checkpoint. If the ApplicationMaster is lost and its retries are exhausted, the job fails and the next attempt rebuilds the listing from scratch. What makes that survivable is -update: the re-run reconsiders every file but copies only what is missing or differs, so the second attempt costs one full listing walk plus the remaining bytes. A long copy submitted without -update is a copy you cannot resume.

Overlapping sources. With several source paths, two of them can resolve to the same destination-relative path. DistCp rejects that during listing construction rather than letting two maps race to write one target - a planning error that surfaces before any data moves, which is the good case.

The source keeps changing. Files deleted between the listing walk and the map that eventually copies them cause task failures; files being appended to are copied at whatever length the map happens to observe. Neither is a bug - a copy over a live namespace has no defined point in time. That is the problem snapshots solve, and it is developed as an architecture question in DistCp architecture and HDFS snapshots.

Bandwidth caps are per map. -bandwidth is a per-map limit in MB/s, so the aggregate ceiling is that cap multiplied by however many maps are actually running at once - a YARN scheduling outcome, not a number you set. Raising -m to speed a job up therefore raises the effective network ceiling too, which is how a copy that behaved politely in testing saturates a WAN link in production.

When DistCp is the wrong tool

DistCp copies files. It has no idea that a directory is a Hive partition, an Iceberg table, or a set of files whose consistency with a metastore matters, and it will cheerfully leave a table half-copied from the metastore's point of view. Table-aware replication belongs at the table-format or metastore layer, with DistCp underneath it as the byte mover rather than the coordinator.

It is also the wrong shape for continuous, low-latency replication. The minimum unit of work is an entire job submission - listing walk, YARN allocation, copy, commit - which is minutes of fixed cost before any policy you write on top of it. DistCp therefore bottoms out at a replication lag measured in tens of minutes, not seconds. Event-driven mirroring of a continuously changing dataset is a different architecture; MirrorMaker 2 shows what that looks like on the Kafka side.

And for a handful of files, or a few gigabytes, the MapReduce overhead is pure loss. A plain filesystem copy beats a job submission that spends a minute on listing and scheduling before it opens its first stream.