SparkSQL与Hive查询不一致问题

问题

Spark sql 查询出的数据量与hive不一致，重启spark就没问题，查询结果一致。

或者报错说读取的文件不存在，类似如下的错误

FileReadException: Error while reading file xxx.It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved.

解决方案

spark存在一个机制，为了提高性能，会缓存parquet的元数据信息。当通过hive或其他方式更新了parquet表时，缓存的元数据信息未更新，导致SparkSQL查询不到新插入的数据

那么由于缓存机制，spark 没有去主动感知表底层元数据的变化，查询跟实际数据会不一致。有两种方式：

1. 禁用 spark built-in 优化，设置 spark.sql.hive.convertMetastoreParquet、spark.sql.hive.convertMetastoreOrc = false

2. 每次查询之前 refresh table

spark.sql("""REFRESH TABLE """)
spark.catalog.refreshTable("my_table")
sqlContext.refreshTable("my_table")

1. 禁用 hive metadata 缓存 ：spark.sql.filesourceTableRelationCacheSize 配置为 0

方案 1 过于粗暴，直接禁用优化，不推荐；

方案 2 稍微麻烦一点，对于即时查询类应用不太适合，业务上不太接受；

方案 3 有一丢丢的性能损失，但是更适合；

相关原理

查看官方文档，spark sql 读取 hive 表的时候会做一些性能优化

例如参数 ：spark.sql.hive.convertMetastoreParquet （默认值为 true），spark sql 会使用自己内置的 parquet reader and writer 读写通过 hive 创建的 parquet 格式的表。优化之前是用的 hive 默认的 Hive sede。为了解决数据不一致问题，网上的文章都是建议把这个参数设为 false，但是这样就用不到 spark sql 的优化，不太满意网上这个结果。

spark.sql.hive.convertMetastoreParquet：When set to true, the built-in Parquet reader and writer are used to process parquet tables created by using the HiveQL syntax, instead of Hive serde

进一步研究，开启这项优化后，为了进一步提升性能，spark sql 还会缓存 parquest metadata，如果表在 spark 外部（例如 hive 或者 hdfs）被更新了，需要手动刷新 metadata，不然读取的还是旧数据。

Spark SQL caches Parquet metadata for better performance. When Hive metastore Parquet table conversion is enabled, metadata of those converted tables are also cached. If these tables are updated by Hive or other external tools, you need to refresh them manually to ensure consistent metadata.

这项有优化处理对 orc 同理（参数：spark.sql.hive.convertMetastoreOrc），orc 优化以及默认配置是在 spark 2.4 加入进来的。参考文档：

Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, CREATE TABLE t(id int) STORED AS ORC would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark’s ORC data source table and ORC vectorization would be applied. To set false to spark.sql.hive.convertMetastoreOrc restores the previous behavior.

查看 Cache 在 SessionCatalog.scala 里除初始化创建，缓存的数据是表名对应的 LogicalPlan，也就是整个逻辑执行计划。缓存是使用的 guava cache，缓存大小由 spark.sql.filesourceTableRelationCacheSize 参数配置控制。

private val tableRelationCache: Cache[QualifiedTableName, LogicalPlan] = {
     val cacheSize = conf.tableRelationCacheSize
     CacheBuilder.newBuilder().maximumSize(cacheSize).build[QualifiedTableName, LogicalPlan]()
   }

缓存的使用也是有限制的，截取部分代码，只有当查询的 path，schema，patition 一致才会取缓存，另外只有需要转换的才会走缓存，也就是配置了 spark.sql.hive.convertMetastoreParquet 、spark.sql.hive.convertMetastoreOrc 等的才会走缓存，但是这几个默认值都是 true，所以 parquet orc 类型的都会走缓存。

catalogProxy.getCachedTable(tableIdentifier) match {
   case null => None // Cache miss
   case logical @ LogicalRelation(relation: HadoopFsRelation, _, _, _) =>
     val cachedRelationFileFormatClass = relation.fileFormat.getClass
 ​
     expectedFileFormat match {
       case `cachedRelationFileFormatClass` =>
         // If we have the same paths, same schema, and same partition spec,
         // we will use the cached relation.
         val useCached =
           relation.location.rootPaths.toSet == pathsInMetastore.toSet &&
             logical.schema.sameType(schemaInMetastore) &&
             // We don't support hive bucketed tables. This function `getCached` is only used for
             // converting supported Hive tables to data source tables.
             // 只有 converting 的 hive 表才会用缓存。
             relation.bucketSpec.isEmpty &&
             relation.partitionSchema == partitionSchema.getOrElse(StructType(Nil))
 ​
         if (useCached) {
           Some(logical)

参考文章：

 https://www.jianshu.com/p/1b3e6daeb27d