hadoop - 如何在 Hadoop/Spark 中重命名大量文件？

Question

我有一个包含超过 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 不可序列化的错误(因此可能无法在并行操作中使用)

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

Answer 1

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