Hadoop集群间文件拷贝

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 Driver
• Copy-listing generator
• Input-formats 和 Map-Reduce components

DistCp Driver

// 构造方法，根据输入的参数创建DistCppublicDistCp(Configuration configuration,DistCpOptions inputOptions)throwsException{Configuration config =newConfiguration(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.
     */publicintrun(String[] argv){if(argv.length <1){OptionsParser.usage();returnDistCpConstants.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();returnDistCpConstants.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消费。其主要包括以下模块：1. CopyListing: 上层接口，具体的copy-listing-generator通过实现此接口来实现列表生成功能。提供工厂方法供构造CopyListing的实现来选择；[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-emW801oA-1652686190230)(https://i.imgur.com/EjuUxrL.png)]1. SimpleCopyListing: CopyListing的一个实现，接收多源路径，并且递归地列出每个path下的所有单独文件和目录，用于copy；2. GlobbedCopyListing: CopyListing的一个实现，支持源路径存在通配符；3. FileBasedCopyListing: CopyListing的一个实现，从一个指定文件中读取源路径列表；根据是否在DistCpOptions中指定源文件列表，source-list通过以下方式生成：1. 如果没有指定source-file-list, 使用GlobbedCopyListing。 扩展所有的通配符，所有扩展发送给SimpleCopyListing，依次构造listing(通过向下递归方式)；2. 如果指定source-file-list, 使用FileBasedCopyListing。从指定的文件中读取Source-paths, 然后发送给GlobbedCopyListing，像上面一样构造listing。GlobbedCopyListing构造listing如下：

@OverridepublicvoiddoBuildListing(Path pathToListingFile,DistCpOptions options)throwsIOException{List<Path> globbedPaths =newArrayList<Path>();if(options.getSourcePaths().isEmpty()){thrownewInvalidInputException("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{thrownewInvalidInputException(p +" doesn't exist");}}DistCpOptions optionsGlobbed =newDistCpOptions(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常量配置。privateList<InputSplit>getSplits(Configuration configuration,int numSplits,long totalSizeBytes)throwsIOException{List<InputSplit> splits =newArrayList<InputSplit>(numSplits);long nBytesPerSplit =(long)Math.ceil(totalSizeBytes *1.0/ numSplits);CopyListingFileStatus srcFileStatus =newCopyListingFileStatus();Text srcRelPath =newText();long currentSplitSize =0;long lastSplitStart =0;long lastPosition =0;finalPath 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超出了最大字节数限制，将当前文件添加到新的splitif(currentSplitSize + srcFileStatus.getLen()> nBytesPerSplit && lastPosition !=0){FileSplit split =newFileSplit(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 =newFileSplit(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拖累整个任务。privateList<InputSplit>createSplits(JobContext jobContext,List<DynamicInputChunk> chunks)throwsIOException{int numMaps =getNumMapTasks(jobContext.getConfiguration());finalint nSplits =Math.min(numMaps, chunks.size());List<InputSplit> splits =newArrayList<InputSplit>(nSplits);for(int i=0; i< nSplits;++i){TaskID taskId =newTaskID(jobContext.getJobID(),TaskType.MAP, i); chunks.get(i).assignTo(taskId); splits.add(newFileSplit(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’复制，保证能者多劳，主要干两件事：1. 将每个chunk的内容递交给DistCp的mapper；2. 当当前chunk消费完后，立即获取下一个chunk。

publicbooleannextKeyValue()throwsIOException,InterruptedException{if(chunk ==null){if(LOG.isDebugEnabled())LOG.debug(taskId +": RecordReader is null. No records to be read.");returnfalse;}if(chunk.getReader().nextKeyValue()){++numRecordsProcessedByThisMap;returntrue;}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)returnfalse;if(chunk.getReader().nextKeyValue()){++numRecordsProcessedByThisMap;returntrue;}else{returnfalse;}}

• 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://<namenode_hostname>:<http_port>

。当主版本相同时，使用hdfs协议性能会更好。

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初始化时处理了配置、命令参数等。

publicDistCp(Configuration configuration,DistCpOptions inputOptions)throwsException{Configuration config =newConfiguration(configuration);
    config.addResource(DISTCP_DEFAULT_XML);setConf(config);this.inputOptions = inputOptions;this.metaFolder   =createMetaFolderPath();}

run()是一个调度方法，调起execute()，校验输入、处理异常等，

// 调度方法，省掉了部分catch@Overridepublicintrun(String[] argv){if(argv.length <1){OptionsParser.usage();returnDistCpConstants.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);returnDistCpConstants.UNKNOWN_ERROR;}returnDistCpConstants.SUCCESS;}publicJobexecute()throwsException{Job job =createAndSubmitJob();if(inputOptions.shouldBlock()){waitForJobCompletion(job);}return job;}

excute()方法是执行任务，调用createAndSubmitJob()，该方法创建并提交任务到hadoop集群运行。

/**
     * 创建并提交MR任务
     * @return 已提交的MR任务实例
     */publicJobcreateAndSubmitJob()throwsException{assert inputOptions !=null;assertgetConf()!=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结合一个随机数生成一个临时目录。

privatePathcreateMetaFolderPath()throwsException{Configuration configuration =getConf();Path stagingDir =JobSubmissionFiles.getStagingDir(newCluster(configuration), configuration);Path metaFolderPath =newPath(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文件。

protectedPathcreateInputFileListing(Job job)throwsIOException{Path fileListingPath =getFileListingPath();CopyListing copyListing =CopyListing.getCopyListing(job.getConfiguration(),
            job.getCredentials(), inputOptions);
        copyListing.buildListing(fileListingPath, inputOptions);return fileListingPath;}protectedPathgetFileListingPath()throwsIOException{String fileListPathStr = metaFolder +"/fileList.seq";Path path =newPath(fileListPathStr);returnnewPath(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)]

publicvoiddoBuildListing(SequenceFile.Writer fileListWriter,DistCpOptions options)throwsIOException{// 省略...}privateSequenceFile.WritergetWriter(Path pathToListFile)throwsIOException{FileSystem fs = pathToListFile.getFileSystem(getConf());if(fs.exists(pathToListFile)){
      fs.delete(pathToListFile,false);}returnSequenceFile.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()，一层层向上，然后送给调度器。下面关注与文件拷贝相关的几点：

privateJobcreateJob()throwsIOException{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要处理的数据的元数据信息，如起始位置、长度和所在的节点；

publicabstractclassInputFormat<K,V>{publicabstractList<InputSplit>getSplits(JobContext context
                                   )throwsIOException,InterruptedException;publicabstractRecordReader<K,V>createRecordReader(InputSplit split,TaskAttemptContext context
                                            )throwsIOException,InterruptedException;}publicabstractclassInputSplit{publicabstractlonggetLength()throwsIOException,InterruptedException;publicabstractString[]getLocations()throwsIOException,InterruptedException;@EvolvingpublicSplitLocationInfo[]getLocationInfo()throwsIOException{returnnull;}}

CopyMapper

在创建任务中，通过

job.setMapperClass(CopyMapper.class);

设置了mapper类。CopyMapper

是真正的mapper操作，hadoop中的mapper任务就是按照这个逻辑走的。在DistCp中没有reduce，仅有mapper任务，CopyMapper与其他的mapper没有什么区别，也是实现了基类

Mapper

中的

setup()

和

map()` 两个核心方法。

• setup: 初始化、参数读取、文件校验等前期工作。

@Overridepublicvoidsetup(Context context)throwsIOException,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 =newPath(conf.get(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH));Path targetFinalPath =newPath(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, 就是普通的文件流操作了。

publicvoidmap(Text relPath,CopyListingFileStatus sourceFileStatus,Context context)throwsIOException,InterruptedException{copyFileWithRetry(description, sourceCurrStatus, target, context, action, fileAttributes);}longcopyBytes(FileStatus sourceFileStatus,long sourceOffset,OutputStream outStream,int bufferSize,Mapper.Context context)throwsIOException{Path source = sourceFileStatus.getPath();byte buf[]=newbyte[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 <logdir> 将日志记录到 <logdir> DistCp为每个尝试拷贝的操作记录日志并作为map任务的输出。如果一个map失败了，重新执行这个map，日志将不会保留。
-v SKIP/COPY日志中的附加信息 (path, size) 该选项只能与
-log 一起使用.
-m <num_maps> 并行拷贝的最大数量（多少个copy） 指定拷贝数据的map数量。 注意增加map数量并不能够有效提高吞吐量。
-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> 使用 <urilist_uri> 作为源文件列表 意思就是把每个源文件名列在命令行中。
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 <n> Limit the total number of files to be <= n
Deprecated! Ignored in the new DistCp.
-sizelimit <n> Limit the total size to be <= n bytes
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 <tmp_dir>} Specify atomic commit, with optional tmp directory.
-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 <ssl_conf_file> Specify SSL Config file, to be used with HSFTP source When using the hsftp protocol with a source, the security- related properties may be specified in a config-file and passed to DistCp. <ssl_conf_file> needs to be in the classpath.
-async 异步执行DistCp. Hadoop任务一运行立即返回。 记录Hadoop任务id， 以便追踪。
-diff <oldSnapshot> <newSnapshot> Use snapshot diff report between given two snapshots to identify the difference between source and target, and apply the diff to the target to make it in sync with source. This option is valid only with
-update option and the following conditions should be satisfied.

1. Both the source and the target FileSystem must be DistributedFileSystem.
2. Two snapshots <oldSnapshot> and <newSnapshot> have been created on the source FS, and <oldSnapshot> is older than <newSnapshot>.
3. The target has the same snapshot <oldSnapshot>. No changes have been made on the target since <oldSnapshot> was created, thus <oldSnapshot> has the same content as the current state of the target. All the files/directories in the target are the same with source’s <oldSnapshot>.

-rdiff <newSnapshot> <oldSnapshot> Use snapshot diff report between given two snapshots to identify what has been changed on the target since the snapshot
<oldSnapshot> was created on the target, and apply the diff reversely to the target, and copy modified files from the source’s
<oldSnapshot>, to make the target the same as
<oldSnapshot>. This option is valid only with
-update option and the following conditions should be satisfied.

1. 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 <oldSnapshot> to target’s current state).
2. Two snapshots <newSnapshot> and <oldSnapshot> have been created on the target FS, and <oldSnapshot> is older than <newSnapshot>. No change has been made on target since <newSnapshot> was created on the target.
3. The source has the same snapshot <oldSnapshot>, which has the same content as the <oldSnapshot> on the target. All the files/directories in the target’s <oldSnapshot> are the same with source’s <oldSnapshot>.

-numListstatusThreads Number of threads to use for building file listing At most 40 threads.
-skipcrccheck 是否跳过源路径和目标路径之间的 crc 检查。
-blocksperchunk <blocksperchunk> 每个chunk的块数量，指定后，将文件分割成多个块并行复制。 If set to a positive value, files with more blocks than this value will be split into chunks of
<blocksperchunk> blocks to be transferred in parallel, and reassembled on the destination. By default,
<blocksperchunk> is 0 and the files will be transmitted in their entirety without splitting. This switch is only applicable when the source file system implements getBlockLocations method and the target file system implements concat method.
-copybuffersize <copybuffersize> Size of the copy buffer to use. By default,
<copybuffersize> is set to 8192B