当你尝试读取一个不存在的文件时,就会出现这个错误。
如果我没理解错的话,我怀疑你在保存新数据帧时(使用.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)。
希望这有所帮助 :)