如何在Hadoop 3.0中进行CopyMerge？

Question

我知道hadoop 2.7版的FileUtil具有copyMerge功能，可以将多个文件合并到一个新文件中。

但是，3.0版的API不再支持copyMerge功能。

在hadoop的3.0版本中，如何将目录中的所有文件合并到一个新的单个文件中，有什么想法吗？

Answer 1

我也有同样的问题，必须重新实现copyMerge（虽然是在PySpark中，但使用与原始copyMerge相同的API调用）。

不知道为什么Hadoop 3中没有等效的功能。我们必须经常将文件从HDFS目录合并到HDFS文件。

这是我在上面提到的pySpark中的实现https://github.com/Tagar/stuff/blob/master/copyMerge.py

Answer 2

FileUtil#copyMerge方法已删除。主要变更详见：

https://issues.apache.org/jira/browse/HADOOP-12967

https://issues.apache.org/jira/browse/HADOOP-11392

可以使用getmerge

用法：hadoop fs-getmerge[-nl]

将源目录和目标文件作为输入，并将src中的文件连接到目标本地文件中。（可选）-nl可以设置为允许在每个文件的末尾添加换行符（LF）-“跳过空文件”可用于避免出现空文件时出现不需要的换行符。

示例：

hadoop fs -getmerge -nl /src /opt/output.txt
hadoop fs -getmerge -nl /src/file1.txt /src/file2.txt /output.txt

退出代码：成功时返回0，错误时返回非零。

https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/FileSystemShell.html#getmerge

Answer 3

自FileUtil。copyMerge（）已被弃用并从版本3开始的API中删除，我们始终可以自己重新实现它。

这是以前版本的原始Java实现。

以下是Scala翻译：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.IOUtils
import java.io.IOException

def copyMerge(
    srcFS: FileSystem, srcDir: Path,
    dstFS: FileSystem, dstFile: Path,
    deleteSource: Boolean, conf: Configuration
): Boolean = {

  if (dstFS.exists(dstFile)) {
    throw new IOException(s"Target $dstFile already exists")
  }

  // Source path is expected to be a directory:
  if (srcFS.getFileStatus(srcDir).isDirectory) {

    val outputFile = dstFS.create(dstFile)
    try {
      srcFS
        .listStatus(srcDir)
        .sortBy(_.getPath.getName)
        .collect {
          case status if status.isFile =>
            val inputFile = srcFS.open(status.getPath)
            try { IOUtils.copyBytes(inputFile, outputFile, conf, false) }
            finally { inputFile.close() }
        }
    } finally { outputFile.close() }

    if (deleteSource) srcFS.delete(srcDir, true) else true
  }
  else false
}