4

我想更改 Databricks Delta 表的列名。

所以我做了以下事情：

// Read old table data
val old_data_DF = spark.read.format("delta")
.load("dbfs:/mnt/main/sales")

// Created a new DF with a renamed column
val new_data_DF = old_data_DF
      .withColumnRenamed("column_a", "metric1")
      .select("*")

// Dropped and recereated the Delta files location
dbutils.fs.rm("dbfs:/mnt/main/sales", true)
dbutils.fs.mkdirs("dbfs:/mnt/main/sales")

// Trying to write the new DF to the location
new_data_DF.write
.format("delta")
.partitionBy("sale_date_partition")
.save("dbfs:/mnt/main/sales")

在写信给 Delta 时，我在最后一步遇到错误：

java.io.FileNotFoundException: dbfs:/mnt/main/sales/sale_date_partition=2019-04-29/part-00000-769.c000.snappy.parquet
A file referenced in the transaction log cannot be found. This occurs when data has been manually deleted from the file system rather than using the table `DELETE` statement

显然数据已被删除，很可能我错过了上述逻辑中的某些内容。现在唯一包含数据的地方是new_data_DF. 写入类似的位置dbfs:/mnt/main/sales_tmp也会失败

我应该怎么做才能将数据从new_data_DFDelta 位置写入？

4

2 回答 2

5

一般来说，最好避免rm在 Delta 表上使用。Delta 的事务日志在大多数情况下可以防止最终的一致性问题，但是，当您在很短的时间内删除并重新创建表时，不同版本的事务日志可能会忽隐忽现。

相反，我建议使用 Delta 提供的事务原语。例如，要覆盖表中的数据，您可以：

df.write.format("delta").mode("overwrite").save("/delta/events")

如果您的表已经损坏，您可以使用FSCK修复它。

于 2019-05-06T23:34:57.030 回答
3

您可以通过以下方式做到这一点。

// Read old table data
val old_data_DF = spark.read.format("delta")
.load("dbfs:/mnt/main/sales")

// Created a new DF with a renamed column
val new_data_DF = old_data_DF
  .withColumnRenamed("column_a", "metric1")
  .select("*")

// Trying to write the new DF to the location
new_data_DF.write
.format("delta")
.mode("overwrite") // this would overwrite the whole data files
.option("overwriteSchema", "true")  //this is the key line.
.partitionBy("sale_date_partition")
.save("dbfs:/mnt/main/sales")

OverWriteSchema 选项将使用我们在转换过程中更新的最新模式创建新的物理文件。

于 2020-10-20T14:31:52.627 回答