Distcp方式

超、凢脫俗 2022-02-19 06:09 240阅读 0赞

转载：[https://blog.csdn.net/yinansheng1/article/details/78402459][https_blog.csdn.net_yinansheng1_article_details_78402459]

## 一、概述 ##

（分布式拷贝）是用于大规模集群内部和集群之间拷贝的工具。  
它使用Map/Reduce实现文件分发，错误处理和恢复，以及报告生成。  
它把文件和目录的列表作为map任务的输入，每个任务会完成源列表中部分文件的拷贝。  
由于使用了Map/Reduce方法，这个工具在语义和执行上都会有特殊的地方。

###    二、使用DistCp ###

DistCp最常用于在集群之间的拷贝：  
hadoop distcp hdfs://nn1:8020/source hdfs://nn2:8020/destination  
上述命令会把nn1集群的/source目录下的所有文件或目录展开并存储到一个临时文件中，这些文件内容的拷贝工作被分配给多个map任务，  
然后每个NodeManager分别执行从nn1到nn2的拷贝操作。注意：DistCp使用绝对路径进行操作。  
命令行中还可以指定多个源目录：  
hadoop distcp hdfs://nn1:8020/source/a hdfs://nn1:8020/source/b hdfs://nn2:8020/destination  
或者使用-f选项，从文件里获得多个源：  
hadoop distcp -f hdfs://nn1:8020/srclist hdfs://nn2:8020/destination  
其中srclist 的内容是：  
hdfs://nn1:8020/source/a  
hdfs://nn1:8020/source/b  
当从多个源拷贝时，如果两个源冲突，DistCp会停止拷贝并提示出错信息， 如果在目的位置发生冲突，会根据选项设置解决。  
默认情况会跳过已经存在的目标文件（比如不用源文件做替换操作）。每次操作结束时  
都会报告跳过的文件数目，但是如果某些拷贝操作失败了，但在之后的尝试成功了， 那么报告的信息可能不够精确。  
每个Datanode必须都能够与源宿端Datanode进行访问和交互。  
每个Datanode必须都能够与源宿端Datanode进行访问和交互。  
对于HDFS来说，源和目的端要运行相同版本的协议或者使用向下兼容的协议。详细请参阅“在版本之间复制”章节。拷贝完成后，建议生成  
源端和目的端文件的列表，并交叉检查，来确认拷贝是否真正成功。 因为DistCp使用Map/Reduce和FileSystem  
API进行操作，所以这三者或它们之间有任何问题  
都会影响拷贝操作。一些Distcp命令的成功执行可以通过再次执行带-update参数的该命令来完成，  
但用户在如此操作之前应该对该命令的语法很熟悉。  
值得注意的是，当另一个客户端同时在向源文件写入时，拷贝很有可能会失败。 尝试覆盖HDFS上正在被写入的文件的操作也会失败。  
如果一个源文件在拷贝之前被移动或删除了，拷贝失败同时输出异常 FileNotFoundException。

###    三、命令行选项 ###

DistCp命令行选项如表1所示：

<table> 
 <tbody> 
 <tr> 
 <td>标识</td> 
 <td>描述</td> 
 <td>备注</td> 
 </tr> 
 <tr> 
 <td>-p[rbugp]</td> 
 <td>Preserve r:replication number b: block size u: user g: group p: permission</td> 
 </tr> 
 <tr> 
 <td>-i</td> 
 <td>忽略失败</td> 
 <td>这个选项会比默认情况提供关于拷贝的更精确的统计，</td> 
 </tr> 
 <tr> 
 <td>-log &lt;logdir&gt;</td> 
 <td>记录日志到 &lt;logdir&gt;</td> 
 </tr> 
 <tr> 
 <td>-m &lt;num_maps&gt;</td> 
 <td>同时拷贝的最大数目</td> 
 <td>指定了拷贝数据时map的数目。请注意并不是map数越多吞吐量越大。</td> 
 </tr> 
 <tr> 
 <td>-overwrite</td> 
 <td>覆盖目标</td> 
 <td>用户要小心使用这个选项。</td> 
 </tr> 
 <tr> 
 <td>-update</td> 
 <td>如果源和目标的大小不一样则进行覆盖</td> 
 <td>用户使用要小心。</td> 
 </tr> 
 <tr> 
 <td>-f &lt;urilist_uri&gt;</td> 
 <td>用&lt;urilist_uri&gt; 作为源文件列表</td> 
 <td>这等价于把所有文件名列在命令行中。 urilist_uri 列表应该是完整合法的URI。</td> 
 </tr> 
 </tbody> 
</table>

修改次数不会被保留。并且当指定 -update 时，更新的状态不会 被同步，除非文件大小不同（比如文  
同时它还将保留失败拷贝操作的日志，这些日志信息可以用于调试。最后，如果一个map失败了，但并  
DistCp为每个文件的每次尝试拷贝操作都记录日志，并把日志作为map的输出。 如果一个map失败了  
如果一个map失败并且没有使用-i选项，不仅仅那些拷贝失败的文件，这个分块任务中的所有文件都会  
这不是"同步"操作。 执行覆盖的唯一标准是源文件和目标文件大小是否相同；如果不同，则源文件替  
表1 DistCp命令行参数表

###    四、更新与覆盖 ###

DistCp –update选项用于复制目标中不存在的文件或者文件内容不相同的文件。DistCp  
\-overwrite选项将覆盖目标文件，即使它们存在于源中，或者它们具有相同的内容。-update和-overwrite选项值得进一步讨论，因为它们处  
理源路径的方式与默认值不同，有些细节需要注意。  
这里给出一些 -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  
当不使用-update或-overwrite选项时，DistCp默认会在/target下创建/first和/second目录。因此将在/target之前先创建目录。  
从而：  
hadoop 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”被复制，因为它不存在/target中。因为目标文件内容与源文件内容不匹配，文件“10”和文件“20”被覆盖。如果使用-update  
选项，文件“1”也被覆盖。

###    五、DistCp命令的安全设置 ###

安全设置揭示了DistCp命令是运行在源集群上还是运行在目标集群上。一般认为，如果一个集群是安全的，另一个集群是不安全的，则Dist 
Cp应该从安全集群运行，否则可能存在与安全相关的问题。 
将数据从安全集群复制到非安全集群时，DistCp客户端需要设置如下配置： 
<property> 
<name>ipc.client.fallback-to-simple-auth-allowed</name> 
<value>true</value> 
</property> 
将数据从安全集群复制到安全集群时，core-site.xml文件中需要设置如下配置： 
<property> 
<name>hadoop.security.auth\_to\_local</name> 
<value></value> 
<description>Maps kerberos principals to local user names</description> 
</property>

###    六、安全保护：Kerberos Principal Name ###

distcp hdfs://cluster1-secure hdfs://cluuter2-secure  
考虑上述命令，这里存在一个问题：SASL RPC客户端要求远程服务器的Kerberos Principal  
Name必须与自己的配置中的服务器主体匹配。 因此，必须将相同的主体名称分配给源和目标集群中相关的NameNodes。  
例如，如果源集群中的NameNode的Kerberos Principal为nn/host1@realm，则目标集群中NameNode的Kerberos  
Principal也必须为nn/host2@realm，而不能为nn2/host2@realm。

###    七、安全保护：ResourceManager映射规则 ###

当在两个安全的BCH集群间复制文件，如果两个集群存在不同的区域，那么需要进一步配置ResourceManager（RM）。为了使DistCP命 
令成功，必须在两个集群中使用相同的RM映射规则。 
例如，如果安全集群1具有以下RM映射规则： 
<property> 
<name>hadoop.security.auth\_to\_local</name> 
<value> 
RULE:2:$1@$0s/.*/yarn/ 
DEFAULT 
</value> 
</property>* 
并且安全集群2具有以下RM映射规则： 
<property> 
<name>hadoop.security.auth\_to\_local</name> 
<value> 
RULE:2:$1@$0s/./yarn/ 
DEFAULT 
</value> 
</property> 
从集群1到集群2的DistCp作业将会失败，因为集群2中的RM映射规则与群集1中的RM映射规则不同，导致集群2无法正确解析集群1 
中的RM映射规则供yarn使用。 
解决方案是：在集群1和集群2中使用相同的RM映射规则。 
<property> 
<name>hadoop.security.auth\_to\_local</name> 
<value> 
RULE:2:$1@$0s/.*/yarn/ 
RULE:2:$1@$0s/.*/yarn/ 
DEFAULT 
</value> 
</property>

###    八 、HADistCp ###

在HA集群之间复制数据，需要在hdfs-site.xml文件中设置dfs.internal.nameservices属性来显式指定属于本地群集的名称服务，同时设置d 
fs.nameservices属性来指定在本地和远程集群中的所有名称服务。 
使用以下步骤在HA群集之间复制数据： 
（1）设置dfs.nameservices = HAA, HAB。 
（2）添加dfs.internal.nameservices属性： 
集群A： 
dfs.internal.nameservices = HAA 
集群B： 
dfs.internal.nameservices = HAB 
（3）将dfs.ha.namenodes. <nameservice>添加到两个集群中： 
集群A： 
dfs.ha.namenodes.HAB = nn1,nn2 
集群B： 
dfs.ha.namenodes.HAA = nn1,nn2 
（4）添加dfs.namenode.rpc-address. <cluster>.<nn>属性： 
集群A： 
dfs.namenode.rpc-address.HAB.nn1 = <NN1\_fqdn>:8020 
dfs.namenode.rpc-address.HAB.nn2 = <NN2\_fqdn>:8020 
集群B： 
dfs.namenode.rpc-address.HAA.nn1 = <NN1\_fqdn>:8020 
dfs.namenode.rpc-address.HAA.nn2 = <NN2\_fqdn>:8020 
（5）添加dfs.client.failover.proxy.provider. <cluster>属性： 
集群A： 
dfs.client.failover.proxy.provider.HAB= 
org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider 
集群B： 
dfs.client.failover.proxy.provider.HAA= 
org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider 
（6）重新启动HDFS服务，然后使用NameService运行distcp命令。 例如： 
hadoop distcp hdfs://falconG/tmp/testDistcp hdfs://falconE/tmp/

###    九、DistCp和Hadoop版本 ###

源和目标集群的Hadoop版本决定应使用哪种类型的文件系统来从源集群读取并写入目标集群。BCH为Hadoop2的版本，可能存在这样的场  
景，已有的集群（可能是Hadoop1版本）需要向BCH集群拷贝数据，由于Hadoop  
1.x和2.x具有不同的RPC版本，客户端无法同时理解，因此对源和目的都不可能使用“hdfs”。在这种情况下，WebHdfsFilesystem（webh  
dfs://）可以在源和目标集群中使用，或者HftpFilesystem（hftp://）可用于从源集群读取数据。

###    十、DistCp数据复制矩阵：Hadoop1 / Hadoop2到Hadoop2 ###

表2提供了使用DistCp将数据从HDP1或HDP2群集复制到HDP2群集时的配置，设置和结果的相关信息。

<table> 
 <tbody> 
 <tr> 
 <td>源</td> 
 <td>目的</td> 
 <td>源配置</td> 
 <td>目的配置</td> 
 <td>DistCp应当运行环境</td> 
 <td>结果</td> 
 </tr> 
 <tr> 
 <td>Hadoop1</td> 
 <td>Hadoop2</td> 
 <td>insecure + hdfs</td> 
 <td>insecure + webhdfs</td> 
 <td>Hadoop1 (source)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop1</td> 
 <td>Hadoop2</td> 
 <td>secure + hdfs</td> 
 <td>secure + webhdfs</td> 
 <td>Hadoop1 (source)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop1</td> 
 <td>Hadoop2</td> 
 <td>secure + hftp</td> 
 <td>secure + hdfs</td> 
 <td>Hadoop2 (destination)</td> 
 <td>success</td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td>Hadoop1</td> 
 <td>Hadoop2</td> 
 <td>secure + hftp</td> 
 <td>secure + swebhdfs</td> 
 <td>Hadoop2(destination)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop1</td> 
 <td>Hadoop2</td> 
 <td>secure + hdfs</td> 
 <td>insecure + webhdfs</td> 
 <td>Hadoop1 (source)</td> 
 <td>Possible issues are discussed&nbsp;here.</td> 
 </tr> 
 <tr> 
 <td>Hadoop2</td> 
 <td>Hadoop2</td> 
 <td>secure + hdfs</td> 
 <td>insecure + hdfs</td> 
 <td>secure Hadoop2 (source)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop2</td> 
 <td>Hadoop2</td> 
 <td>secure + hdfs</td> 
 <td>secure + hdfs</td> 
 <td>either Hadoop2 (source or destination)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop2</td> 
 <td>Hadoop2</td> 
 <td>secure + hdfs</td> 
 <td>secure + webhdfs</td> 
 <td>Hadoop2 (source)</td> 
 <td>success</td> 
 </tr> 
 <tr> 
 <td>Hadoop2</td> 
 <td>Hadoop2</td> 
 <td>secure + hftp</td> 
 <td>secure + hdfs</td> 
 <td>Hadoop2 (destination)</td> 
 <td>success</td> 
 </tr> 
 </tbody> 
</table>

表2 使用DistCp将数据从Hadoop1或Hadoop2群集复制到Hadoop2群集时的相关信息  
对于上表：  
（1）术语“安全”意味着设置了Kerberos安全性。  
（2）hsftp可用于Hadoop1和Hadoop2。 它向hftp添加了https支持。

###    十一、 从Hadoop2复制数据到Hadoop1群集 ###

DistCp支持从Hadoop1集群复制数据到Hadoop2群集，但Hadoop1集群无法意识Hadoop2集群中引入的Checksum机制。使用DistCp从  
Hadoop2复制数据到Hadoop1需满足如下条件：跳过checksum和或者确保要复制的文件位于CRC32中。

###    十二、 DistCp架构 ###

DistCp由以下组件组成： Distcp驱动程序、 复制列表生成器、 InputFormats和MapReduce组件。

####    12.1、Distcp驱动程序 ####

DistCp驱动程序组件负责解析命令行中传递给DistCp命令的参数。它通过OptionsParser和DistCpOptionsSwitch来完成上述功能。将命令  
参数组成适当的DistCpOptions对象，并初始化DistCp。 这些参数包括：来源路径、目标位置、复制选项（例如是否更新 -  
复制，覆盖，要保存的文件属性等）。  
通过以下方式协调复制操作：  
（1）调用复制列表生成器来创建要复制的文件列表。  
（2）设置和启动Hadoop MapReduce作业以执行复制操作。  
（3）根据选项，要么立即将句柄返回给Hadoop MapReduce作业，要么等到完成。  
解析器元素只能在命令行执行（或者调用了DistCp ::  
run（）方法）。通过构造DistCpOptions对象并适当地初始化DistCp对象，DistCp类也可以以编程方式使用。

####    12.2、复制列表生成器 ####

复制列表生成器类负责创建要从源复制的文件/目录的列表。  
他们检查源路径（文件/目录，包括通配符）的内容，并将需要复制的所有路径记录到SequenceFile中以供DistCp Hadoop作业使用。  
该模块的主要类包括：  
（1）  
CopyListing：任何复制列表生成器都需要实现CopyListing接口。该接口还提供了工厂方法给实现了该接口的复制列表生成器类选择使用。  
（2）SimpleCopyListing：该类实现了CopyListing接口，可以接受多个源路径（文件/目录），并递归列出每个副本的所有单个文件和目  
录。  
（3）GlobbedCopyListing：增加了在源路径中扩展通配符功能的CopyListing的另一个实现。  
（4）FileBasedCopyListing：从指定文件读取源路径列表的CopyListing的实现。  
根据DistCpOptions中是否指定了源文件列表，源列表以下列方式中的某一种方式生成：  
（1）如果没有源文件列表，则使用GlobbedCopyListing。  
所有通配符都被扩展，所有的扩展都转发到SimpleCopyListing，而SimpleCopyListing反过来构造列表（通过每个路径的递归下降）。  
（2）如果指定了源文件列表，则使用FileBasedCopyListing。 源路径从指定的文件读取，然后转发到GlobbedCopyListing。  
然后如上所述构建列表。  
您也可以通过提供自定义的CopyListing接口的实现来自定义构建复制列表的方法。 在如何考虑复制路径时，DistCp的行为不同于传统的Dis  
tCp。传统实现仅列出必须将其复制到目标上的路径。  
例如，如果文件已存在于目标（并且未指定-overwrite），那么在MapReduce复制作业中甚至不考虑该文件。  
在设置期间（即在MapReduce作业之前）确定这一点涉及文件大小和校验和比较，这存在潜在的耗时。DistCp推迟这样的检查，直到MapR  
educe工作，从而减少设置时间。 通过在多个Maps中并行检查，性能会得到进一步增强。

####    12.3、InputFormats和MapReduce组件 ####

InputFormats和MapReduce组件负责将文件和目录从源复制到目标路径。  
在复制列表生成期间创建的列表文件在执行复制时会被消耗。 这里相关类包括：  
UniformSizeInputFormat：org.apache.hadoop.mapreduce.InputFormat的实现类，该类提供了与Legacy  
DistCp的等价性，以平衡多个Map之间的负载。UniformSizeInputFormat的目的是使每个Map复制大致相同的字节数。因此，列表文件被  
分割成路径组，使得每个InputSplit中的文件大小的总和几乎相等。 分裂并不总是完美的，但这的确降低了设置时间。  
DynamicInputFormat和DynamicRecordReader：DynamicInputFormat实现了org.apache.hadoop.mapreduce.InputFormat接口，是  
DistCp的新功能。 列表文件被分成几个“块文件”，块文件的确切数量是Hadoop作业中请求的Map数量的倍数。  
在启动作业之前，每个Map任务都将“分配”其中一个块文件（通过将块重命名为任务的id）。  
使用DynamicRecordReader从每个块读取路径，并在CopyMapper中处理。  
在处理了一个组块中的所有路径之后，当前组块被删除并尝试获取一个新的组块。 该过程继续，直到没有更多的块可用。  
这种“动态”方法允许更快的Map任务，这比慢的Map消耗更多路径，从而加快了DistCp作业的整体运行。  
CopyMapper：此类用于实现物理文件复制。 根据输入选项（在作业配置中指定）检查输入路径，以确定是否需要复制文件。  
当且仅当以下至少一个为真时才会复制该文件：  
（1） 目标不存在同名的文件。  
（2） 具有相同名称的文件存在于目标，但具有不同的文件大小。  
（3）目标上存在同名的文件，但是具有不同的校验和，并且没有提到skipcrccheck。  
（4） 目标中存在同名的文件，但指定了-overwrite。  
（5）具有相同名称的文件存在于目标位置，但块大小不同（需要保留块大小）。  
CopyCommitter：该类负责DistCp作业的提交阶段，包括：保存目录权限（如果在选项中指定）、清理临时文件，工作目录等。

[https_blog.csdn.net_yinansheng1_article_details_78402459]: https://blog.csdn.net/yinansheng1/article/details/78402459