【发布时间】：2022-01-20 10:15:42
【问题描述】：
在 Azure HDInsight 集群上使用 Python，我们使用以下代码将 Spark 数据帧作为 Parquet 文件保存到 Azure Data Lake Storage Gen2：
df.write.parquet('abfs://my_dwh_container@my_storage_account.dfs.core.windows.net/mypath, 'overwrite', compression='snappy')
这通常有效，但是当我们最近升级集群以同时运行更多脚本（大约 10 到 15 个）时，我们始终会在一小部分脚本中遇到以下异常：
Py4JJavaError：调用 o2232.parquet 时出错。 ： java.io.FileNotFoundException：操作失败：“指定路径 不存在。”, 404, PUT, https://my_storage_account.dfs.core.windows.net/mypath/_temporary/0?resource=directory&timeout=90, PathNotFound, "指定的路径不存在。"
我认为所有 Spark 作业和任务实际上都成功了，也是保存表的那个，但是 Python 脚本退出时出现异常。
背景资料
我们使用的是 Spark 2.4.5.4.1.1.2。使用 Scala 版本 2.11.12、OpenJDK 64 位服务器 VM、1.8.0_265、Hadoop 3.1.2.4.1.1.2
堆栈跟踪：
File "/usr/hdp/current/spark2-client/python/pyspark/sql/readwriter.py", line 843, in parquet
df_to_save.write.parquet(blob_path, mode, compression='snappy')
self._jwrite.parquet(path)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
answer, self.gateway_client, self.target_id, self.name)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o2232.parquet.
: java.io.FileNotFoundException: Operation failed: "The specified path does not exist.", 404, PUT, https://my_dwh_container@my_storage_account.dfs.core.windows.net/mypath/_temporary/0?resource=directory&timeout=90, PathNotFound, "The specified path does not exist. RequestId:1870ec49-e01f-0101-72f8-f260fe000000 Time:2021-12-17T03:42:35.8434071Z"
at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.checkException(AzureBlobFileSystem.java:1178)
at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.mkdirs(AzureBlobFileSystem.java:477)
at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:2288)
at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.setupJob(FileOutputCommitter.java:382)
at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.setupJob(HadoopMapReduceCommitProtocol.scala:162)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:139)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:159)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult$lzycompute(commands.scala:104)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult(commands.scala:102)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.doExecute(commands.scala:122)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:676)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:676)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:676)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:285)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:271)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:229)
at org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:566)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
日志：
21/12/17 03:42:02 INFO DAGScheduler [Thread-11]: Job 2 finished: saveAsTable at NativeMethodAccessorImpl.java:0, took 1.120535 s
21/12/17 03:42:02 INFO FileFormatWriter [Thread-11]: Write Job 11fc45a5-d398-4f9a-8350-f928c3722886 committed.
21/12/17 03:42:02 INFO FileFormatWriter [Thread-11]: Finished processing stats for write job 11fc45a5-d398-4f9a-8350-f928c3722886.
(...)
21/12/17 03:42:05 INFO ParquetFileFormat [Thread-11]: Using default output committer for Parquet: org.apache.parquet.hadoop.ParquetOutputCommitter
21/12/17 03:42:05 INFO FileOutputCommitter [Thread-11]: File Output Committer Algorithm version is 2
21/12/17 03:42:05 INFO FileOutputCommitter [Thread-11]: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false, move _temporary folders into Trash: false
21/12/17 03:42:05 INFO SQLHadoopMapReduceCommitProtocol [Thread-11]: Using user defined output committer class org.apache.parquet.hadoop.ParquetOutputCommitter
21/12/17 03:42:05 INFO FileOutputCommitter [Thread-11]: File Output Committer Algorithm version is 2
21/12/17 03:42:05 INFO FileOutputCommitter [Thread-11]: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false, move _temporary folders into Trash: false
21/12/17 03:42:05 INFO SQLHadoopMapReduceCommitProtocol [Thread-11]: Using output committer class org.apache.parquet.hadoop.ParquetOutputCommitter
21/12/17 03:42:28 ERROR ApplicationMaster [Driver]: User application exited with status 1
21/12/17 03:42:28 INFO ApplicationMaster [Driver]: Final app status: FAILED, exitCode: 1, (reason: User application exited with status 1)
这个异常的另一个版本确实发生在 Spark 任务中，然后失败，但 Spark 会自动重新启动失败的任务，并且通常它会成功。在某些情况下，AM 会报告应用程序失败，但我不明白为什么，因为所有作业都成功了。
可能的原因
正如Spark _temporary creation reason 中所见，我希望在完成所有任务之前不会移动 _temporary 目录。
查看堆栈跟踪，它发生在 AzureBlobFileSystem.mkdirs 中，这表明它正在尝试在 _temporary/0 下的某处创建子目录，但找不到 0 目录。我不确定那个时候_temporary 目录是否存在。
相关问题
- https://issues.apache.org/jira/browse/SPARK-2984 听起来确实很相似，但我没有看到任务被重新启动，因为它们需要很长时间，而且这应该在很久以前就已经修复了。不过，我不确定在 Spark UI 中是否可以看到推测执行。
- Saving dataframe to local file system results in empty results 我们没有保存到任何本地文件系统（即使错误消息显示 https，堆栈跟踪显示 AzureBlobFileSystem。
- Spark Dataframe Write to CSV creates _temporary directory file in Standalone Cluster Mode 我们正在使用 HDFS 以及文件输出提交程序 2
- Multiple spark jobs appending parquet data to same base path with partitioning我不认为两个工作在这里使用同一个目录
- https://community.datastax.com/questions/3082/while-writing-to-parquet-file-on-hdfs-throwing-fil.html 我认为这不是权限问题，因为大多数情况下它确实有效。
- Extremely slow S3 write times from EMR/ Spark 据我所知，我们在重命名缓慢方面没有任何问题（反正文件不是很大）。我认为在重命名之前它会失败，所以零重命名提交者在这里没有帮助？
- https://support.huaweicloud.com/intl/en-us/trouble-mrs/mrs_03_0084.html 建议查看hdfs的namenode审计日志，但还没找到。
-
https://github.com/apache/hadoop/blob/b7d2135f6f5cea7cf5d5fc5a2090fc5d8596969e/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java#L355 由于堆栈跟踪显示它在 mkdirs 上失败，我猜
_temporary本身不存在，但我不明白为什么 mkdirs 不创建它。但我不认为 AzureBlobFileSystem 是开源的？ - 我确实找到了 some version of AzureBlobFileSystem.java，但根据堆栈跟踪，它会转到带有
PATH_ALREADY_EXISTS标志的 checkException，这对我来说没有意义。
可以尝试的选项：
- Pyspark dataframe write parquet without deleting /_temporary folder 我们可以尝试的是，首先保存到不同的 HDFS，然后复制最终文件。我不确定它为什么会有所帮助，因为我们已经保存到 HDFS（嗯，它的扩展，ADFS）。
- https://community.cloudera.com/t5/Support-Questions/How-to-change-Spark-temporary-directory-when-writing-data/td-p/237389 我们可以自己尝试使用追加和删除文件。
- Change spark _temporary directory path 使用我们自己的 FileOutputCommitter 对于这个问题来说听起来有点矫枉过正
【问题讨论】：
标签： python apache-spark hadoop hdfs azure-blob-storage