Hadoop集群间文件拷贝
distcp使用
DistCp Version 2(分布式copy)是用于集群间/集群内的文件copy工具， 使用MapReduce实现分布式、错误处理、恢复和报告。distCp会根据目录文件生成map任务， 每一个任务会copy部分文件内容。
基本使用
最常使用的是集群间copy hadoop distcp hdfs://nn1:8020/foo/bar hdfs://nn2:8020/bar/foo
该命令会将nn1下的/foo/bar放在一个临时文件里，并将内容分割为一系列map tasts，
然后在每一个NodeManager上从nn1向nn2拷贝。 也可在命令行指定多个源文件 hadoop distcp hdfs://nn1:8020/foo/a
hdfs://nn1:8020/foo/b
hdfs://nn2:8020/bar/foo
同样的，使用-f指定文件，srclist包含a和b hadoop distcp -f hdfs://nn1:8020/srclist
hdfs://nn2:8020/bar/foo
在多源copy中，如果源文件冲突，distCp将会终止并发送错误报告，而在目标上的冲突将会根据指定的选项（option）处理。默认情况下，跳过目标集群已经存在的文件而不是替换，跳过文件的数量将会在任务结束的时候报告。不过，如果存在copy任务失败，后续重试成功的情况，任务最终报告的跳过文件的数量将会不准确。
注意：每个NodeManager都必须与源文件系统、目标文件系统正常通信。对HDFS来说，源集群和目标集群必须是同一个协议版本（并非组件版本强一致），或使用向下兼容协议
Copy完成之后，最好进行产出、交叉验证，保证集群间copy结果无误。
如果源集群存在客户端写操作，copy任务很可能会失败，尝试覆盖一个在目标集群上正在写的文件也会失败，如果在copy任务开始之前，源文件被删除，copy任务将会失败并抛出FileNotFoundException
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
当distCp没有使用-update or -overwrite, distCp 默认将会在/target下创建 first/ 和 second/，例如: sh 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`, 会将源目录下的内容copy到目标下, 而不是源目录，例如:
```sh
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将会失败。 考虑如下copy操作 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 由于不存在，将会直接copy；10 和 20 由于内容不一致，将会覆盖；如果使用 -update, 1 也会被覆盖。
DistCp体系结构
DistCp主要有以下三个组件组成:
DistCp DriverCopy-listing generatorInput-formats 和 Map-Reduce components
DistCp Driver
// 构造方法，根据输入的参数创建DistCp
public DistCp(Configuration configuration, DistCpOptions inputOptions) throws Exception {
Configuration config = new Configuration(configuration);
config.addResource(DISTCP_DEFAULT_XML);
setConf(config);
this.inputOptions = inputOptions;
this.metaFolder = createMetaFolderPath();
}
/**
* 实现 Tool::run(). 组织源文件拷贝到目标位置
* 1. 创建将要拷贝到目标的文件列表
* 2. 运行Map任务. (委托给 execute().)
* @param argv List of arguments passed to DistCp, from the ToolRunner.
* @return On success, it returns 0. Else, -1.
*/
public int run(String[] argv) {
if (argv.length < 1) {
OptionsParser.usage();
return DistCpConstants.INVALID_ARGUMENT;
}
try {
inputOptions = (OptionsParser.parse(argv));
setTargetPathExists();
LOG.info("Input Options: " + inputOptions);
} catch (Throwable e) {
LOG.error("Invalid arguments: ", e);
System.err.println("Invalid arguments: " + e.getMessage());
OptionsParser.usage();
return DistCpConstants.INVALID_ARGUMENT;
}
......
}
DistCp Driver 组件负责以下内容:
将命令行命令参数解析传给DistCp, 通过:
OptionsParser： 通过 public static DistCpOptions parse(String args[]); 方法解析命令行参数，并创建相关的Options对象；DistCpOptionsSwitch： public enum DistCpOptionSwitch, 一个枚举类，与命令行参数的key（-i, -p）相映射。 将命令参数封装在合适的 DistCpOptions 对象中，初始化 DistCp，参数包括：
源路径目标位置复制选项 (例如：是否采用 update、overwrite 复制，保留文件那些属性等。) 通过以下过程组织copy
调用copy-listing-generator 创建要copy的文件列表；构件并执行 Map-Reduce 来执行copy；Based on the options, either returning a handle to the Hadoop MR Job immediately, or waiting till completion.基于用户指定的选项，立即返回MR任务操作或等待任务完成。 Copy-Listing 生成器 The copy-listing-generator类负责创建源集群要复制的文件/文件夹列表。检验源路径中的内容，记录所有需要copy到SequenceFile中的paths，供DistCp消费。其主要包括以下模块：
CopyListing: 上层接口，具体的copy-listing-generator通过实现此接口来实现列表生成功能。提供工厂方法供构造CopyListing的实现来选择； [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-emW801oA-1652686190230)(https://i.imgur.com/EjuUxrL.png)]
SimpleCopyListing: CopyListing的一个实现，接收多源路径，并且递归地列出每个path下的所有单独文件和目录，用于copy；GlobbedCopyListing: CopyListing的一个实现，支持源路径存在通配符；FileBasedCopyListing: CopyListing的一个实现，从一个指定文件中读取源路径列表； 根据是否在DistCpOptions中指定源文件列表，source-list通过以下方式生成：
如果没有指定source-file-list, 使用GlobbedCopyListing。 扩展所有的通配符，所有扩展发送给SimpleCopyListing，依次构造listing(通过向下递归方式)；如果指定source-file-list, 使用FileBasedCopyListing。从指定的文件中读取Source-paths, 然后发送给GlobbedCopyListing，像上面一样构造listing。 GlobbedCopyListing构造listing如下：
@Override
public void doBuildListing(Path pathToListingFile,
DistCpOptions options) throws IOException {
List
if (options.getSourcePaths().isEmpty()) {
throw new InvalidInputException("Nothing to process. Source paths::EMPTY");
}
for (Path p : options.getSourcePaths()) {
FileSystem fs = p.getFileSystem(getConf());
FileStatus[] inputs = fs.globStatus(p);
if(inputs != null && inputs.length > 0) {
for (FileStatus onePath: inputs) {
globbedPaths.add(onePath.getPath());
}
} else {
throw new InvalidInputException(p + " doesn't exist");
}
}
DistCpOptions optionsGlobbed = new DistCpOptions(options);
optionsGlobbed.setSourcePaths(globbedPaths);
simpleListing.buildListing(pathToListingFile, optionsGlobbed);
}
InputFormats 和 MapReduce 组件
这两个组件负责实际上从源到目标的文件复制。当copy开始实行时，在copy-listing时生成的listing-file此时将被消费。
Hadoop中，InputFormat接口定义的方法就是如何读取文件和分割文件以提供分片给mapper。 InputSplit算是Hadoop的一种存储格式，是Hadoop定义的用来传送给每个单独的map的数据，InputSplit存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。生成InputSplit的方法可以通过InputFormat来设置。
UniformSizeInputFormat: 实现了 org.apache.hadoop.mapreduce.InputFormat. UniformSizeInputFormat的宗旨是使每个map任务大约有相同的字节数量。 通过配置文件确定map数量和每个map任务的字节数，map数量通过JobContex配置，总copy字节数通过DistCp常量配置。 private List
List
long nBytesPerSplit = (long) Math.ceil(totalSizeBytes * 1.0 / numSplits);
CopyListingFileStatus srcFileStatus = new CopyListingFileStatus();
Text srcRelPath = new Text();
long currentSplitSize = 0;
long lastSplitStart = 0;
long lastPosition = 0;
final Path listingFilePath = getListingFilePath(configuration);
if (LOG.isDebugEnabled()) {
LOG.debug("Average bytes per map: " + nBytesPerSplit +
", Number of maps: " + numSplits + ", total size: " + totalSizeBytes);
}
SequenceFile.Reader reader=null;
try {
reader = getListingFileReader(configuration);
while (reader.next(srcRelPath, srcFileStatus)) {
// 如果添加该文件导致每个map超出了最大字节数限制，将当前文件添加到新的split
if (currentSplitSize + srcFileStatus.getLen() > nBytesPerSplit && lastPosition != 0) {
FileSplit split = new FileSplit(listingFilePath, lastSplitStart,
lastPosition - lastSplitStart, null);
if (LOG.isDebugEnabled()) {
LOG.debug ("Creating split : " + split + ", bytes in split: " + currentSplitSize);
}
splits.add(split);
lastSplitStart = lastPosition;
currentSplitSize = 0;
}
currentSplitSize += srcFileStatus.getLen();
lastPosition = reader.getPosition();
}
if (lastPosition > lastSplitStart) {
FileSplit split = new FileSplit(listingFilePath, lastSplitStart,
lastPosition - lastSplitStart, null);
if (LOG.isDebugEnabled()) {
LOG.info ("Creating split : " + split + ", bytes in split: " + currentSplitSize);
}
splits.add(split);
}
} finally {
IOUtils.closeStream(reader);
}
return splits;
}
DynamicInputFormat and DynamicRecordReader: DynamicInputFormat 实现了 org.apache.hadoop.mapreduce.InputFormat， 在DFS上将copy-list分成一个个chunk，创建一系列splits，每个split根据自己的能力消费chunks，这种模式可以避免最慢的mapper拖累整个任务。 private List
List
throws IOException {
int numMaps = getNumMapTasks(jobContext.getConfiguration());
final int nSplits = Math.min(numMaps, chunks.size());
List
for (int i=0; i< nSplits; ++i) {
TaskID taskId = new TaskID(jobContext.getJobID(), TaskType.MAP, i);
chunks.get(i).assignTo(taskId);
splits.add(new FileSplit(chunks.get(i).getPath(), 0,
// Setting non-zero length for FileSplit size, to avoid a possible
// future when 0-sized file-splits are considered "empty" and skipped
// over.
getMinRecordsPerChunk(jobContext.getConfiguration()),
null));
}
DistCpUtils.publish(jobContext.getConfiguration(),
CONF_LABEL_NUM_SPLITS, splits.size());
return splits;
}
DynamicRecordReader 配合DynamicInputFormat实现一种’Worker-pattern’复制，保证能者多劳，主要干两件事：
将每个chunk的内容递交给DistCp的mapper；当当前chunk消费完后，立即获取下一个chunk。
public boolean nextKeyValue()
throws IOException, InterruptedException {
if (chunk == null) {
if (LOG.isDebugEnabled())
LOG.debug(taskId + ": RecordReader is null. No records to be read.");
return false;
}
if (chunk.getReader().nextKeyValue()) {
++numRecordsProcessedByThisMap;
return true;
}
if (LOG.isDebugEnabled())
LOG.debug(taskId + ": Current chunk exhausted. " +
" Attempting to pick up new one.");
chunk.release();
timeOfLastChunkDirScan = System.currentTimeMillis();
isChunkDirAlreadyScanned = false;
chunk = chunkContext.acquire(taskAttemptContext);
if (chunk == null) return false;
if (chunk.getReader().nextKeyValue()) {
++numRecordsProcessedByThisMap;
return true;
}
else {
return false;
}
}
CopyMapper: 实现 Mapper, 该类实现物理上的文件拷贝. 输入路径，参照指定参数决定是否拷贝文件. 只有出现以下情形之一，文件才会被拷贝:
目标不存在同名文件；目标上存在同名文件，但文件大小不同；目标上存在同名文件，但是文件校验和不同，并且没有指定 -skipcrccheck；目标上存在同名文件，但是指定了 -overwrite；标上存在同名的文件, 但在块大小上有所不同（块大小需要被保留）。 CopyCommitter: 继承了 FileOutputCommitter, 负责DistCp的任务提交阶段:
维护目录访问权限（如果指定了相应参数）；清理元文件夹、临时文件（meta-folder, DistCp维护的文件表）；将数据从临时工作文件夹原子移动（atom-move）到最终路径（如果指定atomic-commit）；从目标上删除源上丢失的的文件（如果指定操作）；清理所有不完全拷贝文件。
Map sizing
默认地，DistCp尝试以可比较地调整每个map的大小，让其拷贝大约相同字节数量。注意，文件是最好的粒度级别，因此，增加并行的copier（map）的数量并不总能相应的提高并行拷贝的数量和总的吞吐量。
新的DistCp同时提供一种策略来"动态"调整maps大小的，允许较快的data-nodes扶正更多的字节。使用 -strategy dynamic, 将文件（files）分割成多个sets，而不是为每个map任务分配固定的源文件集合。sets的数量限制maps的数量，通常使用因子2-3，每个map获取并拷贝一个chunk中所有列出的文件。当一个使用完一个chunk，会获取并处理一个新的chunk，直到没有更多的chunks。
最终，处理速度较快的map任务将会处理更多的chunks，就mapper的处理能力来看，是公平的。
这种动态策略通过DynamicInputFormat实现。大多数情况下能够获得较好的性能。
对于长时间运行、周期性的任务，建议将maps的数量调整为源群集和目标群集的大小、副本的大小以及可用带宽。（Tuning the number of maps to the size of the source and destination clusters, the size of the copy, and the available bandwidth is recommended for long-running and regularly run jobs.）
不同版本间拷贝
对于两个不同主版本Hadoop之间的拷贝来说(例如：1.X和2.X)，其中一个通常使用 WebHdfsFileSystem，不同于 HftpFileSystem，webhdfs适用于读和写操作，hftp是只读操作。DistCp可以运行在源集群或目标集群上。远程集群指定 webhdfs://
MapReduce 和 其他副作用
As has been mentioned in the preceding, should a map fail to copy one of its inputs, there will be several side-effects. 之前提到过，万一一个map拷贝inputs（属于该map的）中的一个失败了，将会带来一些副作用：
除非指定了 -overwrite，否则之前的map成功拷贝的文件将被标记为"skipped"；如果一个map在尝试最大次数(mapreduce.map.maxattempts)后失败，剩下的map任务将被杀死(除非指定 -i)；If mapreduce.map.speculative is set set final and true, the result of the copy is undefined.如果设置 mapreduce.map.speculative(hadoop的推测执行，通常集群中的机器配置差异较大才会打开) 为 “true”，拷贝的结果是未定义。
从源码理解DistCp
DistCp
DistCp实现Tool，这样可以使用ToolRunner调度。ToolRunner调度器执行时，调用DistCp的run()方法，一切从这里开始。
阅读DistCp的构造器，首先在DistCp初始化时处理了配置、命令参数等。
public DistCp(Configuration configuration, DistCpOptions inputOptions) throws Exception {
Configuration config = new Configuration(configuration);
config.addResource(DISTCP_DEFAULT_XML);
setConf(config);
this.inputOptions = inputOptions;
this.metaFolder = createMetaFolderPath();
}
run()是一个调度方法，调起execute()，校验输入、处理异常等，
// 调度方法，省掉了部分catch
@Override
public int run(String[] argv) {
if (argv.length < 1) {
OptionsParser.usage();
return DistCpConstants.INVALID_ARGUMENT;
}
try {
inputOptions = (OptionsParser.parse(argv));
setTargetPathExists();
LOG.info("Input Options: " + inputOptions);
} catch (Throwable e) {
}
try {
execute(); // 调用执行
} catch (Exception e) {
LOG.error("Exception encountered ", e);
return DistCpConstants.UNKNOWN_ERROR;
}
return DistCpConstants.SUCCESS;
}
public Job execute() throws Exception {
Job job = createAndSubmitJob();
if (inputOptions.shouldBlock()) {
waitForJobCompletion(job);
}
return job;
}
excute()方法是执行任务，调用createAndSubmitJob()，该方法创建并提交任务到hadoop集群运行。
/**
* 创建并提交MR任务
* @return 已提交的MR任务实例
*/
public Job createAndSubmitJob() throws Exception {
assert inputOptions != null;
assert getConf() != null;
Job job = null;
try {
synchronized(this) {
//Don't cleanup while we are setting up.
metaFolder = createMetaFolderPath();
jobFS = metaFolder.getFileSystem(getConf());
job = createJob();
}
if (inputOptions.shouldUseDiff()) {
if (!DistCpSync.sync(inputOptions, getConf())) {
inputOptions.disableUsingDiff();
}
}
createInputFileListing(job);
job.submit();
submitted = true;
} finally {
if (!submitted) {
cleanup();
}
}
String jobID = job.getJobID().toString();
job.getConfiguration().set(DistCpConstants.CONF_LABEL_DISTCP_JOB_ID, jobID);
LOG.info("DistCp job-id: " + jobID);
return job;
}
下面分析源码中的几个关键字。
metaFolder
在DistCp中定义为 private Path metaFolder，是一个 org.apache.hadoop.fs.Path 类型，顾名思义，是准备元数据的地方。
createMetaFolderPath 结合一个随机数生成一个临时目录。
private Path createMetaFolderPath() throws Exception {
Configuration configuration = getConf();
Path stagingDir = JobSubmissionFiles.getStagingDir(
new Cluster(configuration), configuration);
Path metaFolderPath = new Path(stagingDir, PREFIX + String.valueOf(rand.nextInt()));
if (LOG.isDebugEnabled())
LOG.debug("Meta folder location: " + metaFolderPath);
configuration.set(DistCpConstants.CONF_LABEL_META_FOLDER, metaFolderPath.toString());
return metaFolderPath;
}
createInputFileListing(Job job) 注意到在metaFolder下，通过getFileListingPath()生成fileList.seq文件，稍后会往fileList.seq中写入数据，这是一个SequenceFile文件，SequenceFile是一个由二进制序列化过的key/value的字节流组成的文本存储文件，这个文件里将存放所有需要拷贝的源目录/文件信息列表。我们下面将介绍fileList.seq文件。
protected Path createInputFileListing(Job job) throws IOException {
Path fileListingPath = getFileListingPath();
CopyListing copyListing = CopyListing.getCopyListing(job.getConfiguration(),
job.getCredentials(), inputOptions);
copyListing.buildListing(fileListingPath, inputOptions);
return fileListingPath;
}
protected Path getFileListingPath() throws IOException {
String fileListPathStr = metaFolder + "/fileList.seq";
Path path = new Path(fileListPathStr);
return new Path(path.toUri().normalize().toString());
}
fileList.seq文件
这个文件里将存放所有需要拷贝的源目录/文件信息列表，注意到上面 createInputFileListing 中的 copyListing.buildListing(fileListingPath, inputOptions) 方法，如前所述，CopyListing有三种实现，最终都调用了其中SimpleCopyList类的buildList方法，该方法就是用来收集要拷贝的文件列表并写入fileList.seq文件，如下简略代码中，第一个参数就是Hadoop的Writer，通过 getWriter(Path) 获取。写入SequenceFile中的Key是源文件的Text格式的相对路径，即relPath；而Value则记录源文件的FileStatus格式的org.apache.hadoop.fs.FileStatus信息，这里FileStatus是hadoop已经封装好了的描述HDFS文件信息的类，但是DISTCP为了更好的处理数据，重新继承并封装了CopyListingFileStatus类。如下图，主要增加了aclEntries和xAttrs成员变量，关于文件权限和属性的，并覆盖了图中的方法。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SfNa7vYP-1652686190231)(https://i.imgur.com/38o3TKy.png)]
public void doBuildListing(SequenceFile.Writer fileListWriter,
DistCpOptions options) throws IOException {
// 省略...
}
private SequenceFile.Writer getWriter(Path pathToListFile) throws IOException {
FileSystem fs = pathToListFile.getFileSystem(getConf());
if (fs.exists(pathToListFile)) {
fs.delete(pathToListFile, false);
}
return SequenceFile.createWriter(getConf(),
SequenceFile.Writer.file(pathToListFile),
SequenceFile.Writer.keyClass(Text.class),
SequenceFile.Writer.valueClass(CopyListingFileStatus.class),
SequenceFile.Writer.compression(SequenceFile.CompressionType.NONE));
}
Job
如下为创建任务的源码，和平时创建任务一样，创建Job对象，设置名称，配置输入输出路径，设置Mapper类和Reducer，注意这里只有map任务，设置输入输出类型等等，在这里最终将创建好的job返回给excute()，一层层向上，然后送给调度器。下面关注与文件拷贝相关的几点：
private Job createJob() throws IOException {
String jobName = "distcp";
String userChosenName = getConf().get(JobContext.JOB_NAME);
if (userChosenName != null)
jobName += ": " + userChosenName;
Job job = Job.getInstance(getConf());
job.setJobName(jobName);
job.setInputFormatClass(DistCpUtils.getStrategy(getConf(), inputOptions));
job.setJarByClass(CopyMapper.class);
configureOutputFormat(job);
job.setMapperClass(CopyMapper.class);
job.setNumReduceTasks(0);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputFormatClass(CopyOutputFormat.class);
job.getConfiguration().set(JobContext.MAP_SPECULATIVE, "false");
job.getConfiguration().set(JobContext.NUM_MAPS,
String.valueOf(inputOptions.getMaxMaps()));
if (inputOptions.getSslConfigurationFile() != null) {
setupSSLConfig(job);
}
inputOptions.appendToConf(job.getConfiguration());
return job;
}
InputFormat
在MapReduce中，InputFormat用来描述输入数据的格式，提供以下两个功能：
数据切分，按照某个策略将输入数据且分成若干个split，以便确定MapTask的个数即Mapper的个数，在MapReduce框架中，一个split就意味着需要一个Map Task;为Mapper提供输入数据，即给定一个split，(使用其中的RecordReader对象)将之解析为一个个的key/value键值对。
如下，getSplits用于切分数据，是一种逻辑分片的概念，数据还是按block存储，查看InputSplit源码，InputSplit只记录了Mapper要处理的数据的元数据信息，如起始位置、长度和所在的节点；
public abstract class InputFormat
public abstract
List
) throws IOException, InterruptedException;
public abstract
RecordReader
TaskAttemptContext context
) throws IOException,
InterruptedException;
}
public abstract class InputSplit {
public abstract long getLength() throws IOException, InterruptedException;
public abstract
String[] getLocations() throws IOException, InterruptedException;
@Evolving
public SplitLocationInfo[] getLocationInfo() throws IOException {
return null;
}
}
在DistCp中，有两种输入格式，在上一节创建job代码中可以看到 job.setInputFormatClass(DistCpUtils.getStrategy(getConf(), inputOptions)); 是根据配置设置输入格式的。 查看 DistCpConstants 类可以发现 public static final String UNIFORMSIZE = "uniformsize";, 默认使用的是uniformsize，我们可以通过参数指定使用dynamic模式，两者区别参考1.3.2小节。
CopyMapper
在创建任务中，通过 job.setMapperClass(CopyMapper.class); 设置了mapper类。CopyMapper是真正的mapper操作，hadoop中的mapper任务就是按照这个逻辑走的。在DistCp中没有reduce，仅有mapper任务，CopyMapper与其他的mapper没有什么区别，也是实现了基类Mapper中的setup()和map()` 两个核心方法。
setup: 初始化、参数读取、文件校验等前期工作。
@Override
public void setup(Context context) throws IOException, InterruptedException {
conf = context.getConfiguration();
syncFolders = conf.getBoolean(DistCpOptionSwitch.SYNC_FOLDERS.getConfigLabel(), false);
ignoreFailures = conf.getBoolean(DistCpOptionSwitch.IGNORE_FAILURES.getConfigLabel(), false);
skipCrc = conf.getBoolean(DistCpOptionSwitch.SKIP_CRC.getConfigLabel(), false);
overWrite = conf.getBoolean(DistCpOptionSwitch.OVERWRITE.getConfigLabel(), false);
append = conf.getBoolean(DistCpOptionSwitch.APPEND.getConfigLabel(), false);
preserve = DistCpUtils.unpackAttributes(conf.get(DistCpOptionSwitch.
PRESERVE_STATUS.getConfigLabel()));
targetWorkPath = new Path(conf.get(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH));
Path targetFinalPath = new Path(conf.get(
DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH));
targetFS = targetFinalPath.getFileSystem(conf);
if (targetFS.exists(targetFinalPath) && targetFS.isFile(targetFinalPath)) {
overWrite = true; // When target is an existing file, overwrite it.
}
if (conf.get(DistCpConstants.CONF_LABEL_SSL_CONF) != null) {
initializeSSLConf(context);
}
}
map: 从方法参数可以看出，是对InputFormat分片数据每一行进行处理，其中省略了大部分代码，基本都是针对命令行参数对文件做一些前期校验，是否跳过等。最终如果需要拷贝，会通过 RetriableFileCopyCommand 类使用真正拷贝字节的方法 copyBytes, 就是普通的文件流操作了。
public void map(Text relPath, CopyListingFileStatus sourceFileStatus,
Context context) throws IOException, InterruptedException {
copyFileWithRetry(description, sourceCurrStatus, target, context, action, fileAttributes);
}
long copyBytes(FileStatus sourceFileStatus, long sourceOffset,
OutputStream outStream, int bufferSize, Mapper.Context context)
throws IOException {
Path source = sourceFileStatus.getPath();
byte buf[] = new byte[bufferSize];
ThrottledInputStream inStream = null;
long totalBytesRead = 0;
try {
inStream = getInputStream(source, context.getConfiguration());
int bytesRead = readBytes(inStream, buf, sourceOffset);
while (bytesRead >= 0) {
totalBytesRead += bytesRead;
if (action == FileAction.APPEND) {
sourceOffset += bytesRead;
}
outStream.write(buf, 0, bytesRead);
updateContextStatus(totalBytesRead, context, sourceFileStatus);
bytesRead = readBytes(inStream, buf, sourceOffset);
}
outStream.close();
outStream = null;
} finally {
IOUtils.cleanup(LOG, outStream, inStream);
}
return totalBytesRead;
}
命令参数
Flag Description Notes
-p[rbugpcaxt] Preserve r: replication number
b: block size
u: user
g: group
p: permission
c: checksum-type
a: ACL
x: XAttr
t: timestamp 当指定
-update，更新的状态
不会同步，除非文件大小不同（例如：重新创建文件）。 如果指定
-pa， DistCp 保留文件权限。
-i 忽略失败 相较默认情况，该选项会提供更精确的拷贝统计，并且保存拷贝失败的日志，方便调试。在没有文成任务中所有分块的尝试之前，一个map的失败不会导致整个任务失败。
-log
-v SKIP/COPY日志中的附加信息 (path, size) 该选项只能与
-log 一起使用.
-m
-overwrite 覆盖目标文件 如果map失败了并且没有指定
-i, 不仅仅那些失败的文件，这个split中的所有文件将被重新拷贝。 正如使用手册中的说明，这同时会改变生成目标路径的语义(it also changes the semantics for generating destination paths)，用户应该谨慎使用。
-update 如果目标的size(大小), blocksize(块大小), 或 checksum(校验和) 不同，则进行覆盖 这并非“同步”操作，是否执行覆盖的标准是源和目标的文件大小、块大小和校验和是否相同; 如果不同源文件替换目标文件。
-append 同名而不同长度的文件进行增量拷贝 如果源文件的长度比目标文件大，将会检验共同长度部分的校验和，如果校验和匹配, 使用read和append功能，仅将不同的部分拷贝。
-append 与
-update 并不带
-skipcrccheck
-f
urilist_uri 列表应该是一个完整的、合法的URI。
-filters The path to a file containing a list of pattern strings, one string per line, such that paths matching the pattern will be excluded from the copy. Support regular expressions specified by java.util.regex.Pattern.
-filelimit
Deprecated! Ignored in the new DistCp.
-sizelimit
Deprecated! Ignored in the new DistCp.
-delete Delete the files existing in the dst but not in src The deletion is done by FS Shell. So the trash will be used, if it is enable. Delete is applicable only with update or overwrite options.
-strategy {dynamic|uniformsize} 选择DistCp的拷贝策略 默认使用uniformsize。 (即，根据拷贝文件的总大小平衡每个map，与传统相仿。) 如果指定 “dynamic”， 将会使用
DynamicInputFormat。 (详见体系架构)
-bandwidth 给每个map指定带宽, in MB/second. Each map will be restricted to consume only the specified bandwidth. This is not always exact. The map throttles back its bandwidth consumption during a copy, such that the
net bandwidth used tends towards the specified value.
-atomic {-tmp
-atomic instructs DistCp to copy the source data to a temporary target location, and then move the temporary target to the final-location atomically. Data will either be available at final target in a complete and consistent form, or not at all. Optionally,
-tmp may be used to specify the location of the tmp-target. If not specified, a default is chosen.
Note: tmp_dir must be on the final target cluster.
-mapredSslConf
-async 异步执行DistCp. Hadoop任务一运行立即返回。 记录Hadoop任务id， 以便追踪。
-diff
-update option and the following conditions should be satisfied.
Both the source and the target FileSystem must be DistributedFileSystem. Two snapshots
-rdiff
-update option and the following conditions should be satisfied.
Both the source and the target FileSystem must be DistributedFileSystem. The source and the target can be two different clusters/paths, or they can be exactly the same cluster/path. In the latter case, modified files are copied from target’s
-numListstatusThreads Number of threads to use for building file listing At most 40 threads.
-skipcrccheck 是否跳过源路径和目标路径之间的 crc 检查。
-blocksperchunk
-copybuffersize
相关文章
您阅读本篇文章共花了：
发表评论