RM重启恢复时Azure Blob Storage上Delta Lake重命名_delta_log失败

问题场景回顾 #

你碰到的是Delta Lake 0.5.0版本在Azure Blob Storage上执行并行追加操作，ResourceManager(RM)重启后恢复时的IO异常，核心错误是无法将临时日志文件重命名为正式的Delta日志文件：

java.io.IOException: 无法将临时文件wasbs://<container_name>@.blob.core.windows.net/delta_table/_delta_log/.00000000000000000243.json.f0bf5c51-b7ae-4da8-931e-b1acc21170f5.tmp重命名为正式日志文件wasbs://<container_name>@.blob.core.windows.net/delta_table/_delta_log/00000000000000000243.json

完整堆栈跟踪如下：

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)

核心原因分析 #

我之前处理过类似的Delta + Azure Blob场景，这个问题的根源有两个关键点：

1. 默认LogStore的局限性：Delta 0.5.0默认用HDFSLogStore，这个实现依赖HDFS的原子rename操作，但Azure Blob的rename是非原子的（本质是先复制再删除原文件），无法满足Delta乐观事务的日志一致性要求。
2. RM重启的残留状态：RM重启中断了正在执行的并行追加操作，临时日志文件未被正常清理，恢复时Delta尝试重命名这些文件，加上对象存储的特性限制，最终导致重命名失败。

可行解决方案 #

1. 升级Delta Lake版本（推荐）

Delta从0.6.0版本开始推出了AzureBlobStorageLogStore，专门针对Azure Blob做了适配，用Azure的putIfAbsent原子操作替代rename，彻底解决了对象存储上的日志一致性问题。升级后会自动适配Azure环境，无需额外配置。

2. 手动配置专用LogStore（不升级时的临时方案）

如果暂时无法升级，可在Spark配置中指定使用Azure专用的LogStore：

# Spark配置文件中添加
spark.sql.delta.logStore.class org.apache.spark.sql.delta.storage.AzureBlobStorageLogStore

或者在代码中动态设置：

# PySpark示例
spark.conf.set("spark.sql.delta.logStore.class", "org.apache.spark.sql.delta.storage.AzureBlobStorageLogStore")

3. 清理残留临时文件

手动删除_delta_log目录下的所有.tmp文件，再重新启动恢复操作：

# Azure CLI命令
az storage blob delete-batch --account-name <你的存储账户名> --source <容器名>/delta_table/_delta_log --pattern "*.tmp"

4. 优化并行写操作

调整Spark并行度参数（比如spark.sql.shuffle.partitions），减少并发写操作的数量，降低日志冲突概率。同时确保所有写操作都使用Delta官方API，不要直接修改底层存储文件。

验证步骤 #

1. 升级或配置LogStore后，执行一次简单的追加操作，检查_delta_log目录下是否生成了正常的日志文件，无残留.tmp文件。
2. 模拟RM重启场景，再次执行并行追加操作，确认恢复时不再出现重命名异常。

内容的提问来源于stack exchange，提问作者Shalaj