在 Azure Blob 存储上使用 Delta Lake 时无法重命名“_delta_log”目录中的 json 文件

Question

在单个表上进行并行追加操作时重命名 _delta_log json 文件时面临问题

RM restartUser 类抛出异常后恢复的尝试：java.io.IOException: rename from wasbs://@.blob.core.windows.net/delta_table/_delta_log/.000000000000000000243.json.f0bf5c51-b7ae- 4da8-931e-b1acc21170f5.tmp 到 wasbs://@.blob.core.windows.net/delta_table/_delta_log/00000000000000000243.json 失败。

这里我使用的是 0.5.0 版本的 delta ,请检查下面的堆栈跟踪

at org.apache.hadoop.fs.FileSystem.rename(FileSystem.java:1548)
at org.apache.hadoop.fs.DelegateToFileSystem.renameInternal(DelegateToFileSystem.java:204)
at org.apache.hadoop.fs.AbstractFileSystem.renameInternal(AbstractFileSystem.java:769)
at org.apache.hadoop.fs.AbstractFileSystem.rename(AbstractFileSystem.java:699)
at org.apache.hadoop.fs.FileContext.rename(FileContext.java:1032)
at org.apache.spark.sql.delta.storage.HDFSLogStore.writeInternal(HDFSLogStore.scala:102)
at org.apache.spark.sql.delta.storage.HDFSLogStore.write(HDFSLogStore.scala:78)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply$mcJ$sp(OptimisticTransaction.scala:388)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply(OptimisticTransaction.scala:383)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply(OptimisticTransaction.scala:383)
at org.apache.spark.sql.delta.DeltaLog.lockInterruptibly(DeltaLog.scala:207)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$class.org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit(OptimisticTransaction.scala:382)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$checkAndRetry$1.apply$mcJ$sp(OptimisticTransaction.scala:550)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$checkAndRetry$1.apply(OptimisticTransaction.scala:449)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$checkAndRetry$1.apply(OptimisticTransaction.scala:449)
at com.databricks.spark.util.DatabricksLogging$class.recordOperation(DatabricksLogging.scala:77)
at org.apache.spark.sql.delta.OptimisticTransaction.recordOperation(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.metering.DeltaLogging$class.recordDeltaOperation(DeltaLogging.scala:103)
at org.apache.spark.sql.delta.OptimisticTransaction.recordDeltaOperation(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$class.checkAndRetry(OptimisticTransaction.scala:449)
at org.apache.spark.sql.delta.OptimisticTransaction.checkAndRetry(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply$mcJ$sp(OptimisticTransaction.scala:433)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply(OptimisticTransaction.scala:383)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit$1.apply(OptimisticTransaction.scala:383)
at org.apache.spark.sql.delta.DeltaLog.lockInterruptibly(DeltaLog.scala:207)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$class.org$apache$spark$sql$delta$OptimisticTransactionImpl$$doCommit(OptimisticTransaction.scala:382)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply$mcJ$sp(OptimisticTransaction.scala:293)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply(OptimisticTransaction.scala:252)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply(OptimisticTransaction.scala:252)
at com.databricks.spark.util.DatabricksLogging$class.recordOperation(DatabricksLogging.scala:77)
at org.apache.spark.sql.delta.OptimisticTransaction.recordOperation(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.metering.DeltaLogging$class.recordDeltaOperation(DeltaLogging.scala:103)
at org.apache.spark.sql.delta.OptimisticTransaction.recordDeltaOperation(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.OptimisticTransactionImpl$class.commit(OptimisticTransaction.scala:252)
at org.apache.spark.sql.delta.OptimisticTransaction.commit(OptimisticTransaction.scala:78)
at org.apache.spark.sql.delta.commands.WriteIntoDelta$$anonfun$run$1.apply(WriteIntoDelta.scala:67)
at org.apache.spark.sql.delta.commands.WriteIntoDelta$$anonfun$run$1.apply(WriteIntoDelta.scala:64)
at org.apache.spark.sql.delta.DeltaLog.withNewTransaction(DeltaLog.scala:396)
at org.apache.spark.sql.delta.commands.WriteIntoDelta.run(WriteIntoDelta.scala:64)
at org.apache.spark.sql.delta.sources.DeltaDataSource.createRelation(DeltaDataSource.scala:133)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
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)

Answer 1

堆栈跟踪显示您使用了不正确的 LogStore 实现：HDFSLogStore。 HDFSLogStore 是为Hadoop distributed file system (HDFS) 构建的。

为了在 Delta Lake 中使用 Azure Blob 存储，应设置以下配置。

spark.delta.logStore.class=org.apache.spark.sql.delta.storage.AzureLogStore

有关完整说明，请参阅 https://docs.delta.io/latest/delta-storage.html#azure-blob-storage。