Spark Delta 协议无法恢复

Question

我正在尝试从增量表中删除几个分区，然后写入它，但是当我这样做时，我收到以下错误：

User class threw exception: java.lang.IllegalStateException: 
The protocol of your Delta table couldn't be recovered while Reconstructing
version: 1. Did you manually delete files in the _delta_log directory?
Set spark.databricks.delta.stateReconstructionValidation.enabled
to "false" to skip validation.

诀窍是我知道没有发生这种手动删除。

步骤：

1. 从我每天向其中插入数据并深度复制的工作增量表开始：

val oldtable = DeltaTable.forPath("s3://oldbucket/oldpath/")
oldtable.clone("s3://newtable/newpath/", false, true)

1. 重新分区数据。我已验证这可行 - 我手动检查，我可以读取/计算此表上的统计信息。

val newtable = DeltaTable.forPath("s3://newtable/newpath/")
newtable
  .toDF
  .drop("bad_partition")
  .repartition($"date")
  .write.format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .partitionBy("date") 
  .save("s3://newtable/newpath/")
newtable.vacuum()

所有这些命令都发生在数据块中。

然后我尝试从 EMR 写入同一张表：

val dTable = DeltaTable.forPath(writePath)
dTable.as("previous")
  .merge(df.repartition(partitionColNames.map(n => col(n)) : _*).as("new"), mergeCondition)
  .whenNotMatched()
  .insertAll()
  .whenMatched()
  .updateAll()
  .execute()

我得到了之前的异常。

我尝试使用FSCK REPAIR TABLE 修复表，但错误仍然存​​在。到处搜索，但在其他任何地方都没有看到这个问题。这里出了什么问题？

Answer 1

你也可以试试VACUUM tablename。以便您删除不再被 Delta 表引用的文件。参考link