Apache Spark Session: IOException: mkdir of (path) failed.处理

Question

我正在测试Apache Spark 2.0的新版本，通过尝试使用结构化流功能，使用一个非常简单的代码创建一个具有流数据的数据集，然后打印创建的数据集。这是我的代码:

    SparkSession mySession= SparkSession.builder().appName("ProcessData").master("local[*]").config("spark.sql.warehouse.dir","System.getProperty("user.dir")/warehouse").getOrCreate();
    Dataset<Row> measurements=mySession.readStream().format("socket").option("host","localhost").option("port",5556).load();
    StreamingQuery printDataset=measurements.writeStream().format("console").start();
    printDataset.awaitTermination();

问题是我得到一个IOException: mkdir(临时目录)失败。有人能帮我解决这个问题吗?非常感谢。

这是显示的完整错误:

Exception in thread "main" java.io.IOException: mkdir of C:/Users/Manuel%20Mourato/AppData/Local/Temp/temporary-891579db-0442-4e1c-8642-d41c7885ab26/offsets failed
at org.apache.hadoop.fs.FileSystem.primitiveMkdir(FileSystem.java:1065)
at org.apache.hadoop.fs.DelegateToFileSystem.mkdir(DelegateToFileSystem.java:176)
at org.apache.hadoop.fs.FilterFs.mkdir(FilterFs.java:197)
at org.apache.hadoop.fs.FileContext$4.next(FileContext.java:730)
at org.apache.hadoop.fs.FileContext$4.next(FileContext.java:726)
at org.apache.hadoop.fs.FSLinkResolver.resolve(FSLinkResolver.java:90)
at org.apache.hadoop.fs.FileContext.mkdir(FileContext.java:733)
at org.apache.spark.sql.execution.streaming.HDFSMetadataLog$FileContextManager.mkdirs(HDFSMetadataLog.scala:281)
at org.apache.spark.sql.execution.streaming.HDFSMetadataLog.<init>(HDFSMetadataLog.scala:57)
at org.apache.spark.sql.execution.streaming.StreamExecution.<init>(StreamExecution.scala:131)
at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:251)
at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:287)
at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:231)

Answer 1

可以试试这个方法吗?

SparkSession mySession= SparkSession.builder().appName("ProcessData").master("local[*]").config("spark.sql.warehouse.dir",System.getProperty("user.dir") + "/warehouse").getOrCreate();

Answer 2

为什么使用System。getProperty in string?另外，请检查是否存在这样的文件夹，即:

val tempDir = System.getProperty("user.dir");
val path = tempDir + "/warehouse";
SparkSession mySession= SparkSession.builder().appName("ProcessData").master("local[*]").config("spark.sql.warehouse.dir", path).getOrCreate();

也请检查您是否有写权限到这个路径。如果您手动创建仓库目录并设置权限，应该会很好-您将确保一切正常

编辑:明白了!首先，您应该检查AppData/Local/Temp的写权限，因为它是标准的临时目录。

这个错误是由OffsetLog引起的。您可以通过添加option("checkpointLocation", ...)

Answer 3

确保在配置中设置自己的检查点目录，该目录具有写权限，一种方法是在相同的应用程序代码中创建检查点目录，例如

.config("spark.sql.streaming.checkpointLocation", "C:\sparkApp\checkpoints\")