当你尝试读取一个不存在的文件时，就会出现这个错误。
如果我没理解错的话，我怀疑你在保存新数据帧时（使用.mode("overwrite")），覆盖了所有文件。在此过程中，你试图读取已被删除的文件，会抛出异常 - 这使得表在更新期间不可用。
据我所知，没有直接刷新元数据的方法。
以下是解决此问题的两种方法之一（多种可能性）：
1 - 使用追加模式
如果你只想将新的数据帧附加到旧的数据帧上，那么无需创建临时文件夹并覆盖旧文件。你可以将保存模式从"覆盖"更改为"追加"。这样，你就可以向现有的Parquet文件添加分区，而无需重写现有分区。
df.write
.mode("append")
.parquet("/temp_table")
这是迄今为止最简单的解决方案，不需要读取已经存储的数据。然而，如果您需要更新旧数据（例如：如果您正在进行upsert操作），则此方法无法奏效。对于这种情况，您可以选择第二种方案：
2 - 使用Hive视图
您可以创建Hive表，并使用视图指向最新（并且可用的）表格。
以下是这种方法背后的逻辑示例：
第一部分：
- 如果视图不存在，则创建一个名为_alpha0的新表来存储新数据。
- 创建表之后，我们将创建一个视图，如下所示：select * from _alpha0
第二部分：
- 如果视图存在，则需要查看它指向哪个表格<（table_name>_alphaN）>
- 进行所有想要执行的操作，并将其保存为名为_alpha(N+1)的表格
- 创建表之后，我们将更改视图以便选择* from_alpha(N+1)
以下是代码示例：
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types._
import spark.implicits._
def getCurrentTable(spark: SparkSession, databaseName:String, tableName: String): Option[String] = {
if(spark.catalog.tableExists(s"${databaseName}.${tableName}")) {
val rdd_desc = spark.sql(s"describe formatted ${databaseName}.${tableName}")
.filter("col_name == 'View Text'")
.rdd
if(rdd_desc.isEmpty()) {
None
}
else {
Option(
rdd_desc.first()
.get(1)
.toString
.toLowerCase
.stripPrefix("select * from ")
)
}
}
else
None
}
def saveDataframe(spark: SparkSession, databaseName:String, tableName: String, new_df: DataFrame, rounds: Int = 3): Unit ={
val currentTable = getCurrentTable(spark, databaseName, tableName).getOrElse(s"${databaseName}.${tableName}_alpha${rounds-1}")
val nextAlphaTable = currentTable.replace(s"_alpha${currentTable.last}",s"_alpha${(currentTable.last.toInt + 1) % rounds}")
new_df.write
.mode("overwrite")
.format("parquet")
.option("compression","snappy")
.saveAsTable(nextAlphaTable)
spark.sql(s"create or replace view ${databaseName}.${tableName} as select * from ${nextAlphaTable}")
}
val df = Seq((1,"I"),(2,"am"),(3,"a"),(4,"dataframe")).toDF("id","text")
val new_data = Seq((5,"with"),(6,"new"),(7,"data")).toDF("id","text")
val dbName = "test_db"
val tableName = "alpha_test_table"
println(s"Current table: ${getCurrentTable(spark, dbName, tableName).getOrElse("Table does not exist")}")
println("Saving dataframe")
saveDataframe(spark, dbName, tableName, df)
println("Dataframe saved")
println(s"Current table: ${getCurrentTable(spark, dbName, tableName).getOrElse("Table does not exist")}")
spark.read.table(s"${dbName}.${tableName}").show
val processed_df = df.unionByName(new_data)
println("Saving new dataframe")
saveDataframe(spark, dbName, tableName, processed_df)
println("Dataframe saved")
println(s"Current table: ${getCurrentTable(spark, dbName, tableName).getOrElse("Table does not exist")}")
spark.read.table(s"${dbName}.${tableName}").show
结果：
Current table: Table does not exist
Saving dataframe
Dataframe saved
Current table: test_db.alpha_test_table_alpha0
+---+---------+
| id| text|
+---+---------+
| 3| a|
| 4|dataframe|
| 1| I|
| 2| am|
+---+---------+
Saving new dataframe
Dataframe saved
Current table: test_db.alpha_test_table_alpha1
+---+---------+
| id| text|
+---+---------+
| 3| a|
| 4|dataframe|
| 5| with|
| 6| new|
| 7| data|
| 1| I|
| 2| am|
+---+---------+
通过这样做，您可以保证视图
<table_name> 的一个版本始终可用。这也有一个优点（或不是，取决于您的情况），即维护表格的先前版本。即
<table_name_alpha1> 的上一个版本将是
<table_name_alpha0>。
3 - 奖励
如果升级您的Spark版本是一个选项，可以查看 Delta Lake（最低Spark版本：2.4.2）。
希望这有所帮助 :)
