Spark 将输出写回输入目录

Question

我最近遇到一个场景，需要从目录中读取HDFS的输入

 /user/project/jsonFile

并将结果写回同一目录：

 /user/project/jsonFile

读取 jsonFile 后，执行多个连接，并将结果写入 /user/project/jsonFile 使用：

result.write().mode(SaveMode.Overwrite).json("/user/project/jsonFile");

下面是我看到的错误：

[task-result-getter-0]o.a.s.s.TaskSetManager: Lost task 10.0 in stage 7.0 (TID 2508, hddev1db015dxc1.dev.oclc.org, executor 3): java.io.FileNotFoundException: File does not exist: /user/project/jsonFile
    at org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:87)
    at org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:77)
    
It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved.
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.org$apache$spark$sql$execution$datasources$FileScanRDD$$anon$$readCurrentFile(FileScanRDD.scala:127)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:177)

为什么它会抛出java.io.FileNotFoundException: File does not exist? result包含写回 HDFS 的联接输出的数据集，一旦result数据集可用，spark 不应该能够将数据写回到同一输入目录中的 HDFS 吗？

或者

这让我认为一些执行器完成了对输入的连接，并且它们准备将结果写回 HDFS，而一些执行器仍在从同一 HDFS 目录读取数据的过程中，该目录现在被覆盖，导致 FileNotFound。真的吗？

谢谢你的帮助

Answer 1

从同一目录读取和写入时，您正在使用覆盖。一种方法是使用追加而不是覆盖

result.write().mode(SaveMode.Append).json("/user/project/jsonFile");

另一种解决方法是将写入的数据存储在另一个文件夹中，然后将其作为源读取到您的初始位置。

read from source
make your data transformations
write transformed data into tempLocation
read from tempLocation
write into source