Hadoop DistCp 源码分析：命令入口到核心调用链

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

前言

上一篇文章《Hadoop DistCp 使用详解与原理分析》介绍了 DistCp 的用法和基本原理。本文从源码出发，分析两个问题：

  1. hadoop distcp 命令从敲下回车到 Java 代码，经历了什么？
  2. DistCp 的 13 个核心类之间是如何协作的？

基于 Hadoop 3.1.1 源码，所有引用均标注文件路径和方法名。

命令入口：从 Shell 到 Java

Linux Bash 分发（bin/hadoop）

Linux 的 bin/hadoop 脚本运行时，走到 hadoopcmd_case 函数。distcp 不在 case 列表中明确列出的子命令中，因此落到 *) 通配分支：

1
2
3
4
5
6
7
# bin/hadoop : hadoopcmd_case() 中的 *) 分支
*)
  HADOOP_CLASSNAME="${subcmd}"
  if ! hadoop_validate_classname "${HADOOP_CLASSNAME}"; then
    hadoop_exit_with_usage 1
  fi
;;

这里将 HADOOP_CLASSNAME 直接设为 "distcp"。然后脚本末尾调用 hadoop_generic_java_subcmd_handler，该函数拼接类路径并执行java命令。

不过 distcp 实际是一个 Java 类的工具，Linux 下更常见的做法是通过 hadoop_subcommand_distcp 动态函数来处理。但 Hadoop 3.1.1 的 bin/hadoop 中没有为 distcp 定义单独的子命令函数，因此最终会走到通用处理逻辑，通过 hadoop_add_classpath 把 TOOL_PATH 加入类路径后，以 org.apache.hadoop.tools.DistCp 启动。

Windows CMD 分发（bin/hadoop.cmd）

CMD 脚本的做法更直观，维护了一个 corecommands 列表，distcp 在其中：

1
2
3
4
5
6
7
8
rem bin/hadoop.cmd corecommands 列表
set corecommands=fs version jar checknative conftest distch distcp daemonlog ...
for %%i in ( %corecommands% ) do (
  if %hadoop-command% == %%i set corecommand=true
)
if defined corecommand (
  call :%hadoop-command%
)

匹配后跳转到 :distcp 标签，直接指定主类：

1
2
3
4
5
rem bin/hadoop.cmd :distcp 标签
:distcp
  set CLASS=org.apache.hadoop.tools.DistCp
  set CLASSPATH=%CLASSPATH%;%TOOL_PATH%
  goto :eof

最终执行：

1
java ... org.apache.hadoop.tools.DistCp <剩余参数>

备选入口：mapred distcp

bin/mapred（Linux）和 bin/mapred.cmd（Windows）同样支持 mapred distcp，指向同一个类 org.apache.hadoop.tools.DistCp。

DistCp.main()

1
2
3
4
5
6
7
8
// DistCp.java main()
public static void main(String argv[]) {
  DistCp distCp = new DistCp();  // 无参构造，用于 ToolRunner
  Cleanup CLEANUP = new Cleanup(distCp);
  ShutdownHookManager.get().addShutdownHook(CLEANUP, SHUTDOWN_HOOK_PRIORITY);
  exitCode = ToolRunner.run(getDefaultConf(), distCp, argv);
  System.exit(exitCode);
}

getDefaultConf() 加载 distcp-default.xml 和 distcp-site.xml 两个配置文件。

ToolRunner 回调

1
2
3
4
5
6
7
8
// ToolRunner.java run()
public static int run(Configuration conf, Tool tool, String[] args) {
  GenericOptionsParser parser = new GenericOptionsParser(conf, args);
  // ① 过滤 -D、-conf 等 Hadoop 通用参数
  tool.setConf(conf);
  String[] toolArgs = parser.getRemainingArgs();
  return tool.run(toolArgs);  // ② 回调 DistCp.run()
}

回到 DistCp.run()：

1
2
3
4
5
6
7
8
// DistCp.java run()
public int run(String[] argv) {
  // ① 参数解析
  context = new DistCpContext(OptionsParser.parse(argv));
  checkSplitLargeFile();          // ② 校验大文件分块支持
  setTargetPathExists();          // ③ 检查目标路径是否存在
  execute();                      // ④ 执行拷贝
}

参数解析

OptionsParser.parse()

OptionsParser 基于 Apache Commons CLI 实现。关键步骤：

1
2
3
4
5
6
7
8
9
10
// OptionsParser.java parse()
public static DistCpOptions parse(String[] args) {
  // 1. 注册 CLI 选项（遍历 DistCpOptionSwitch 枚举）
  // 2. CustomParser 解析命令行参数
  // 3. 分离源路径和目标路径：
  //    - 最后一个参数 = targetPath
  //    - 之前的所有参数 = sourcePaths（或 -f 指定文件列表）
  // 4. Builder 模式逐个设置选项
  // 5. builder.build() 触发校验
}

-D 参数在更早的 GenericOptionsParser 阶段已被过滤掉，不会到 OptionsParser 这一层。

DistCpOptions.Builder.validate()

参数校验规则（DistCpOptions.java Builder.validate()）：

  • -overwrite 与 -update 互斥
  • -atomic 不能与 -update/-overwrite 一起使用
  • -delete 与 -diff/-rdiff 互斥
  • -append 必须配合 -update
  • -append 不能与 -skipcrccheck 一起使用
  • -diff/-rdiff 必须配合 -update
  • -v 必须配合 -log

校验通过后，才创建不可变的 DistCpOptions 对象。

DistCpContext

DistCpContext 包装了不可变的 DistCpOptions 和可变的运行时状态：

1
2
3
4
5
6
7
// DistCpContext.java
public class DistCpContext {
  private final DistCpOptions options;  // 不可变参数
  private List<Path> sourcePaths;       // 运行时路径列表（可被改写）
  private boolean targetPathExists;     // 目标是否存在
  private boolean preserveRawXattrs;    // 是否保留 raw.* xattr
}

sourcePaths 是可变的——GlobbedCopyListing 展开通配符后、DistCpSync 处理快照后都会改写它。

核心调用链

总体流程

1
2
3
4
5
6
7
8
9
10
11
DistCp.main()
  → ToolRunner.run()
    → DistCp.run()
      → OptionsParser.parse() → DistCpOptions (参数解析)
      → DistCpContext (运行时上下文)
      → execute()
        → createAndSubmitJob()
          → createJob()         配置 MR 作业
          → prepareFileListing() 生成拷贝文件列表
          → job.submit()        提交 MR 作业
        → waitForJobCompletion() 等待完成

createJob() — 配置 MR 作业

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// DistCp.java createJob()
private Job createJob() throws IOException {
  Job job = Job.getInstance(getConf());

  // InputFormat：通过配置选择 UniformSizeInputFormat 或 DynamicInputFormat
  job.setInputFormatClass(DistCpUtils.getStrategy(getConf(), context));

  // 纯 Map 作业
  job.setMapperClass(CopyMapper.class);
  job.setNumReduceTasks(0);

  // OutputFormat
  job.setOutputFormatClass(CopyOutputFormat.class);
  configureOutputFormat(job);  // 设置工作目录、日志路径

  context.appendToConf(job.getConfiguration());  // 选项写入配置，传递到 Mapper
  return job;
}

InputFormat 选择策略

1
2
3
4
5
6
7
8
9
10
// DistCpUtils.java getStrategy()
public static Class<? extends InputFormat> getStrategy(Configuration conf,
    DistCpContext context) {
  // 配置键：distcp.{策略名}.strategy.impl
  String confLabel = "distcp."
      + StringUtils.toLowerCase(context.getCopyStrategy())
      + ".strategy" + ".impl";
  // 默认回退：UniformSizeInputFormat
  return conf.getClass(confLabel, UniformSizeInputFormat.class, InputFormat.class);
}

context.getCopyStrategy() 默认值为 "uniformsize"（定义在 DistCpOptions.Builder 中），拼接后为 distcp.uniformsize.strategy.impl。

对比 distcp-default.xml 中的配置：

1
2
3
4
5
6
7
8
9
<!-- distcp-default.xml -->
<property>
    <name>distcp.static.strategy.impl</name>
    <value>org.apache.hadoop.tools.mapred.UniformSizeInputFormat</value>
</property>
<property>
    <name>distcp.dynamic.strategy.impl</name>
    <value>org.apache.hadoop.tools.mapred.lib.DynamicInputFormat</value>
</property>

注意：代码拼接的是 distcp.uniformsize.strategy.impl，而 XML 中定义的键是 distcp.static.strategy.impl。两者不匹配，因此代码中的 conf.getClass() 会查找不到，回退到默认值 UniformSizeInputFormat.class。这意味着即使想通过 distcp-default.xml 替换默认 InputFormat，配置键 distcp.static.strategy.impl 也不会被代码使用，需通过 distcp.uniformsize.strategy.impl 这个键来配置。

prepareFileListing() — 文件列表生成（核心分歧点）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// DistCp.java prepareFileListing()
private void prepareFileListing(Job job) throws Exception {
  if (context.shouldUseSnapshotDiff()) {
    // 路径 A：-diff/-rdiff 快照差异模式
    DistCpSync distCpSync = new DistCpSync(context, getConf());
    if (distCpSync.sync()) {
      createInputFileListingWithDiff(job, distCpSync);
    } else {
      throw new Exception("DistCp sync failed");
    }
  } else {
    // 路径 B：常规模式
    createInputFileListing(job);
  }
}

路径 A：快照差异（-diff / -rdiff）

1
2
3
4
5
6
7
// DistCp.java createInputFileListingWithDiff()
private Path createInputFileListingWithDiff(Job job, DistCpSync distCpSync) {
  // 直接创建 SimpleCopyListing（含 distCpSync 引用），不走工厂方法
  CopyListing copyListing = new SimpleCopyListing(
      job.getConfiguration(), job.getCredentials(), distCpSync);
  copyListing.buildListing(fileListingPath, context);
}

路径 B：常规模式

1
2
3
4
5
6
7
// DistCp.java createInputFileListing()
protected Path createInputFileListing(Job job) throws IOException {
  // 通过工厂方法选择 CopyListing 实现
  CopyListing copyListing = CopyListing.getCopyListing(
      job.getConfiguration(), job.getCredentials(), context);
  copyListing.buildListing(fileListingPath, context);
}

CopyListing 工厂方法 — 三层选择

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// CopyListing.java getCopyListing()
public static CopyListing getCopyListing(Configuration configuration,
    Credentials credentials, DistCpContext context) throws IOException {

  String customClass = configuration.get(
      DistCpConstants.CONF_LABEL_COPY_LISTING_CLASS, "");

  if (!customClass.isEmpty()) {
    // 优先级 1：用户通过 distcp.copy.listing.class 自定义
    return instantiateClass(customClass);
  }
  if (context.getSourceFileListing() == null) {
    // 优先级 2：命令行源路径 → GlobbedCopyListing（支持通配符）
    return new GlobbedCopyListing(configuration, credentials);
  } else {
    // 优先级 3：-f 指定文件列表 → FileBasedCopyListing
    return new FileBasedCopyListing(configuration, credentials);
  }
}

CopyListing 三层委托链

1
2
3
FileBasedCopyListing           — 从 -f 文件读取路径列表
  → 委托 GlobbedCopyListing    — 展开通配符（globStatus）
    → 委托 SimpleCopyListing   — 递归遍历目录树，写 SequenceFile

每个委托层都会改写 context.setSourcePaths()，将展开后的路径传递给下一层。

SimpleCopyListing.doBuildListing() 是核心：

  • 使用 ProducerConsumer 模式多线程递归遍历目录
  • 线程数由 -numListstatusThreads 控制（默认 1，上限 40）
  • 每个文件/目录写一条 SequenceFile 记录
  • key = 相对路径（Text），value = 文件状态（CopyListingFileStatus）

CopyListing.buildListing() 作为模板方法，统一流程：

1
2
3
4
5
6
7
8
9
// CopyListing.java buildListing()
public final void buildListing(Path pathToListFile, DistCpContext distCpContext) {
  validatePaths(distCpContext);                  // 1. 校验路径
  doBuildListing(pathToListFile, distCpContext);  // 2. 子类构建列表
  config.set(CONF_LABEL_LISTING_FILE_PATH, ...);  // 3. 写入配置
  config.setLong(CONF_LABEL_TOTAL_BYTES_TO_BE_COPIED, ...);
  config.setLong(CONF_LABEL_TOTAL_NUMBER_OF_RECORDS, ...);
  validateFinalListing(pathToListFile, ...);       // 4. 校验重复 + ACL/XAttr
}

DistCpSync — 快照差异路径

当 -diff 或 -rdiff 启用时，prepareFileListing() 先调 DistCpSync.sync() 做同步：

1
2
3
4
5
6
7
8
9
10
11
12
DistCpSync.sync()
  1. preSyncCheck()
     - 只能有一个源目录
     - 源和目标都必须是 DistributedFileSystem
     - 目标端从 fromSnapshot 以来不能有变化
  2. getAllDiffs()
     - 调用 fs.getSnapshotDiffReport(ssDir, from, to)
     - 按 DiffType 分类（MODIFY/CREATE/DELETE/RENAME）
  3. syncDiff()
     - moveToTmpDir()：将 rename/delete 的文件移到临时目录（按 source 倒序）
     - moveToTarget()：从临时目录 rename 到最终位置（按 target 正序）
  4. 改写 sourcePaths 为快照路径

随后 prepareDiffListForCopyListing() 只返回 MODIFY 和 CREATE 条目（DELETE/RENAME 已由 sync 处理），给 SimpleCopyListing.doBuildListingWithSnapshotDiff() 使用。

UniformSizeInputFormat 与 DynamicInputFormat

UniformSizeInputFormat（默认）

按字节数均匀切分分片：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// UniformSizeInputFormat.java getSplits()
long nBytesPerSplit = (long) Math.ceil(totalSizeBytes * 1.0 / numSplits);

while (reader.next(srcRelPath, srcFileStatus)) {
  // 累积大小超过阈值时切分，不跨文件
  if (currentSplitSize + srcFileStatus.getChunkLength() > nBytesPerSplit
      && lastPosition != 0) {
    splits.add(new FileSplit(listingFilePath, lastSplitStart, ...));
    lastSplitStart = lastPosition;
    currentSplitSize = 0;
  }
  currentSplitSize += srcFileStatus.getChunkLength();
  lastPosition = reader.getPosition();
}

DynamicInputFormat

Worker 模式——文件列表拆成小 chunk，Mapper 处理完一个领下一个：

1
2
3
4
5
splitCopyListingIntoChunksWithShuffle()
  1. 计算每个 chunk 的条目数：numEntriesPerChunk = ceil(numRecords / (splitRatio * numMaps))
  2. 遍历 SequenceFile，轮询写入多个 chunk 文件
  3. 创建 N 个空 split，每个关联一个 chunk（初始分配）
  4. Mapper 的 DynamicRecordReader 处理完当前 chunk 后自动领下一个

适合文件大小差异大的场景，避免 UniformSize 中”大文件卡住一个 Mapper，其他空闲”的问题。

CopyMapper 端：实际文件拷贝

整体流程

1
2
3
4
5
6
7
8
9
10
11
12
13
CopyMapper.setup()
  → 从 Configuration 读取参数（syncFolders、skipCrc、overWrite 等）

CopyMapper.map(relPath, sourceFileStatus)
  → 构造 targetPath = targetWorkPath + relPath
  → 获取 sourceCurrStatus（实时 FileStatus）
  → 若为目录：createTargetDirsWithRetry()
  → checkUpdate() 决策动作
    ├── SKIP：目标存在且一致
    ├── APPEND：尾部追加
    └── OVERWRITE：覆盖写入
  → copyFileWithRetry() 执行拷贝
  → DistCpUtils.preserve() 保留文件属性

checkUpdate() — 拷贝决策

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// CopyMapper.java checkUpdate()
private FileAction checkUpdate(FileSystem sourceFS,
    CopyListingFileStatus source, Path target, FileStatus targetFileStatus) {
  if (targetFileStatus != null && !overWrite) {
    if (canSkip(sourceFS, source, targetFileStatus)) {
      return FileAction.SKIP;           // 目标存在且一致
    } else if (append) {
      long targetLen = targetFileStatus.getLen();
      if (targetLen < source.getLen()) {
        // 尾部校验和一致 → APPEND
        if (sourceChecksum.equals(targetFS.getFileChecksum(target))) {
          return FileAction.APPEND;
        }
      }
    }
  }
  return FileAction.OVERWRITE;          // 覆盖
}

canSkip() — 能否跳过

1
2
3
4
5
6
7
8
9
10
// CopyMapper.java canSkip()
private boolean canSkip(...) {
  if (!syncFolders) return true;  // 非 -update：目标存在即跳过
  // -update 模式：比较长度 + BlockSize + CRC
  boolean sameLength = target.getLen() == source.getLen();
  if (sameLength && sameBlockSize) {
    return skipCrc || DistCpUtils.checksumsAreEqual(...);
  }
  return false;  // 大小或时间不一致，需要拷贝
}

RetriableFileCopyCommand — 带重试的文件拷贝

1
2
3
4
5
6
7
8
9
10
11
12
RetriableFileCopyCommand.execute()
  → doCopy()
    → copyToFile()
      → OVERWRITE：创建临时文件 .distcp.tmp.attempt_xxx
      → APPEND：直接 append
    → copyBytes()
      → getInputStream() 打开源，包装 ThrottledInputStream 限速
      → seekIfRequired() 定位到偏移量
      → 循环 read → write
    → compareFileLengths()   校验长度
    → compareCheckSums()     校验 CRC
    → promoteTmpToTarget()   rename 临时文件 → 目标

失败后自动重试（继承 RetriableCommand，默认 3 次指数退避）。

ThrottledInputStream — 令牌桶限速

1
2
3
4
5
6
7
// ThrottledInputStream.java throttle()
private void throttle() throws IOException {
  while (getBytesPerSec() > maxBytesPerSec) {
    Thread.sleep(SLEEP_DURATION_MS);  // 50ms
    totalSleepTime += SLEEP_DURATION_MS;
  }
}

速率计算为整个生命周期平均值，短期可能超出但长期趋近设定值。

CopyCommitter — 作业后处理

commitJob() 在 MapReduce 框架回调中执行：

1
2
3
4
5
6
7
8
9
10
CopyCommitter.commitJob()
  1. concatFileChunks()        大文件分块合并（HDFS concat）
  2. super.commitJob()         FileOutputCommitter 标准提交
  3. cleanupTempFiles()        清理 .distcp.tmp.attempt_* 残留
  4. preserveFileAttributesForDirectories()  保留目录属性
  5. 分支（互斥）：
     ├─ -delete：deleteMissing()  比对源/目标列表，删除目标多余文件
     ├─ -atomic：commitData()     rename 临时目录 → 最终目录
     └─ -track： trackMissing()   保存缺失文件列表
  6. cleanup()                  删除 metaFolder

类运行位置汇总

类 运行端 职责
DistCp Driver 入口，配置作业，提交
DistCpOptions Driver 不可变参数模型，通过 Configuration 传递到 Mapper
OptionsParser Driver 命令行解析
DistCpContext Driver 运行时上下文
DistCpConstants Driver + Mapper 常量定义
DistCpSync Driver 快照差异同步
CopyListing (abstract) Driver 文件列表抽象
SimpleCopyListing Driver 递归遍历目录树，生成 SequenceFile
GlobbedCopyListing Driver 展开通配符，委托 SimpleCopyListing
FileBasedCopyListing Driver 从文件读路径，委托 GlobbedCopyListing
CopyMapper Mapper 执行文件拷贝
RetriableFileCopyCommand Mapper 带重试的拷贝（包装限速流）
ThrottledInputStream Mapper 令牌桶带宽限速
UniformSizeInputFormat Driver + Mapper 按字节数均匀分片
DynamicInputFormat Driver + Mapper 动态 Worker 模式分片
CopyCommitter Driver 作业后处理（MR 框架回调）
CopyOutputFormat Driver + Mapper OutputFormat 配置
DistCpUtils Driver + Mapper 工具方法
DiffInfo Driver 快照差异数据载体

调用流程图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
hadoop distcp -update src tgt
  │
  ├─ bin/hadoop.cmd → set CLASS=org.apache.hadoop.tools.DistCp   [cmd脚本]
  │
  ├─ DistCp.main()                                                [DistCp.java]
  │   └─ ToolRunner.run(getDefaultConf(), distCp, argv)
  │        └─ GenericOptionsParser 过滤 -D 参数
  │
  ├─ DistCp.run()
  │   ├─ OptionsParser.parse(argv) → DistCpOptions
  │   │   └─ Builder.build() → validate() 校验参数
  │   ├─ new DistCpContext(options)
  │   └─ execute()
  │
  ├─ DistCp.execute()
  │   └─ createAndSubmitJob()
  │
  ├─ DistCp.createJob()
  │   ├─ DistCpUtils.getStrategy() → UniformSizeInputFormat
  │   ├─ job.setMapperClass(CopyMapper.class)
  │   ├─ configureOutputFormat() // 工作目录/日志路径
  │   └─ context.appendToConf(config)
  │
  ├─ prepareFileListing(job)
  │   └─ 常规模式 → createInputFileListing(job)
  │       └─ CopyListing.getCopyListing() 工厂方法
  │           └─ GlobbedCopyListing → globStatus → 委托
  │               └─ SimpleCopyListing → 递归遍历目录树
  │                   → 写入 SequenceFile（relativePath, FileStatus）
  │
  ├─ job.submit() → MR 作业提交
  │
  └─ [MR 运行时]
      ├─ UniformSizeInputFormat.getSplits() 生成分片
      └─ CopyMapper.map() 执行拷贝
           ├─ checkUpdate() → canSkip() → decision
           ├─ RetriableFileCopyCommand.execute()
           │   ├─ ThrottledInputStream.throttle() 限速
           │   ├─ copyBytes() → read → write
           │   └─ promoteTmpToTarget() → rename
           └─ DistCpUtils.preserve() 保留属性
               └─ [作业完成]
                  └─ CopyCommitter.commitJob()
                     ├─ preserveFileAttributesForDirectories()
                     ├─ deleteMissing() / commitData() / trackMissing()
                     └─ cleanup()

设计要点

  1. 装饰器 + 模板方法：FileBasedCopyListing → GlobbedCopyListing → SimpleCopyListing，每层只做一件事，CopyListing.buildListing() 作为模板定义固定流程。
  2. Builder 模式：DistCpOptions 使用 Builder 构建，build() 时做参数校验，构建后不可变。
  3. DistCpContext 可变性：sourcePaths 和 targetPathExists 可在运行时改写，适应 Glob 展开和快照处理。
  4. 配置键不对齐：代码按 distcp.{strategy}.strategy.impl 查找，而 distcp-default.xml 中定义的是 distcp.static.strategy.impl，实际上未被使用，依赖代码中的回退默认值 UniformSizeInputFormat.class。
  5. 纯 Map 作业：job.setNumReduceTasks(0)，不设 Reduce 阶段，所有文件拷贝在 Mapper 中完成。
赏

感谢您的支持!

支付宝
微信

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

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

本文标题：Hadoop DistCp 源码分析：命令入口到核心调用链

本文链接：https://dongkelun.com/2026/08/01/hadoop/hadoop-distcp-sourceCode/

欢迎关注我的公众号

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~