通过 Spark 覆盖 HDFS 文件/目录

Question

问题

我有一个保存在 HDFS 中的文件，我想做的就是运行我的 spark 应用程序，计算结果 javaRDD 并使用 saveAsTextFile() 将新的“文件”存储在 HDFS 中。

但是，如果文件已经存在，Spark 的saveAsTextFile() 将不起作用。它不会覆盖它。

我尝试了什么

所以我搜索了一个解决方案，我发现一种可能的方法是在尝试保存新文件之前通过 HDFS API 删除文件。

我添加了代码：

FileSystem hdfs = FileSystem.get(new Configuration());
Path newFolderPath = new Path("hdfs://node1:50050/hdfs/" +filename);

if(hdfs.exists(newFolderPath)){
    System.out.println("EXISTS");
    hdfs.delete(newFolderPath, true);
}

filerdd.saveAsTextFile("/hdfs/" + filename);

当我尝试运行我的 Spark 应用程序时，该文件已被删除，但我收到了 FileNotFoundException。

考虑到当有人试图从路径读取文件并且文件不存在时会发生此异常，这是没有意义的，因为删除文件后，没有代码尝试读取它。

我的部分代码

 JavaRDD<String> filerdd = sc.textFile("/hdfs/" + filename)    // load the file here
 ...
 ...
 // Transformations here
 filerdd = filerdd.map(....);
 ...
 ...

 // Delete old file here
 FileSystem hdfs = FileSystem.get(new Configuration());
 Path newFolderPath = new Path("hdfs://node1:50050/hdfs/" +filename);

 if(hdfs.exists(newFolderPath)){
    System.out.println("EXISTS");
    hdfs.delete(newFolderPath, true);
 }

 // Write new file here
 filerdd.saveAsTextFile("/hdfs/" + filename);

我在这里尝试做最简单的事情，但我不知道为什么这不起作用。也许 filerdd 以某种方式连接到路径？？

Answer 1

问题是您使用相同的路径进行输入和输出。 Spark 的 RDD 会延迟执行。当您调用saveAsTextFile 时它会运行。此时，您已经删除了newFolderPath。所以filerdd会抱怨。

无论如何，您不应该使用相同的路径进行输入和输出。