如何重命名 Hadoop/Spark 中的大量文件？答案

【问题标题】：How to rename huge amount of files in Hadoop/Spark?如何重命名 Hadoop/Spark 中的大量文件？
【发布时间】：2014-08-29 06:38:04
【问题描述】：

我有一个包含 +100,000 个文件的输入文件夹。

我想对它们进行批量操作，即以某种方式重命名它们，或者根据每个文件名中的信息将它们移动到新路径。

我想使用 Spark 来做这件事，但不幸的是，当我尝试以下代码时：

 final org.apache.hadoop.fs.FileSystem ghfs = org.apache.hadoop.fs.FileSystem.get(new java.net.URI(args[0]), new org.apache.hadoop.conf.Configuration());
        org.apache.hadoop.fs.FileStatus[] paths = ghfs.listStatus(new org.apache.hadoop.fs.Path(args[0]));
        List<String> pathsList = new ArrayList<>();
        for (FileStatus path : paths) {
            pathsList.add(path.getPath().toString());
        }
        JavaRDD<String> rddPaths = sc.parallelize(pathsList);

        rddPaths.foreach(new VoidFunction<String>() {
            @Override
            public void call(String path) throws Exception {
                Path origPath = new Path(path);
                Path newPath = new Path(path.replace("taboola","customer"));
                ghfs.rename(origPath,newPath);
            }
        });

我收到 hadoop.fs.FileSystem 不可序列化的错误（因此可能无法用于并行操作）

知道如何解决它或以其他方式解决吗？

【问题讨论】：

标签： hadoop parallel-processing bigdata apache-spark

【解决方案1】：

问题是您正在尝试序列化 ghfs 对象。如果您使用 mapPartitions 并在每个分区中重新创建 ghfs 对象，您只需进行一些小的更改即可运行您的代码。

【讨论】：

谢谢！这正是我将要采取的方向！

【解决方案2】：

您也需要在VoidFunction 内执行FileSystem.get。

驱动程序需要一个文件系统来获取文件列表，但每个工作人员也需要一个文件系统来重命名。驱动程序无法将其文件系统传递给工作人员，因为它不是可序列化的。但是工人可以得到自己的文件系统就好了。

在 Scala API 中，您可以使用 RDD.foreachPartition 来编写代码，即每个分区只执行一次 FileSystem.get，而不是每行一次。它可能在 Java API 中也可用。

【讨论】：

谢谢！这正是我将要采取的方向（目前只有我使用 Java...）
在所有争论之后，OP 接受了一个迟到的答案，这基本上是你的一个子集
争吵太多的恶业:)。
使用 spark 重命名 hdfs 中的部分文件嗨我正在尝试将 part-*.csv 文件从 hdfs 重命名为 .csv 我在 sn-p 下面尝试了但目录结构没有根据我的需要创建这里是我的试试，

【解决方案3】：

我建议只在非 map reduce 上下文中（仅在驱动程序中）像使用文件系统类一样重命名它们，重命名 100k 文件没什么大不了的，它太慢了，然后你可以尝试多线程它。只需做类似的事情

FileSystem fileSystem = new Path("").getFileSystem(new Configuration());
File [] files =  FileUtil.listFiles(directory)
for (File file : files) {
    fileSystem.rename(new Path(file.getAbsolutePath()),new Path("renamed"));
}

顺便说一句，您在 spark 中遇到的错误是因为 spark 需要它用来实现 Serializable 的对象，而 FileSystem 不需要。

我无法确认这一点，但似乎 HDFS 中的每个重命名都会涉及 NameNode，因为它会跟踪文件的完整目录结构和节点位置 (confirmation link)，这意味着它不能有效地并行完成.根据this answer，重命名是仅元数据操作，因此它应该非常快速地连续运行。

【讨论】：

问题不在于修改 RDD 中的任何内容。它是关于以分布式方式为 RDD 的每一行运行一些操作。也许你有耐心在一台机器上运行重命名，但绝对有效地使用 Spark 将这项工作分发到多台机器。
@DanielDarabos 好的，你说得对，RDD 的不变性与从答案中删除的这个问题无关。但我仍然认为 spark 对于重命名某些文件来说是多余的
@DanielDarabos 对此也没有引用我的话，但是重命名同一目录中的文件肯定不能在 HDFS 上并行完成，因此您的解决方案会遇到映射器被阻止的问题
可能！如果 OP 能够报告这两种方法的相对速度，那就太好了。
@DanielDarabos 所以我无法确认，因为 hdfs 的来源有太多可能来自的地方，但我相信每个重命名操作都涉及名称节点（添加到答案）。也反对，是你，除非你仍然认为我的回答有问题，我会很感激你删除它

【解决方案4】：

当我的 hdfs 存档目录达到最大项目限制时，我遇到了类似的问题

Request error: org.apache.hadoop.hdfs.protocol.FSLimitException$MaxDirectoryItemsExceededException
The directory item limit of /my/archive is exceeded: limit=1048576 items=1048576

我决定将所有项目从上一年（2015 年）移到单独的子文件夹中。这里是纯shell解决方案

export HADOOP_CLIENT_OPTS="-XX:-UseGCOverheadLimit -Xmx4096m"
hdfs dfs -ls /my/archive \
    | grep 2015- \
    | awk '{print $8}' \
    | gnu-parallel -X -s 131000 hdfs dfs -mv {} /my/archive/2015

备注：

hdfs dfs -ls 需要客户端选择覆盖，因为文件量很大。有关详细信息，请参阅here。
hdfs dfs 客户端的参数列表长度有限制：大约 131000 (2^17) 字符。
移动 420k 文件需要几分钟时间。

【讨论】：