DistCp and disaster recovery

A second cluster, and the menu that didn't arrive

DistCp and disaster recovery

Words you'll meet in this lesson

Architecture at a glance

Our lab: two clusters on separate networks. DistCp's map tasks run on prod's YARN and write straight into the DR cluster's DataNodes. Snapshots on both sides make each night's sync small.

From our lab (two Hadoop 3.5.0 clusters)

DR at three levels

DistCp options and DR designs

DR designs

Animated example: when Chennai floods

Two clusters, one nightly sync, one drill

Prod is the Module 8 cluster; DR is the federation cluster from Lesson 7.2 (hdfs://nn-canteen:8020). Commands run on a prod worker, which join_clusters.sh connects to both networks.

1. First full copy

   hadoop distcp /prod/canteen hdfs://nn-canteen:8020/dr/

   exit 0, took 25540 ms Bytes Copied=169964453 Files Copied=5 Launched map tasks=5
   A map-only job: the client listed 5 files, YARN ran 5 copy maps on prod's workers, and each map streamed blocks from prod DataNodes into DR DataNodes, then compared checksums.

2. A day later: plain DistCp again

   hadoop distcp /prod/canteen hdfs://nn-canteen:8020/dr/

   Bytes Copied=70147 Files Copied=1 Files Skipped=4 143 /dr/canteen/menu.csv 5093962 /dr/canteen/feedback.txt
   Only the new file was copied. The changed menu.csv (same name, same size) and the deleted feedback.txt were left as they were, and the job still reported success.

3. -update, then -update -delete

   hadoop distcp -update /prod/canteen hdfs://nn-canteen:8020/dr/canteen
   hadoop distcp -update -delete /prod/canteen hdfs://nn-canteen:8020/dr/canteen
   hdfs dfs -cat hdfs://nn-canteen:8020/dr/canteen/menu.csv | head -1

   Bytes Copied=143 Files Copied=1 Files Skipped=4 ... Files Skipped=5 chai,drink,18,A
   With -update, source and target paths are the folders themselves. Size was equal, but the checksums differed, so menu.csv was copied. -delete then removed feedback.txt from DR.

4. Sync only a snapshot difference

   hdfs dfs -createSnapshot /prod/canteen s1     # and s1 on the DR copy
   hdfs dfs -mv  .../orders_day31.csv .../orders_2026-10-31.csv ; put orders_2026-11-01.csv
   hdfs dfs -createSnapshot /prod/canteen s2
   hdfs snapshotDiff /prod/canteen s1 s2
   hadoop distcp -update -diff s1 s2 /prod/canteen hdfs://nn-canteen:8020/dr/canteen

   M . + ./orders_2026-11-01.csv R ./orders_day31.csv -> ./orders_2026-10-31.csv exit 0, took 15085 ms Bytes Copied=28086 Files Copied=1 Launched map tasks=1
   DistCp asked the NameNode for the diff instead of comparing every file. The rename was applied as a rename on DR (no bytes copied) and only the new 28 KB file travelled. On millions of files this is the difference between hours of listing and minutes.

5. Throttle the copy

   hadoop distcp -bandwidth 1 /prod/canteen/student_orders.csv hdfs://nn-canteen:8020/dr/throttled.csv

   exit 0, took 34192 ms Bytes Copied=19418699
   At 1 MB/s per map, 19.4 MB needs at least 19 s of copying plus job start-up. Throttling protects the link between sites during business hours.

6. Disaster drill

   hadoop jar mr.jar ItemRevenue hdfs://nn-canteen:8020/dr/canteen/canteen_orders.csv hdfs://nn-canteen:8020/dr/report
   hdfs dfs -put day32.csv /prod/canteen/orders_2026-11-02.csv      # written after the last sync

   completed successfully chai 50170 752550 dosa 49948 2247660 idli 49984 1499520 prod files: 7 DR files: 6 missing on DR: orders_2026-11-02.csv
   The Lesson 8.1 report gives identical numbers from DR data. But anything written after the last sync exists only on prod: if prod were gone now, that file would be lost. That gap is the RPO.

Our sync, run by run

Six steps from the lab: full copy, silent skip, update, delete, snapshot diff, drill.

DistCp and DR at work

Simplified; organisations differ in tools and policies.

Decide like DistCp

This model reproduces our second run. Each file has a size and a checksum. Plain DistCp only checks whether the name exists at the target; -update also compares size and checksum; -delete removes target files the source no longer has.

prod = {"canteen_orders.csv": (4692009, "c1"), "menu.csv": (143, "m2"),
        "orders_day31.csv": (70147, "d1"), "orders_x30.csv": (140759640, "x1"),
        "student_orders.csv": (19418699, "s1")}
dr = {"canteen_orders.csv": (4692009, "c1"), "menu.csv": (143, "m1"),
      "feedback.txt": (5093962, "f1"), "orders_x30.csv": (140759640, "x1"),
      "student_orders.csv": (19418699, "s1")}

def plan(src, dst, update=False, delete=False):
    copy, skip = [], []
    for name, (size, crc) in src.items():
        if name not in dst:
            copy.append(name)
        elif update and dst[name] != (size, crc):   # size OR checksum differs
            copy.append(name)
        else:
            skip.append(name)
    gone = [n for n in dst if n not in src] if delete else []
    return copy, skip, gone

for opts in [{}, {"update": True}, {"update": True, "delete": True}]:
    c, s, g = plan(prod, dr, **opts)
    print(opts or "no options", "-> copy", c, "| skipped", len(s), "| delete", g)

• elif update and dst[name] != (size, crc): # size OR checksum differsOur menu.csv: same 143 bytes, different checksum. Only -update notices.
• gone = [n for n in dst if n not in src] if delete else []-delete is what removed feedback.txt from the DR copy.

Run it on your PC: two clusters, one sync, one drill

You need the Module 8 lab (prod) and the Lesson 7.2 federation lab (DR) running together; stop the Module 5 lab to save memory. join_clusters.sh connects only prod's three workers to the DR network: both clusters use names like namenode, and when we joined every container to both networks, a client resolved resourcemanager to the wrong network and got "Connection refused". Then run distcp_lab.sh inside lab8-worker1-1. It takes about 4 minutes.

# ---- join_clusters.sh (run on your PC) ----
# join_clusters.sh: let the lab8 workers (where DistCp's map tasks run) reach the second cluster.
# Only the workers join: the clusters both use names like "namenode", and a container on two networks
# can answer to the wrong address. Run on your PC.
docker compose -p lab72 start                       # the Lesson 7.2 federation cluster = our "DR" site
for c in worker1 worker2 worker3; do docker network connect lab72_default lab8-$c-1; done
docker exec lab8-worker1-1 getent hosts namenode resourcemanager nn-canteen
docker cp lab8-namenode-1:/tmp/mr/mr.jar /tmp/mr.jar && docker cp /tmp/mr.jar lab8-worker1-1:/tmp/mr.jar

# ---- distcp_lab.sh ----
# distcp_lab.sh: copy a production folder to a second (DR) cluster, keep it in sync, then fail over.
# PROD = the Module 8 cluster (hdfs://namenode). DR = the federation cluster of Lesson 7.2 (hdfs://nn-canteen:8020).
# The lab8 workers are joined to the DR cluster's Docker network (join_clusters.sh). Run inside lab8-worker1-1.
DR=hdfs://nn-canteen:8020
P=/prod/canteen
D=$DR/dr/canteen
run() { local name=$1; shift; local s=$(date +%s%N)
  hadoop distcp "$@" > /tmp/$name.log 2>&1; echo "   exit $?, took $(( ($(date +%s%N) - s) / 1000 / 1000 )) ms"
  grep -E "Bytes Copied|Bytes Skipped|Files Copied|Files Skipped|DIR_COPY|Launched map tasks|Deleted|Files Deleted|Bytes Expected" /tmp/$name.log | sed 's/^\s*/   /' | sort -u; }
hdfs dfs -rm -r -f -skipTrash $P $DR/dr > /dev/null 2>&1
hdfs dfs -mkdir -p $P $DR/dr
hdfs dfs -cp /user/hadoop/data/{canteen_orders.csv,student_orders.csv,menu.csv,feedback.txt,orders_x30.csv} $P/
echo "== 0. what we protect (prod)"
hdfs dfs -count -h $P
echo "== 1. first full copy to the DR cluster (one map per file here)"
run d1 $P $DR/dr/
hdfs dfs -count -h $D
echo "== 2. a day later: one new file, one changed file, one deleted file"
hdfs dfs -cat $P/canteen_orders.csv | head -5001 > /tmp/day31.csv; hdfs dfs -put -f /tmp/day31.csv $P/orders_day31.csv
hdfs dfs -cat $P/menu.csv | sed 's/^chai,drink,15/chai,drink,18/' > /tmp/menu.csv; hdfs dfs -put -f /tmp/menu.csv $P/menu.csv
hdfs dfs -rm -skipTrash $P/feedback.txt > /dev/null
echo "-- plain distcp again (no -update): the existing files are NOT replaced"
run d2a $P $DR/dr/
hdfs dfs -ls $D | awk '{print "   ", $5, $8}' | sed "s#$DR##"
echo "-- distcp -update: copy only what is new or different"
run d2 -update $P $D
echo "-- distcp -update -delete: also remove what prod deleted"
run d3 -update -delete $P $D
hdfs dfs -ls $D | awk '{print "   ", $5, $8}' | sed "s#$DR##"
echo "-- is the price change there?"; hdfs dfs -cat $D/menu.csv | head -1
echo "== 3. snapshot-based sync: only the difference between two snapshots"
hdfs dfsadmin -allowSnapshot $P > /dev/null; hdfs dfsadmin -fs $DR -allowSnapshot /dr/canteen > /dev/null
hdfs dfs -createSnapshot $P s1 > /dev/null; hdfs dfs -createSnapshot $D s1 > /dev/null
hdfs dfs -mv $P/orders_day31.csv $P/orders_2026-10-31.csv
hdfs dfs -cat $P/canteen_orders.csv | head -2001 | tail -2000 > /tmp/day32.csv; hdfs dfs -put /tmp/day32.csv $P/orders_2026-11-01.csv
hdfs dfs -createSnapshot $P s2 > /dev/null
hdfs snapshotDiff $P s1 s2 2>/dev/null | sed 's/^/   /'
run d4 -update -diff s1 s2 $P $D
hdfs dfs -ls $D | awk '{print "   ", $5, $8}' | sed "s#$DR##"
echo "== 4. throttle: -bandwidth 1 (MB/s per map) on the 19 MB student_orders.csv"
run d5 -bandwidth 1 $P/student_orders.csv $DR/dr/throttled.csv
echo "== 5. disaster drill: prod is down, run the report on the DR cluster"
hdfs dfs -rm -r -f -skipTrash $DR/dr/report > /dev/null 2>&1
hadoop jar /tmp/mr.jar ItemRevenue $D/canteen_orders.csv $DR/dr/report > /tmp/dr_job.log 2>&1
grep -oE "completed successfully|failed with state [A-Z]+" /tmp/dr_job.log
hdfs dfs -cat $DR/dr/report/part-r-00000 | head -3
echo "-- recovery point: what DR has vs prod right now"
hdfs dfs -put -f /tmp/day32.csv $P/orders_2026-11-02.csv     # written after the last sync
echo "   prod files: $(hdfs dfs -ls $P | grep -c '^-')   DR files: $(hdfs dfs -ls $D | grep -c '^-')   missing on DR: $(diff <(hdfs dfs -ls $P | grep '^-' | awk -F/ '{print $NF}') <(hdfs dfs -ls $D | grep '^-' | awk -F/ '{print $NF}') | grep '^<' | cut -c3-)"

# Run:  docker cp distcp_lab.sh lab8-worker1-1:/tmp/ && docker exec lab8-worker1-1 bash /tmp/distcp_lab.sh

Applying a snapshot diff

Follow a DR folder as the diff is applied: a rename moves a name without copying bytes, and only the new file is copied.

Worked examples: real problems, solved and traced in memory

Four DR calculations from different industries: RPO and RTO, the sync window, snapshot savings and cost.

sync every 24 h · failover 90 min · limits 4 h / 2 h

RPO 24 h (limit 4) -> FAIL; RTO 1.5 h (limit 2) -> OK
syncs needed per day for RPO 4 h: 6

1.2 PB · 10 Gbit/s · 60% usable

20.8 days

50,000,000 files · 20,000 changes · 2 ms per file

-update compare: 56 h   -diff: 1.3 min

3 options · 0.2 disasters per year

A backup           run   40 lakh + expected loss  57.6 lakh =   97.6 lakh/yr
B active-passive   run  160 lakh + expected loss   2.4 lakh =  162.4 lakh/yr
C active-active    run  400 lakh + expected loss   0.1 lakh =  400.1 lakh/yr

Check yourself

Quick revision

Interview and viva questions

Practice problems

DR exercises with the real numbers from our two clusters.

bytes · ms

6.3 MB/s

print(round(169964453 / 2**20 / 25.540, 1), "MB/s")

Bytes Copied / Bytes Skipped

copied 70147 | stale but skipped: menu.csv 143 bytes

print("copied", 70147, "| stale but skipped: menu.csv", 143, "bytes")

19.4 MB · 1 MB/s

18.5 s minimum

print(round(19418699 / 2**20, 1), "s minimum")

4 syncs a day

lost: 5.5 h | worst case: 6 h

syncs = [0, 6, 12, 18]
d = 17.5
print("lost:", d - max(s for s in syncs if s <= d), "h | worst case: 6 h")

two dicts name → checksum

copy: ['c']
stale but skipped: ['b']
left behind: ['d']

src = {"a": "1", "b": "2", "c": "3"}
dst = {"a": "1", "b": "9", "d": "4"}
print("copy:", [n for n in src if n not in dst])
print("stale but skipped:", [n for n in src if n in dst and src[n] != dst[n]])
print("left behind:", [n for n in dst if n not in src])

30 TB · 200 maps · 80 MB/s

154 GB per map, about 0.5 h

per_map_gb = 30 * 1024 / 200
print(f"{per_map_gb:.0f} GB per map, about {per_map_gb * 1024 / 80 / 3600:.1f} h")

4 drills/yr · ₹15,000 × 6 · 0.2 disasters/yr

drills Rs 360,000 vs expected saving Rs 200,000 per year

drills = 4 * 6 * 15000
saved = 0.2 * (4 - 2) * 500000
print(f"drills Rs {drills:,} vs expected saving Rs {saved:,.0f} per year")

diff lines + sizes

[('copy', './a.csv'), ('delete', './old.csv'), ('rename', './x.csv', './y.csv'), ('copy', './menu.csv')] 28229 bytes

diff = ["M\t.", "+\t./a.csv", "-\t./old.csv", "R\t./x.csv -> ./y.csv", "M\t./menu.csv"]
size = {"./a.csv": 28086, "./menu.csv": 143}
plan, nbytes = [], 0
for line in diff:
    kind, path = line.split("\t")
    if path == ".":
        continue
    if kind in "+M":
        plan.append(("copy", path)); nbytes += size[path]
    elif kind == "-":
        plan.append(("delete", path))
    elif kind == "R":
        plan.append(("rename",) + tuple(path.split(" -> ")))
print(plan, nbytes, "bytes")

Real-world scenario questions (asked in company interviews)

DistCp and DR questions come up in data platform and Hadoop admin interviews.