Hadoop DistCp 使用详解与原理分析

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

欢 迎 关 注 我 的 公 众 号

前言

集群迁移、灾备、环境间复制，经常要搬数据。单机 hadoop fs -cp 串行执行，DistCp 用 MapReduce 做分布式并行拷贝，性能差距很大。

DistCp（Distributed Copy）是 Hadoop 内置的跨集群/跨目录数据同步工具，一个跑在 MapReduce 上的分布式拷贝作业。本文讲用法、原理和部分源码。

版本

Hadoop 3.1.1

基本概念

DistCp 将文件列表拆分成多个分片，交给集群中的多个 Mapper 并行拷贝。每个 Mapper 在目标路径创建对应目录结构并拷贝文件。Reduce 阶段默认不需要（纯拷贝无聚合）。

与单机拷贝的对比：

方式	执行位置	并行度	适用场景
hadoop fs -cp	客户端单机	1（单线程）	小文件、单次同步、不支持增量 也支持跨集群（串行拷贝）
DistCp	集群 YARN 节点	默认上限 20（-m 控制），文件数远大于 -m 时能趋近或达到上限；文件数少于 -m 时达不到上限（分片数 ≤ 文件数，小文件合并会使分片更少）	中/大数据量， 跨集群分布式并行， 支持增量同步

基本命令

最简单的跨集群拷贝

hadoop distcp hdfs://source-nn:8020/data \
                hdfs://target-nn:8020/data

将源集群 /data 目录拷贝到目标集群 /data 目录。如果目标目录不存在则自动创建。

单集群拷贝（同集群不同目录）

hadoop distcp /data/source /data/target

多源合并拷贝

hadoop distcp /data/2024 /data/2025 /data/archive

多个源目录合并拷贝到同一个目标目录。注意：多个源之间不能有同名文件，否则 DistCp 会报 DuplicateFileException 终止，不会自动覆盖。

指定文件列表

hadoop distcp -f /tmp/filelist.txt hdfs://target-nn:8020/data

filelist.txt 每行一个源路径。注意：如果文件在本地文件系统，需加 file:// 前缀，如 file:///tmp/filelist.txt。

增量同步

hadoop distcp -update hdfs://source-nn:8020/data \
                         hdfs://target-nn:8020/data

只拷贝目标中不存在的文件，或源端更新的文件（基于文件大小和修改时间）。

删除多余文件

hadoop distcp -update -delete hdfs://source-nn:8020/data \
                               hdfs://target-nn:8020/data

在 -update 基础上，删除目标端存在但源端已被删除的文件。使目标与源完全一致。

参数详解

拷贝模式（三选一，互斥）

参数	说明	场景
-update	增量同步，只拷贝目标端不存在或大小/时间/CRC 不同的文件。不会同步删除，需配合 -delete	周期性增量同步，日常数据同步首选
-overwrite	无条件全量覆盖每个文件。与 -update 互斥。不会同步删除目标端多余文件，需配合 -delete	全量刷新场景，如数据管道重跑、重建目标目录
-atomic	原子提交：先写临时目录，全部成功后再移到目标。失败则回滚，目标不变。不会同步删除目标端多余文件，且不支持增量（与 -update/-overwrite 互斥）	对数据一致性要求高的场景，比如关键数据迁移

增量与比对

参数	说明	示例
-update	增量同步，只拷贝新增或更新的文件	-update
-append	追加写入（需 -update），文件大小变化时只从源文件偏移量 = 目标文件大小的位置开始读取并追加到目标末尾，不做内容比对	-update -append
-diff <snap1> <snap2>	基于快照差异的增量同步，跳过文件列表扫描	-diff .snap/day1 .snap/day2
-skipcrccheck	跳过 CRC 校验。CRC 是 HDFS 为每个数据块（默认 512 bytes）计算的校验码。-update 模式下文件大小和修改时间相同时，默认会进一步比对 CRC。加上此参数跳过 CRC，仅看大小和时间，能省开销但可能漏掉大小未变但内容损坏的文件	-skipcrccheck

文件选择与列表

参数	说明	示例
-f <uri>	从文件读取源路径列表。文件在本地需 file:// 前缀	-f /tmp/filelist.txt
-filters <file>	从文件读取正则表达式过滤规则，匹配的文件会被排除，不拷贝到目标	-filters /tmp/ignore.txt

同步行为

参数	说明
-delete	删除目标端多余文件（需配合 -update 或 -diff）
-p[rbugpcaxt]	保留属性。-p 后直接跟字母，不加方括号或空格。不加字母时默认保留 rbugpct（副本、块大小、用户、组、权限、校验和类型、时间戳）。a 和 x 需显式指定：a=ACL（访问控制列表，比传统权限更细粒度的用户/组权限规则），x=XAttr（扩展属性，文件系统层的键值对元数据，如 user.myapp.version=1.0）。默认不保留 ACL 和 XAttr 是因为它们受集群环境配置影响大，跨集群迁移时两端映射可能不一致
-i	忽略失败，继续拷贝剩余文件。适合大规模生产同步，避免单个文件失败导致整体重跑。配合 -update 使用效果最佳
-v	详细日志，记录每个文件的拷贝详情（成功/跳过/失败）
-tmp <path>	-atomic 模式下的临时路径

并行度与性能

参数	说明	默认值
-m <num>	控制最大 Mapper 数（分片数上限），默认的均匀大小分配策略 UniformSizeInputFormat 按 totalSize / -m 的阈值切分。文件数远大于 -m 时能趋近或达到上限；文件数少于 -m 时无法达到上限（分片数 ≤ 文件数，小文件合并会使分片更少）	20
-bandwidth <mb>	单个 Mapper 的带宽上限（MB/s），总带宽 ≈ -bandwidth × mapper数	不限
-numListstatusThreads <n>	CopyListing 阶段并发遍历子目录的线程数，上限 40。适合源目录树子目录多的场景	1（单线程）
-strategy <dynamic/uniformsize>	文件分配策略	uniformsize
-copyBufferSize <num>	每个 Mapper 拷贝文件时的读/写缓冲区大小（bytes），调大能减少 IO 次数	8192

通用 Hadoop 参数

参数	说明
-D <property=value>	设置 Hadoop 配置项，传递给 MapReduce 作业。可用于调整队列、限速、token 排除、认证等任意配置

-D 必须放在命令最前面，属于 Hadoop 通用选项（generic options），放在 -p、-update 等 DistCp 参数之后会被解析为源路径。

-D 常用示例：

• -D mapreduce.job.queuename=root.distcp — 指定作业队列
• -D mapreduce.job.hdfs-servers.token-renewal.exclude=<host> — 排除目标集群的 token 自动续期（跨集群同步时常用）
• -D distcp.bandwidth.mb=50 — 通过配置设置带宽上限（等效 -bandwidth 50）
• -D mapreduce.job.name=my_distcp_job — 指定 MR 作业名称（实际显示为 distcp: my_distcp_job），方便 YARN 界面识别
• -D ipc.client.fallback-to-simple-auth-allowed=true — 跨集群同步时两端 Kerberos 配置不一致时，允许回退到简单认证

性能对比

DistCp vs hadoop fs -cp

对比维度	hadoop fs -cp	DistCp
执行模式	单线程客户端执行	MapReduce 分布式执行
并行度	1	默认上限 20（-m 控制），文件数远大于 -m 时能趋近或达到上限；文件数少于 -m 时达不到上限（分片数 ≤ 文件数，小文件合并会使分片更少）
跨集群	支持，但单线程串行	原生支持，分布式并行
增量同步	不支持	支持（-update/-diff）
失败处理	失败即停	支持 -i 跳过失败继续
带宽控制	不支持	支持 -bandwidth
属性保留	-p 支持	-p 支持更完整（ACL/XAttr）
适用数据量	MB~GB	GB~PB

fs -cp 也支持跨集群？ 是的。fs -cp 检测到源和目标 URI 不同时会走 FileUtil.copy() 的跨文件系统路径，分别创建源端和目标端的 DistributedFileSystem 实例，先 read 再 write。但这是单线程串行拷贝，目标存在时报错退出，不支持增量同步，数据量大时不实用。

实测参考（3 节点集群，100GB 数据，千兆网络）：

方式	耗时	说明
fs -cp	~45min	单线程，IO 和网络利用率低
DistCp 默认（20 Map）	~3min	分布式并行，充分打满带宽
DistCp 限速（10MB/s per Map）	~10min	带宽可控，适合生产限流

粗略估算，实际看集群规模和文件分布。

-p 权限检查规则：fs -cp -p vs DistCp -p

-p 参数会尝试保留文件的 owner、group、permission、timestamp 等属性。fs -cp 和 DistCp 在权限检查上差别在于触发时机不同。

fs -cp -p 的权限检查

-p 的本质是：遍历源路径下每个文件和目录，把各自的 owner/group 设置到目标路径的对应文件/目录上。顺序先处理文件（深度优先），再处理父目录。

每次 setOwner 调用都会触发 HDFS 权限检查（FSDirAttrOp.setOwner）：

1. 先检查是否超级用户 → isSuperUser()，是则跳过后续检查直接放行
2. 不是超级用户时，比较当前用户与 setOwner 要设置的 owner 参数（源文件的 owner）：
   • 要设置的 owner 与当前用户不同 → is not a super user (non-super user cannot change owner)
   • 要设置的 owner 与当前用户相同，但 group 参数（源文件的 group）不同 → 检查当前用户是否在目标 group 中
     • 是 → 不报错
     • 否 → does not belong to <group>

fs -cp -p 对每个文件/目录都调 setOwner，每次检查，不满足就抛异常。

不同权限组合的报错表现（以 hive 用户执行 fs -cp -p 为例，hive 属于 hadoop 组，不属于 hdfs 组）：

目录 owner	文件 owner	报错（按顺序，先文件后目录）
hdfs:hdfs	hive:hdfs	文件→owner相同，group需设hdfs，hive不在hdfs组→does not belong to hdfs 目录→owner不同→is not a super user
hive:hdfs	hdfs:hdfs	文件→owner不同→is not a super user 目录→owner相同，group需设hdfs，hive不在hdfs组→does not belong to hdfs
hive:hdfs	hive:hdfs	文件→owner相同，group需设hdfs，hive不在hdfs组→does not belong to hdfs 目录→owner相同，group需设hdfs，hive不在hdfs组→does not belong to hdfs
kafka:hdfs	kafka:hdfs	文件→owner不同→is not a super user 目录→owner不同→is not a super user
hive:hadoop	hive:hadoop	无报错（owner相同，group相同-hadoop组，hive在hadoop组）

---

DistCp -p 的权限检查

DistCp 在 DistCpUtils.preserve() 中处理属性保留，先比较目标属性和源属性，一致就跳过：不调 setOwner，不触发检查，不报错。

// DistCpUtils.preserve() — 以 owner 为例
if (attributes.contains(FileAttribute.USER) &&
    !user.equals(srcFileStatus.getOwner())) {        // ← 目标owner != 源owner？
  user = srcFileStatus.getOwner();
  chown = true;
}
// 只有 chown=true 时才调用 targetFS.setOwner(path, user, group)

---

DistCp 目录权限保留的特殊性

DistCp 的目录属性不在 CopyMapper 中处理，而是由作业完成后 CopyCommitter 统一处理（preserveFileAttributesForDirectories），它只遍历源文件列表中的目录项：

// CopyCommitter.preserveFileAttributesForDirectories()
if (! srcFileStatus.isDirectory()) continue;      // ← 非目录则跳过，只保留目录项

Path targetFile = new Path(targetRoot + "/" + srcRelPath);
if (targetRoot.equals(targetFile) && syncOrOverwrite) continue;  // ← root skip 逻辑

• 文件：setOwner 在 CopyMapper 中直接调用，尝试改 owner
• 目录：由 CopyCommitter 在作业完成后处理
  • 根目录在 -update 或 -overwrite 时被跳过
  • 子目录同样走 preserve() 先比较再操作

完整对比：fs -cp -p vs DistCp -p

对比项	fs -cp -p	DistCp -p
范围	文件+目录递归全部	文件（CopyMapper 拷贝时处理）+ 目录（CopyCommitter 统一处理，根目录仅在 -update/-overwrite 时跳过，其余情况正常保留）
触发时机	每级路径逐个 setOwner	先比较目标属性与源属性，一致则跳过

举例说明触发时机差异：hive 用户拷贝 hive:hdfs 文件时，fs -cp -p 逐级调用 setOwner，hive 不在 hdfs 组直接报错；而 DistCp -p 先比较目标属性与源属性，两者一致（都是 hive:hdfs）则跳过不操作，不触发检查，因此不报错。

HDFS 超级用户配置

dfs.permissions.superusergroup 指定超级用户组（如 hdfs），由 Active NameNode 所在节点读取 /etc/group。HA 集群下需要修改 Active NN 所在节点的组配置，建议两个 NameNode 节点都改。

# 确认 Active NameNode
hdfs haadmin -getServiceState nn1
hdfs haadmin -getServiceState nn2

将用户加入 superusergroup 配置指定的组（如 hdfs）：

usermod -aG hdfs hive
# 或：gpasswd -a hive hdfs

执行后验证：id hive 输出应包含 hdfs 组。

原理分析

为什么 DistCp 性能好

DistCp 是 Map-only 作业，多个 Mapper 并行拷贝文件。单机一次拷一个文件，DistCp 同时拷多个，总吞吐量随 Mapper 数线性增长。

跨集群时数据本地性取决于作业提交到哪个集群：

• 提交到源集群：Mapper 在源端 DataNode 上运行，读源数据走本地
• 提交到目标集群：Mapper 在目标端 DataNode 上运行，读源数据得跨网络

读和写至少有一端跨网络，但并行 IO 还是远好于单线程。

为何可以支持增量同步

DistCp 的增量同步分几种：

-update — 基于文件元数据比对

首次拷贝：全部拷贝
         ↓
第二次：对每个源文件：
  1. 查询目标端是否存在同名文件
  2. 不存在 → 拷贝（新增）
  3. 存在则比较文件大小和修改时间
  4. 大小或修改时间不同 → 拷贝（覆盖或追加）
  5. 大小和修改时间都相同 → 进一步比对 CRC 校验码
     - CRC 一致 → 跳过（内容相同）
     - CRC 不同 → 拷贝覆盖（内容已变）
     - 加了 -skipcrccheck → 跳过 CRC 比对，直接将文件视为一致

这种方式需要列举两端文件列表做比对，源端和目标端各一次 listStatus 调用，文件数量大时遍历开销不可忽略。

-append — 增量追加（依赖 -update）

文件尾部追加的场景（比如日志文件），配合 -update 只传新增部分，不传整个文件。原理：比较源端和目标端同名文件大小，从源文件偏移量 = 目标文件大小的位置开始读取，append 到目标文件末尾。

-append 使用限制与风险：

• HDFS 文件必须支持 append（部分文件系统有限制）
• 不做内容比对，纯靠偏移量截取。如果源端文件是覆盖重写（非尾部追加），追加的数据会与目标端已有内容不一致
• 仅适用于尾部追加写入的日志文件场景，生产环境需谨慎

-diff — 基于 HDFS 快照差异

前提：源目录必须支持快照（HDFS），且已开启快照。目标目录内容必须与源目录 snap_base 时刻的内容一致（首次需全量同步一次作为基准），目标端不需要创建快照。

# 首次使用需要手动准备快照：
hdfs dfsadmin -allowSnapshot /path/to/data                          # 1. 开启快照
hdfs dfs -createSnapshot /path/to/data snap_base                     # 2. 创建基准快照
# ... 数据发生变化 ...
hdfs dfs -createSnapshot /path/to/data snap_current                  # 3. 创建当前快照

hadoop distcp -diff snap_base snap_current                           # 4. 基于快照差异同步
    hdfs://source:8020/path/to/data hdfs://target:8020/path/to/data

# 周期性同步可以脚本化：同步完成后创建新快照作为下一轮基准
hdfs dfs -deleteSnapshot /path/to/data snap_base                     # 删除旧的基准
hdfs dfs -createSnapshot /path/to/data snap_base                     # 创建新基准（与目标端一致）

Diff 差异获取流程：

1. 创建两个快照：之前的状态（snap_base），当前状态（snap_current）
2. 调用 HDFS 快照 diff API 获取（snap_base, snap_current）之间的差异
3. 差异列表直接作为待拷贝文件列表
4. 无需遍历目录树，零比对开销

百万级文件目录，快照差异响应秒级，listStatus 遍历可能要分钟级。

不过 -diff 比 -update 繁琐：得先开快照、创建快照，周期中还要轮换快照（删旧的、建新的）。-update 一行命令就行。文件量大、同步周期短才值得上 -diff，大部分场景 -update 够用。

增量同步决策树

是否开启 -diff 且快照可用？
  ├─ 是 → 基于快照差异获取变更 → 高性能增量
  └─ 否 → 是否开启 -update？
       ├─ 是 → 遍历两端文件列表逐一比对 → 标准增量
       └─ 否 → 全量拷贝

源码分析

类路径：hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/。

类结构

DistCp              — 入口，解析参数，构建 MR 作业
DistCpOptions       — 参数封装
DistCpOptionsParser — 命令行参数解析
CopyListing         — 构建待拷贝文件列表
  ├─ SimpleCopyListing    — 默认，遍历源目录生成列表
  └─ GlobbedCopyListing   — 支持通配符
CopyMapper — 拷贝逻辑（Map 任务）
CopyCommitter       — 作业提交/清理
DynamicInputFormat  — 动态分配策略的 InputFormat
UniformSizeInputFormat — 均匀大小分配策略

DistCp 入口：main → run

DistCp.main()
  └─ DistCp.run()
       ├─ 1. 解析参数 → DistCpOptions
       ├─ 2. 创建 CopyListing 生成待拷贝文件列表（SequenceFile）
       ├─ 3. 构建 Job 配置，设置 InputFormat（Dynamic 或 UniformSize）
       ├─ 4. 设置 Mapper（CopyMapper）及配置
       ├─ 5. job.waitForCompletion(true)
       └─ 6. CopyCommitter 清理 + 输出结果

入口代码简化示意：

// DistCp.java
public static void main(String[] args) {
  DistCp distCp = new DistCp(configuration, options);
  distCp.execute();  // 调用 run()
}

public int run() {
  // 1. 构建 Mapper 配置
  // 2. 创建 CopyListing 生成文件列表
  // 3. 设置 Job（InputFormat → Mapper → OutputFormat）
  // 4. job.waitForCompletion
  // 5. 如果 -atomic，做 mv 提交
}

**作业命名逻辑**（`createJob()`）：
```java
// DistCp.java
String jobName = "distcp";
String userChosenName = getConf().get(JobContext.JOB_NAME);
if (userChosenName != null)
    jobName += ": " + userChosenName;
job.setJobName(jobName);

通过 -D mapreduce.job.name=<name> 自定义名称后，实际显示为 distcp: <name>。

CopyMapper — 拷贝逻辑

CopyMapper.map() 处理单个文件的拷贝决策：

CopyMapper.map()
  ├─ checkFileExist(targetPath)   → 检查目标文件是否存在及元数据
  ├─ 决策是否需要拷贝：
  │   ├─ -overwrite 模式 → 总是拷贝
  │   ├─ -update 模式 → 比较文件大小 + 修改时间 → 不同则拷贝
  │   ├─ -append 模式 → 文件大小变化且内容追加 → 增量追加
  │   └─ 默认 → 目标已存在则跳过
  ├─ 执行拷贝：
  │   ├─ 如果 -append → 获取偏移量 → append 写入
  │   └─ 否则 → 完整拷贝（FileUtil.copy or 流式写入）
  └─ 保留属性（-p 参数）→ 权限、时间戳、ACL…

决策逻辑：

// CopyMapper.java - shouldCopy() 简化逻辑
boolean shouldCopy(FileStatus source, FileStatus target) {
  if (overwrite) return true;                           // -overwrite: 无条件全量覆盖
  if (target == null) return true;                      // 目标不存在
  if (update) {                                         // -update: 增量比对
    if (source.getLen() != target.getLen()) return true;  // 大小变化
    if (source.getModificationTime() !=
        target.getModificationTime()) return true;        // 时间变化
  }
  return false;                                         // 默认: 目标存在则跳过
}

-diff 的快照差异原理

-diff 模式不走 CopyMapper 的元数据比对，而是通过 SnapshotDiffCopyListing 获取快照差异列表：

SnapshotDiffCopyListing
  └─ 调用 FileSystem.getSnapshotDiffReport(sourcePath, snap1, snap2)
       └─ 返回 SnapshotDiffReport（包含：新增/修改/删除 文件列表）

这些差异条目直接作为 InputFormat 的输入
Mapper 对每个差异条目执行：新增 → 拷贝，修改 → 覆盖，删除 → 删除

避免了 listStatus 遍历，百万级目录秒级响应。

文件分配策略

DistCp 默认使用 UniformSizeInputFormat，按 totalSize / -m 的阈值切分分片：

UniformSizeInputFormat:
  1. 计算阈值 nBytesPerSplit = ceil(totalSize / -m)
  2. 遍历排序后的文件列表，累加文件大小
  3. 累加值 + 下一个文件 > nBytesPerSplit 时切分（不跨文件切分）
  4. 每个 Mapper 串行拷贝分到的文件（逐个调用 map()）

  规则：
  - 小文件合并：多个小文件累计超过阈值时被合并到同一分片
  - 大文件独占：单个文件 > 阈值也不会被拆分
  - -m 控制切分粒度，不是上限。m 越小，分片越少

DynamicInputFormat:
  1. 每个 Mapper 启动时只分配一个文件
  2. 处理完立即申请下一个
  3. 解决"大文件卡住一个 Mapper，其他空闲"的问题

两种策略的适用场景：

策略	适合场景	缺点
uniformsize	文件大小分布均匀	少数大文件会导致负载不均
dynamic	文件大小差异极大（混合大文件和小文件）	调度开销略高

-bandwidth 限速原理

每个 Mapper 的写入流上加令牌桶限速：

CopyMapper 写入目标 HDFS
  └─ FsOutput 包装 RateLimitingInputStream
       └─ 令牌桶：每毫秒补充 (bandwidth * 1024 * 1024 / 1000) 字节
            └─ 写入前获取令牌，不足则阻塞

每个 Mapper 独立限速，总带宽 ≈ -bandwidth × mapper数。-bandwidth 10 -m 20 时总带宽约 200MB/s。

-i 适用场景

-i（ignoreFailures）控制单个文件拷贝失败时整个作业是否继续。

建议加 -i 的场景：

• 海量小文件（数十万级），个别失败是常态
• 生产增量同步，目标端文件可能正被读取
• 非关键数据同步，允许部分失败

不建议加 -i 的场景：

• 文件很少，失败重跑成本低
• 要求目标端跟源端严格一致
• 数据迁移验收阶段，需确认 100% 成功

使用注意事项

1. 目标端 HDFS 空间：DistCp 不会预先检查，执行前自己确认。
2. Namenode 压力：文件千万级时，listStatus 和 Mapper 创建的文件数都会压 NN。优先用 -diff 或调 -numListstatusThreads。
3. 网络带宽：DistCp 默认不限速，20 个 Mapper 可能打满千兆网卡。生产用 -bandwidth 限速。
4. 权限与 ACL：跨集群同步两端 UID/GID 可能不一致，-p 可能导致目标端权限错乱。按场景决定是否用 -p。
5. 多版本兼容：不同 Hadoop 版本 RPC 可能有兼容问题。跨大版本同步先测试。Hadoop 2.x 到 3.x 建议用兼容客户端。

总结

• 性能：Map-only 作业，多 Mapper 并行拷贝
• 增量：-update 做元数据比对，-diff 用快照差异，-append 靠偏移量追加
• 控制：-bandwidth 限速、-m 调并行度、-strategy 切换分配策略

赏

感谢您的支持!

支付宝

微信

本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Hadoop DistCp 使用详解与原理分析

本文链接：https://dongkelun.com/2026/07/31/hadoop/hadoop-distcp/

欢迎关注我的公众号