DistCp 指南

基本用法

DistCp最常见的调用方式是跨集群复制：

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

这将在nn1上将/foo/bar命名空间扩展为一个临时文件，将其内容分配到一组map任务中，并在每个NodeManager上启动从nn1到nn2的复制过程。

也可以在命令行中指定多个源目录：

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

或者，等效地，使用 -f 选项从文件中读取：

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

其中 srclist 包含

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

当从多个源复制时，如果两个源发生冲突，DistCp会中止复制并显示错误消息，但目标位置的冲突会根据指定的选项进行解决。默认情况下，目标位置已存在的文件会被跳过（即不会被源文件替换）。每次作业结束时都会报告跳过的文件数量，但如果某个复制器对部分文件失败但在后续尝试中成功，该计数可能不准确。

重要的是，每个NodeManager都能访问并与源和目标文件系统进行通信。对于HDFS，源和目标必须运行相同版本的协议或使用向后兼容的协议；请参阅[版本间复制](#Copying_Between_Versions_of_HDFS)。

复制完成后，建议生成并交叉核对源文件和目标文件的列表，以验证复制是否真正成功。由于DistCp同时使用了Map/Reduce和文件系统API，三者中任何一个或其之间的问题都可能对复制产生不利影响且不易察觉。有些人通过启用-update参数进行第二次传递取得了成功，但用户在尝试此操作前应熟悉其语义。

还需要注意的是，如果另一个客户端仍在向源文件写入数据，复制操作很可能会失败。在HDFS上尝试覆盖正在被写入的目标文件同样会失败。如果在复制之前源文件被（重新）移动，复制操作将因FileNotFoundException而失败。

请参阅详细的命令行参考文档，了解DistCp中所有可用选项的信息。

更新与覆盖

-update 用于从源位置复制目标位置不存在或版本不同的文件。-overwrite 会覆盖目标位置已存在的文件。

更新和覆盖选项需要特别注意，因为它们对源路径的处理方式与默认方式存在非常微妙的差异。假设从/source/first/和/source/second/复制到/target/，其中源路径包含以下内容：

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

当调用DistCp时没有使用-update或-overwrite参数，DistCp默认会在/target目录下创建first/和second/文件夹。因此：

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

将在 /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

当指定-update或-overwrite时，源目录的内容会被复制到目标位置，而不是源目录本身。因此：

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

将在 /target 中生成以下内容：

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

进一步来说，如果两个源文件夹都包含一个同名文件（例如0），那么两个源都会将条目映射到目标位置的/target/0。为了避免这种冲突，DistCp会中止操作。

现在，考虑以下复制操作：

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

包含来源/大小：

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

以及目标/大小：

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

结果将是：

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 被跳过，因为文件长度和内容匹配。2 被复制，因为目标位置不存在该文件。10 和 20 被覆盖，因为它们的内容与源文件不匹配。

如果使用-update选项，1会被跳过，因为文件长度和内容匹配。2会被复制，因为目标位置不存在该文件。10和20会被覆盖，因为它们的内容与源文件不匹配。但如果同时使用-append选项，则只有10会被覆盖（源文件长度小于目标文件），而20会在文件变更处追加内容（如果文件在目标原始长度之前的内容是匹配的）。

如果使用了-overwrite，1也会被覆盖。

同步

-diff 选项通过快照差异将文件从源集群同步到目标集群。它会复制、重命名和删除快照差异列表中的文件。

使用-diff选项时必须包含-update选项。

目前大多数云服务提供商在同步功能方面表现不佳。

用法：

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

示例：

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

上述命令将/src/目录中从快照snap1到snap2的变化（即从snap1到snap2的快照差异）应用到/dst/目录。显然，这要求/src/目录必须同时包含snap1和snap2两个快照。但目标目录/dst/也必须拥有一个与同名的快照，在本例中即snap1。目标目录/dst/自snap1以来不应有新的文件操作（创建、重命名、删除）。请注意，当此命令执行完成时，不会在/dst/目录创建新的快照snap2。

-update 是使用 -diff 选项的必要条件。

例如，在/src/目录中，如果在创建snap1之后、创建snap2之前添加了1.txt并删除了2.txt，上述命令会将1.txt从/src/复制到/dst/，并从/dst/中删除2.txt。

同步行为将通过以下实验进行详细说明。

# 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/

原始命名空间扩展属性保留

本节仅适用于HDFS。

如果目标和所有源路径名都在/.reserved/raw层级结构中，那么'raw'命名空间的扩展属性将被保留。系统使用'raw'扩展属性来实现内部功能，如加密元数据。只有当用户通过/.reserved/raw层级结构访问时，这些属性才对用户可见。

原始扩展属性(raw xattrs)的保留仅取决于是否提供了/.reserved/raw前缀。-p标志(保留标志，如下所述)不会影响原始扩展属性的保留。

为了防止保留原始扩展属性，只需避免在任何源路径和目标路径上使用/.reserved/raw前缀。

如果仅在源路径和目标路径的子集上指定了/.reserved/raw前缀，将显示错误并返回非0退出码。

DistCp 驱动

DistCp驱动组件负责以下功能：

• 解析通过命令行传递给DistCp命令的参数，方式如下：

  • OptionsParser，以及
  • DistCpOptionsSwitch

• 将命令参数组装成适当的DistCpOptions对象，并初始化DistCp。这些参数包括：

  • 源路径
  • 目标位置
  • 复制选项（例如是否更新复制、覆盖、保留哪些文件属性等）

• 通过以下方式编排复制操作：

  • 调用copy-listing-generator来生成待复制文件列表。
  • 设置并启动Hadoop Map-Reduce作业来执行复制操作。
  • 根据选项，要么立即返回Hadoop MR Job的句柄，要么等待直至完成。

解析器元素仅通过命令行（或调用DistCp::run()时）执行。DistCp类也可以通过编程方式使用，方法是构建DistCpOptions对象并适当地初始化DistCp对象。

复制列表生成器

copy-listing-generator类负责创建需要从源位置复制的文件/目录列表。它们会检查源路径（文件/目录，包括通配符）的内容，并将所有需要复制的路径记录到SequenceFile中，供DistCp Hadoop作业使用。该模块中的主要类包括：

1. CopyListing: 任何拷贝列表生成器实现都应遵循的接口。同时提供了用于选择具体CopyListing实现的工厂方法。
2. SimpleCopyListing: CopyListing的一个实现，它接受多个源路径（文件/目录），并递归列出每个路径下的所有单独文件和目录，用于复制。
3. GlobbedCopyListing: CopyListing的另一种实现，可以扩展源路径中的通配符。
4. FileBasedCopyListing: CopyListing的一个实现，从指定文件中读取源路径列表。

根据是否在DistCpOptions中指定了源文件列表，源列表将通过以下方式之一生成：

1. 如果没有源文件列表，则使用GlobbedCopyListing。所有通配符都会被展开，展开结果会转发给SimpleCopyListing，后者通过递归遍历每个路径来构建文件列表。
2. 如果指定了源文件列表(source-file-list)，则使用FileBasedCopyListing。源路径会从指定文件中读取，然后转发给GlobbedCopyListing。随后将按照上述方式构建文件列表。

用户可以通过提供CopyListing接口的自定义实现，来定制构建复制列表的方法。DistCp在此处的行为与传统DistCp不同，主要体现在处理复制路径的方式上。

用户还可以通过传递当前支持的CopyFilter接口实现或编写新的实现，来自定义不应复制的文件的过滤规则。这可以通过在DistCpOptions中设置distcp.filters.class来指定：

1. distcp.filters.class 设置为 "RegexCopyFilter"。如果使用此实现，您需要传递包含用于过滤的正则表达式的 "CopyFilter" distcp.filters.file。支持由java.util.regex.Pattern指定的正则表达式。
2. distcp.filters.class 设置为 "RegexpInConfigurationFilter"。如果使用此实现，还需通过 "DistCpOptions" 中的 distcp.exclude-file-regex 参数传递正则表达式。支持 java.util.regex.Pattern 指定的正则表达式。与 "RegexCopyFilter" 相比，这是一种更动态的方法。
3. distcp.filters.class 设置为 "TrueCopyFilter"。如果未指定上述任何选项，则将其用作默认实现。

旧版实现仅列出那些必须复制到目标路径的文件。例如，如果目标位置已存在某个文件（且未指定-overwrite参数），该文件甚至不会被纳入MapReduce复制作业的考虑范围。在设置阶段（即MapReduce作业开始前）确定这一点需要进行文件大小和校验和比较，这些操作可能非常耗时。

新版DistCp将这些检查推迟到MapReduce作业阶段，从而减少了设置时间。由于这些检查在多个映射任务中并行执行，性能得到进一步提升。

InputFormats 和 MapReduce 组件

InputFormats和MapReduce组件负责将文件和目录从源路径实际复制到目标路径。在复制执行时，会使用在复制列表生成阶段创建的列表文件。这里涉及的主要类包括：

• UniformSizeInputFormat: 该org.apache.hadoop.mapreduce.InputFormat的实现提供了与传统DistCp在平衡映射负载方面的等效性。UniformSizeInputFormat的目标是让每个映射任务复制大致相同数量的字节。相应地，列表文件被分割成路径组，使得每个InputSplit中的文件大小总和几乎与其他映射任务相同。这种分割并非总是完美，但其简单的实现方式保持了较低的设置时间。

• DynamicInputFormat 和 DynamicRecordReader： DynamicInputFormat 实现了 org.apache.hadoop.mapreduce.InputFormat，这是 DistCp 新增的功能。清单文件会被分割成多个"分块文件"，分块文件的确切数量是 Hadoop 作业中请求的 map 任务数量的整数倍。在作业启动前，每个 map 任务会被"分配"一个分块文件（通过将分块重命名为任务 ID）。使用 DynamicRecordReader 从每个分块中读取路径，并在 CopyMapper 中进行处理。当一个分块中的所有路径都处理完成后，当前分块会被删除并获取新的分块。这个过程会持续进行，直到没有更多分块可用。这种"动态"方法允许较快的 map 任务比较慢的任务处理更多路径，从而整体上加速 DistCp 作业。

• CopyMapper: 该类实现物理文件复制功能。系统会根据输入路径与作业配置中指定的输入选项进行比对，以判断文件是否需要复制。满足以下任一条件时，文件才会被复制：

  • 目标位置不存在同名文件。
  • 目标位置存在同名文件，但文件大小不同。
  • 目标位置已存在同名文件，但其校验和不同，且未提及-skipcrccheck参数。
  • 目标位置已存在同名文件，但指定了-overwrite参数。
  • 目标位置已存在同名文件，但其块大小不同且需要保留块大小。

• CopyCommitter: 该类负责DistCp作业的提交阶段，包括：

  • 保留目录权限（如果在选项中指定）
  • 清理临时文件、工作目录等。

Map大小调整

默认情况下，DistCp会尝试使每个map任务处理的数据量相对均衡，确保每个任务复制大致相同的字节数。请注意，文件是最小的粒度单位，因此增加并发复制器（即map任务）的数量并不总能提升并发复制数或整体吞吐量。

新版DistCp还提供了一种“动态”调整map大小的策略，允许速度较快的数据节点比慢速节点复制更多字节。通过使用-strategy dynamic（架构部分已说明），不再为每个map任务分配固定的源文件集，而是将文件分割成多个集合。这些集合的数量通常比map数量多2-3倍。每个map会选取并复制一个块中列出的所有文件。当一个块处理完成后，将获取并处理新块，直到没有剩余块为止。

通过不将源路径分配给固定的映射，速度较快的映射任务（即数据节点）能够比速度较慢的节点消耗更多数据块，从而复制更多数据。虽然这种分布并不均匀，但就每个映射器的容量而言是公平的。

动态策略由DynamicInputFormat实现。在大多数情况下它能提供卓越的性能。

对于长时间运行和定期运行的作业，建议根据源集群和目标集群的大小、复制数据量以及可用带宽来调整映射任务的数量。

在不同版本的HDFS之间复制数据

要在两个不同主版本的Hadoop之间进行复制（例如1.X和2.X之间），通常可以使用WebHdfsFileSystem。与之前的HftpFileSystem不同，由于webhdfs同时支持读写操作，DistCp可以在源集群和目标集群上运行。远程集群应指定为webhdfs://:。当在相同主版本的Hadoop集群之间复制时（例如2.X和2.X之间），使用hdfs协议可获得更好的性能。

使用distcp进行安全有线传输

当webhdfs启用SSL安全协议时，请使用"swebhdfs://"方案。更多信息请参阅SSL Configurations for SWebHDFS。

MapReduce及其他副作用

如前所述，如果某个map未能复制其输入之一，将会产生若干副作用。

• 除非指定了-overwrite参数，否则在重新执行时，之前已由某个map成功复制的文件将被标记为"已跳过"。
• 如果一个map任务失败mapreduce.map.maxattempts次，剩余的所有map任务将被终止（除非设置了-i参数）。
• 如果mapreduce.map.speculative设置为true，则复制结果未定义。

DistCp与对象存储

DistCp 可与对象存储（如 Amazon S3、Azure ABFS 和 Google GCS）协同工作。

前提条件

1. 包含对象存储实现的JAR文件及其所有依赖项都在类路径上。
2. 除非JAR自动注册其捆绑的文件系统客户端，否则可能需要修改配置以声明实现文件系统模式的类。ASF所有的自有对象存储客户端都是自注册的。
3. 相关的对象存储访问凭证必须在集群配置中可用，或者以其他方式在所有集群主机中可用。

DistCp可用于上传数据

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

下载数据

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

在对象存储之间复制数据

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

并在对象存储中复制数据

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

并使用-update仅复制更改过的文件。

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

由于对象存储列出文件速度较慢，在对大型目录树执行-update操作时，建议设置-numListstatusThreads选项（线程数上限为40）。

当在对象存储中使用DistCp -update时，通常仅比较单个文件的修改时间和长度，如果两个存储之间的校验和算法不同，则不会比较任何校验和。

• 在不同校验和算法的两个对象存储之间执行distcp -update时，会通过比较源文件和目标文件的修改时间以及文件大小来决定是否跳过文件复制。该行为由属性distcp.update.modification.time控制，默认值为true。如果源文件的修改时间比目标文件新，则假定内容已更改，应更新该文件。我们需要确保机器之间没有时钟偏差。大多数对象存储确实为目录提供了有效的时间戳，但这并不重要；仅比较文件时间戳。然而，确保客户端计算机的时钟与基础设施的时钟接近非常重要，这样客户端/HDFS集群与对象存储之间的时间戳才能保持一致。否则，可能会遗漏更改的文件或过于频繁地复制文件。

• distcp.update.modification.time 仅在以下情况下使用：两个存储中任意一个没有校验和验证，导致两者之间无法进行校验和比较。即使该属性设置为true，如果两个存储之间存在有效的校验和比较，该属性也不会被使用。

要关闭修改时间检查，请在您的core-site.xml中设置此项

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

注意事项

• -atomic选项会导致临时数据的重命名，因此会显著增加操作结束时提交工作的时间。此外，由于除（可选）wasb://之外的对象存储不提供目录的原子重命名，-atomic操作实际上无法兑现其承诺。应避免使用。

• 不支持 -append 选项。

• 不支持 -diff 和 rdiff 选项

• 无论-skipCrc标志的值如何，都不会执行CRC检查。

• 所有-p选项，包括保留权限、用户和组信息、属性校验和以及复制的选项通常都会被忽略。wasb://连接器会保留这些信息，但不会强制执行权限。

• 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.

• 在同一个对象存储内的复制操作仍然会在Hadoop集群中执行——即使该对象存储在内部实现了更高效的COPY操作

  也就是说，像这样的操作

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

  会将每个字节都传输到Hadoop工作节点，然后再传回存储桶。这不仅速度慢，还可能导致费用产生。

• 使用-direct选项可以直接写入对象存储目标路径，避免可能非常耗时的临时文件重命名操作。