HDFS Federation

二次开发:由于HDFS-2139是单客户端执行方式，相比distcp有数十倍的性能提升，但对于拷贝上亿文件快，几P的数据速度仍然无法满足需求，将fastcp改为分布式模式执行。
   hadoop distfastcp  -strategy dynamic   -prbugpcaxt -update -skipcrccheck -i -delete     hdfs://ip:9000/$1    hdfs://ip:9000/$1
我们将distcp数据拷贝逻辑改造为fastcp逻辑，将distcp中的CopyMapper中的copyFileWithRetry改为调用fastcopy   FastCopy.CopyResult c = fcp.copy(sourceFileStatus.toString(), target.toString(),    (DistributedFileSystem) srcFileSys,(DistributedFileSystem) targetFS);
解决map数据倾斜（利用dynamic strategy,同时将UniformSizeInputFormat的getSplits按照文件个数划分）解决文件更新覆盖删除等问题扩展acl属性及存储策略的拷贝    BlockStoragePolicy[] policies =  ((DistributedFileSystem)fileSystem).getStoragePolicies();    HdfsFileStatus status = ((DistributedFileSystem)fileSystem).getClient().getFileInfo(fileStatus.getPath().makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory()).toUri().getPath());    byte storagePolicyId = status.getStoragePolicy();    for (BlockStoragePolicy p : policies) {      if (p.getId() == storagePolicyId) {        copyListingFileStatus.setStoragePolicy(p.getName());      }    }    if(srcFileStatus.getStoragePolicy()!=null) {      ((DistributedFileSystem) targetFS).setStoragePolicy(path, srcFileStatus.getStoragePolicy());    }
多线程列表生成器，如果分区文件较大，单线程列表生成器成为瓶颈导致任务启动慢                if (explore) {                    CountDownLatch countDownLatch = new CountDownLatch(sourceFiles.length);                    ExecutorService listFileExecutor= Executors.newFixedThreadPool(100);                    for (FileStatus sourceStatus : sourceFiles) {                        listFileExecutor.execute(new ListFile(countDownLatch,sourceFS,sourceStatus,sourcePathRoot,options,preserveAcls,preserveXAttrs,preserveRawXAttrs));                    }                    countDownLatch.await();                    listFileExecutor.shutdown();                }

    for (DatanodeStorageReport r : reports) {          final DDatanode dn = dispatcher.newDatanode(r.getDatanodeInfo());          LOG.info("DFSConfigKeys.DFS_BALANCE_ONLY_DISK :"+ typeStore );          if (this.typeStore) {              for (StorageType t : StorageType.getMovableTypes()) {                  if ("DISK".equals(t.name())) {                      ...                      dispatcher.getStorageGroupMap().put(g);                  }              }          } else {                  for (StorageType t : StorageType.getMovableTypes()) {                        ...                 }          }

mover限定磁盘占用率高server：
    static HashSet<String> getNameNodePathsToMove(Configuration conf,                                                       String... args) throws Exception {      final Options opts = buildCliOptions();      CommandLineParser parser = new GnuParser();      CommandLine commandLine = parser.parse(opts, args, true);      return getNameNodePaths(commandLine, conf);    }    private static HashSet<String> getNameNodePaths(CommandLine line,                                                         Configuration conf) throws Exception {
      HashSet hashSet = new HashSet();      BufferedReader reader = new BufferedReader(              new InputStreamReader(new FileInputStream(line.getOptionValue("d")), "UTF-8"));      try {        String line;        while ((line = reader.readLine()) != null) {          if (!line.trim().isEmpty()) {            hashSet.add(line1);          }        }      } finally {        IOUtils.cleanup(LOG, reader);      }      return hashSet;    }