如何刷新表并同时执行？

Question

我正在使用 Spark Streaming 2.1。我想定期刷新一些缓存表（由 spark 提供的数据源加载，如 parquet、MySQL 或用户定义的数据源）。

1. 如何刷新表格？

   假设我加载了一些表格

   spark.read.format("").load().createTempView("my_table")

   它也被缓存了

   spark.sql("cache table my_table")

   以下代码是否足以刷新表格，以及何时 表下次加载，会自动缓存

   spark.sql("refresh table my_table")

   或者我必须手动执行此操作

   spark.table("my_table").unpersist spark.read.format("").load().createOrReplaceTempView("my_table") spark.sql("cache table my_table")

2. 同时刷新表格是否安全？

   并发我的意思是使用ScheduledThreadPoolExecutor 在主线程之外进行刷新工作。

   如果我在表上调用刷新时 Spark 正在使用缓存表会发生什么？

Answer 1

我在使用 SparkSession 特别是方法表（即spark.table(table_name)）从 hive 中读取表时遇到问题。每次在写完表格并尝试阅读之后 我收到了这个错误：

java.IO.FileNotFoundException ... 基础文件可能已更新。您可以通过在 SQL 中运行“REFRESH TABLE tableName”命令或通过重新创建所涉及的数据集/数据帧来显式地使 Spark 中的缓存无效。

我尝试使用 spark.catalog.refreshTable(table_name) 刷新表，但 sqlContext 也没有用。

我的解决方案写在表格和阅读后使用：

val usersDF = spark.read.load(s"/path/table_name")

一切正常。

这是个问题吗？也许 hdfs 的数据还没有更新？

Answer 2

在 Spark 2.2.0 中，他们引入了在表被 hive 或某些外部工具更新时刷新表元数据的功能。

可以通过API实现，

spark.catalog.refreshTable("my_table")

此 API 将更新该表的元数据以保持一致。