SparkSQL与Hive metastore Parquet转换

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: Spark SQL为了更好的性能,在读写Hive metastore parquet格式的表时,会默认使用自己的Parquet SerDe,而不是采用Hive的SerDe进行序列化和反序列化

本文转载自公众号:大数据学习与分享


Spark SQL为了更好的性能,在读写Hive metastore parquet格式的表时,会默认使用自己的Parquet SerDe,而不是采用Hive的SerDe进行序列化和反序列化。该行为可以通过配置参数spark.sql.hive.convertMetastoreParquet进行控制,默认true。

这里从表schema的处理角度而言,就必须注意Hive和Parquet兼容性,主要有两个区别:

1.Hive是大小写敏感的,但Parquet相反

2.Hive会将所有列视为nullable,但是nullability在parquet里有独特的意义

由于上面的原因,在将Hive metastore parquet转化为Spark SQL parquet时,需要兼容处理一下Hive和Parquet的schema,即需要对二者的结构进行一致化。主要处理规则是:

1.有相同名字的字段必须要有相同的数据类型,忽略nullability。兼容处理的字段应该保持Parquet侧的数据类型,这样就可以处理到nullability类型了(空值问题)

2.兼容处理的schema应只包含在Hive元数据里的schema信息,主要体现在以下两个方面:

(1)只出现在Parquet schema的字段会被忽略

(2)只出现在Hive元数据里的字段将会被视为nullable,并处理到兼容后的schema中

关于schema(或者说元数据metastore),Spark SQL在处理Parquet表时,同样为了更好的性能,会缓存Parquet的元数据信息。此时,如果我们直接通过Hive或者其他工具对该Parquet表进行修改导致了元数据的变化,那么Spark SQL缓存的元数据并不能同步更新,此时需要手动刷新Spark SQL缓存的元数据,来确保元数据的一致性,方式如下:

// 第一种方式应用的比较多
1. sparkSession.catalog.refreshTable(s"${dbName.tableName}")
2. sparkSession.catalog.refreshByPath(s"${path}")

最后说一下最近后台小伙伴在生产中遇到的一个问题,大家如果在业务处理中遇到类似的问题,提供一个思路。

在说问题之前首先了解一个参数spark.sql.parquet.writeLegacyFormat(默认false)的作用:

设置为true时,数据会以Spark1.4和更早的版本的格式写入。比如decimal类型的值会被以Apache Parquet的fixed-length byte array格式写出,该格式是其他系统例如Hive、Impala等使用的。
设置为false时,会使用parquet的新版格式。例如,decimals会以int-based格式写出。如果Spark SQL要以Parquet输出并且结果会被不支持新格式的其他系统使用的话,需要设置为true。
比如,对于decimal数据类型的兼容处理,不设置true时,经常会报类似如下的错误:

Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost, executor driver): parquet.io.ParquetDecodingException: Can not read value at 0 in block -1 in file hdfs://hadoop/data/test_decimal/dt=20200515000000/part-00000-9820eba2-8a40-446d-8c28-37027a1b1f2d-c000.snappy.parquet
  at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:228)
  at parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:201)
  at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:122)
  at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:85)
  at org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat.getRecordReader(MapredParquetInputFormat.java:72)
Caused by: java.lang.UnsupportedOperationException: parquet.column.values.dictionary.PlainValuesDictionary$PlainLongDictionary
  at parquet.column.Dictionary.decodeToBinary(Dictionary.java:44)

此时我们需要将spark.sql.parquet.writeLegacyFormat设置为true来解决上述的异常问题。

但如果同时设置spark.sql.hive.convertMetastoreParquet为false时,要注意一些数据类型以及精度的处理,比如对于decimal类型的处理。通过一个例子复原一下当时的场景:

1.创建Hive外部表testdb.test_decimal,其中字段fee_rate为decimal(10,10)

CREATE EXTERNAL TABLE `testdb`.`test_decimal`(`no` STRING ,
            `fee_rate` DECIMAL(10,10)) 
PARTITIONED BY (`dt` STRING ) 
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH SERDEPROPERTIES ( 'serialization.format' = '1' ) 
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' 
LOCATION 'hdfs://hadoop/data/test_decimal' 
TBLPROPERTIES ( 'transient_lastDdlTime' = '1589160440' ) ;

2.将testdb.item中的数据处理后保存到testdb.test_decimal中

// 这里为了展示方便,直接查询testdb.item中的数据
// 注意: 字段fee_rate的类型为decimal(10,6)
select no, fee_rate from testdb.item  where dt=20190528;
​
// testdb.item中数据示例如下
+-------------------+----------------+
|                 no|       fee_rate|
+-------------------+----------------+
|                  1|        0.000000|
|                  2|        0.000000|
|                  3|        0.000000|
+-------------------+----------------+

3.将testdb.item中的数据保存到testdb.test_decimal中

// tmp是上述查询testdb.item获得的临时表
// 以parquet格式保存到test_decimal的20200529分区中
save overwrite tmp as parquet.`/data/test_decimal/dt=20200529`; 
msck repair TABLE testdb.item;

上述1-3都能成功执行,数据也能保存到testdb.test_decimal中,但是当查询testdb.test_decimal中的数据时,比如执行sql:

select * from testdb.test_decimal where dt = 20200529;

会报如下空指针的异常:

Job aborted due to stage failure: Task 0 in stage 4.0 failed 1 times, most recent failure: Lost task 0.0 in stage 4.0 (TID 4, localhost, executor driver): java.lang.NullPointerException
  at org.apache.spark.sql.hive.HiveShim$.toCatalystDecimal(HiveShim.scala:107)
  at org.apache.spark.sql.hive.HadoopTableReader$$anonfun$14$$anonfun$apply$11.apply(TableReader.scala:415)
  at org.apache.spark.sql.hive.HadoopTableReader$$anonfun$14$$anonfun$apply$11.apply(TableReader.scala:414)
  at org.apache.spark.sql.hive.HadoopTableReader$$anonfun$fillObject$2.apply(TableReader.scala:443)
  at org.apache.spark.sql.hive.HadoopTableReader$$anonfun$fillObject$2.apply(TableReader.scala:434)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)

究其原因是因为按照上述两个参数的配置,testdb.item中fee_rate字段类型为decimal(10,6),数据为0.000000,经过一系列处理0.000000最终会被处理为0,看下边最终导致空指针异常的部分,就会一目了然。

public static BigDecimal enforcePrecisionScale(BigDecimal bd, int maxPrecision, int maxScale) {
        if (bd == null) {
            return null;
        } else {
            bd = trim(bd);
            if (bd.scale() > maxScale) {
                bd = bd.setScale(maxScale, RoundingMode.HALF_UP);
            }
            // testdb.test_decimal中fee_rate的类型decimal(10,10),即precision为10,scale也为10
            // 对应这里即maxPrecision和maxScale分别为10,则maxIntDigits为0
            int maxIntDigits = maxPrecision - maxScale;
            
            // bd对应0。对于0而言,precision为1,scale为0
            // 处理之后 intDigits为1
            int intDigits = bd.precision() - bd.scale();
            return intDigits > maxIntDigits ? null : bd;
        }
}

解决办法也很简单,就是将testdb.test_decimal中的fee_rate数据类型和依赖的表testdb.item中的fee_rate保持完全一致,即也为decimal(10,6)。

这个现象在实际应用环境中经常遇到,通用的解决办法就是将要保存的表中的数据类型与依赖的表(物理表或者临时表)的字段类型保持完全一致。

​----
阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,技术专家直播,问答区近万人Spark技术同学在线提问答疑,只为营造纯粹的Spark氛围,欢迎钉钉扫码加入!
image.png

对开源大数据和感兴趣的同学可以加小编微信(下图二维码,备注“进群”)进入技术交流微信群。
image.png

Apache Spark技术交流社区公众号,微信扫一扫关注

image.png

相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
相关文章
|
2月前
|
SQL 分布式计算 Java
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
62 0
|
4月前
|
SQL 负载均衡 监控
|
4月前
|
SQL 存储 关系型数据库
|
7月前
|
存储 SQL 算法
【Hive】ORC、Parquet等列式存储的优点
【4月更文挑战第14天】【Hive】ORC、Parquet等列式存储的优点
|
SQL 分布式计算 Java
浅析 hive udaf 的正确编写方式- 论姿势的重要性-系列四-如何直接访问metastore service(附源码)
浅析 hive udaf 的正确编写方式- 论姿势的重要性-系列四-如何直接访问metastore service(附源码)
|
SQL 运维 大数据
如何获取大数据平台 CDH 中 hive metastore db 的用户名和密码?
如何获取大数据平台 CDH 中 hive metastore db 的用户名和密码?
如何获取大数据平台 CDH 中 hive metastore db 的用户名和密码?
|
SQL 存储 大数据
关于数据仓库的Hive的Hive架构的MetaStore元数据服务
随着大数据技术的不断发展,数据仓库成为了企业中不可或缺的一部分。而Hive作为一种开源的数据仓库系统,因其易于使用和高效处理等特点,成为了许多企业的首选。然而,对于普通用户来说,直接使用Hive的命令行工具进行操作并不方便。因此,开发者社区中涌现出了大量的Hive GUI工具,其中最为流行的就是Web GUI工具。
320 2
|
SQL 存储 分布式计算
Hive 2.1.1 MetaException(在metastore中找不到消息:版本信息)
Hive 2.1.1 MetaException(在metastore中找不到消息:版本信息)
467 0
|
SQL 存储 分布式计算
SparkSQL 读写_Hive_整合 | 学习笔记
快速学习 SparkSQL 读写_Hive_整合
173 0
SparkSQL 读写_Hive_整合 | 学习笔记
|
SQL 分布式计算 Java
SparkSQL 读写_Hive_写入数据_编码和配置 | 学习笔记
快速学习 SparkSQL 读写_Hive_写入数据_编码和配置
312 0
SparkSQL 读写_Hive_写入数据_编码和配置 | 学习笔记