无法从 HDFS 复制到 S3A

Question

我有一个类可以使用 Apache FileUtil 将目录内容从一个位置复制到另一个位置：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

class Folder {
    private final FileSystem fs;
    private final Path pth;

    // ... constructors and other methods

    /**
     * Copy contents (files and files in subfolders) to another folder.
     * Merges overlapping folders
     * Overwrites already existing files
     * @param destination Folder where content will be moved to
     * @throws IOException If fails
     */
    public void copyFilesTo(final Folder destination) throws IOException {
        final RemoteIterator<LocatedFileStatus> iter = this.fs.listFiles(
            this.pth,
            true
        );
        final URI root = this.pth.toUri();
        while (iter.hasNext()) {
            final Path source = iter.next().getPath();
            FileUtil.copy(
                this.fs,
                source,
                destination.fs,
                new Path(
                    destination.pth,
                    root.relativize(source.toUri()).toString()
                ),
                false,
                true,
                this.fs.getConf()
            );
        }
    }
}

这个类在单元测试中与本地 (file:///) 目录一起工作正常， 但是当我尝试在 Hadoop 集群中使用它来将文件从 HDFS (hdfs:///tmp/result) 复制到 Amazon S3 (s3a://mybucket/out) 时，它不会复制任何内容，也不会抛出错误，只是默默地跳过复制。

当我将同一个类（同时具有 HDFS 或 S3a 文件系统）用于其他目的时，它工作正常，因此此处的配置和 fs 参考应该没问题。

我做错了什么？如何正确将文件从 HDFS 复制到 S3A？

我正在使用Hadoop 2.7.3。

---

更新 我在copyFilesTo 方法中添加了更多日志以记录root、source 和target 变量（并在不更改代码的情况下提取rebase() 方法）：

    /**
     * Copy contents (files and files in subfolders) to another folder.
     * Merges overlapping folders
     * Overwrites already existing files
     * @param dst Folder where content will be moved to
     * @throws IOException If fails
     */
    public void copyFilesTo(final Folder dst) throws IOException {
        Logger.info(
            this, "copyFilesTo(%s): from %s fs=%s",
            dst, this, this.hdfs
        );
        final RemoteIterator<LocatedFileStatus> iter = this.hdfs.listFiles(
            this.pth,
            true
        );
        final URI root = this.pth.toUri();
        Logger.info(this, "copyFilesTo(%s): root=%s", dst, root);
        while (iter.hasNext()) {
            final Path source = iter.next().getPath();
            final Path target = Folder.rebase(dst.path(), this.path(), source);
            Logger.info(
                this, "copyFilesTo(%s): src=%s target=%s",
                dst, source, target
            );
            FileUtil.copy(
                this.hdfs,
                source,
                dst.hdfs,
                target,
                false,
                true,
                this.hdfs.getConf()
            );
        }
    }

    /**
     * Change the base of target URI to new base, using root
     * as common path.
     * @param base New base
     * @param root Common root
     * @param target Target to rebase
     * @return Path with new base
     */
    static Path rebase(final Path base, final Path root, final Path target) {
        return new Path(
            base, root.toUri().relativize(target.toUri()).toString()
        );
    }

在集群中运行后，我得到了这些日志：

io.Folder: copyFilesTo(hdfs:///tmp/_dst): from hdfs:///tmp/_src fs=DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_182008924_1, ugi=hadoop (auth:SIMPLE)]]
io.Folder: copyFilesTo(hdfs:///tmp/_dst): root=hdfs:///tmp/_src
INFO io.Folder: copyFilesTo(hdfs:///tmp/_dst): src=hdfs://ip-172-31-2-12.us-east-2.compute.internal:8020/tmp/_src/one.file target=hdfs://ip-172-31-2-12.us-east-2.compute.internal:8020/tmp/_src/one.file

我在 rebase() 方法中本地化了错误的代码，在 EMR 集群中运行时它无法正常工作，因为 RemoteIterator 正在以远程格式返回 URI：hdfs://ip-172-31-2-12.us-east-2.compute.internal:8020/tmp/_src/one.file 但此方法需要格式 hdfs:///tmp/_src/one.file，这就是原因它在本地与file:/// FS 合作。

Answer 1

我看不出有什么明显的错误。

1. 是 hdfs-hdfs 还是 s3a-s3a？
2. 升级你的hadoop版本； 2.7.x 已经过时了，尤其是 S3A 代码。这个问题不太可能消失，但它会避免其他问题。升级后，切换到fast upload，它将对大文件进行增量更新；目前，您的代码会将每个文件保存到 /tmp 某处，然后在 close() 调用中上传。
3. 打开 org.apache.hadoop.fs.s3a 模块的日志记录并查看它的内容

Answer 2

我不确定这是否是最佳且完全正确的解决方案，但它对我有用。这个想法是在变基之前修复本地路径的主机和端口，工作 rebase 方法将是：

    /**
     * Change the base of target URI to new base, using root
     * as common path.
     * @param base New base
     * @param root Common root
     * @param target Target to rebase
     * @return Path with new base
     * @throws IOException If fails
     */
    @SuppressWarnings("PMD.DefaultPackage")
    static Path rebase(final Path base, final Path root, final Path target)
        throws IOException {
        final URI uri = target.toUri();
        try {
            return new Path(
                new Path(
                    new URIBuilder(base.toUri())
                        .setHost(uri.getHost())
                        .setPort(uri.getPort())
                        .build()
                ),
                new Path(
                    new URIBuilder(root.toUri())
                        .setHost(uri.getHost())
                        .setPort(uri.getPort())
                        .build()
                        .relativize(uri)
                )
            );
        } catch (final URISyntaxException err) {
            throw new IOException("Failed to rebase", err);
        }
    }