溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

如何進(jìn)行SparkSQL與Hive metastore Parquet轉(zhuǎn)換的分析

發(fā)布時(shí)間：2021-12-09 17:39:49 來源：億速云 閱讀：242 作者：柒染 欄目：大數(shù)據(jù)

如何進(jìn)行SparkSQL與Hive metastore Parquet轉(zhuǎn)換的分析，相信很多沒有經(jīng)驗(yàn)的人對(duì)此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個(gè)問題。

Spark SQL為了更好的性能，在讀寫Hive metastore parquet格式的表時(shí)，會(huì)默認(rèn)使用自己的Parquet SerDe，而不是采用Hive的SerDe進(jìn)行序列化和反序列化。該行為可以通過配置參數(shù)spark.sql.hive.convertMetastoreParquet進(jìn)行控制，默認(rèn)true。

這里從表schema的處理角度而言，就必須注意Hive和Parquet兼容性，主要有兩個(gè)區(qū)別：

1.Hive是大小寫敏感的，但Parquet相反

2.Hive會(huì)將所有列視為nullable，但是nullability在parquet里有獨(dú)特的意義

由于上面的原因，在將Hive metastore parquet轉(zhuǎn)化為Spark SQL parquet時(shí)，需要兼容處理一下Hive和Parquet的schema，即需要對(duì)二者的結(jié)構(gòu)進(jìn)行一致化。主要處理規(guī)則是：

1.有相同名字的字段必須要有相同的數(shù)據(jù)類型，忽略nullability。兼容處理的字段應(yīng)該保持Parquet側(cè)的數(shù)據(jù)類型，這樣就可以處理到nullability類型了（空值問題）

2.兼容處理的schema應(yīng)只包含在Hive元數(shù)據(jù)里的schema信息，主要體現(xiàn)在以下兩個(gè)方面：

（1）只出現(xiàn)在Parquet schema的字段會(huì)被忽略

（2）只出現(xiàn)在Hive元數(shù)據(jù)里的字段將會(huì)被視為nullable，并處理到兼容后的schema中
關(guān)于schema（或者說元數(shù)據(jù)metastore），Spark SQL在處理Parquet表時(shí)，同樣為了更好的性能，會(huì)緩存Parquet的元數(shù)據(jù)信息。此時(shí)，如果我們直接通過Hive或者其他工具對(duì)該P(yáng)arquet表進(jìn)行修改導(dǎo)致了元數(shù)據(jù)的變化，那么Spark SQL緩存的元數(shù)據(jù)并不能同步更新，此時(shí)需要手動(dòng)刷新Spark SQL緩存的元數(shù)據(jù)，來確保元數(shù)據(jù)的一致性，方式如下：
// 第一種方式應(yīng)用的比較多1. sparkSession.catalog.refreshTable(s"${dbName.tableName}")2. sparkSession.catalog.refreshByPath(s"${path}")

最后說一下最近后臺(tái)小伙伴在生產(chǎn)中遇到的一個(gè)問題，大家如果在業(yè)務(wù)處理中遇到類似的問題，提供一個(gè)思路。

在說問題之前首先了解一個(gè)參數(shù)spark.sql.parquet.writeLegacyFormat（默認(rèn)false）的作用：

設(shè)置為true時(shí)，數(shù)據(jù)會(huì)以Spark1.4和更早的版本的格式寫入。比如decimal類型的值會(huì)被以Apache Parquet的fixed-length byte array格式寫出，該格式是其他系統(tǒng)例如Hive、Impala等使用的。            
設(shè)置為false時(shí)，會(huì)使用parquet的新版格式。例如，decimals會(huì)以int-based格式寫出。如果Spark SQL要以Parquet輸出并且結(jié)果會(huì)被不支持新格式的其他系統(tǒng)使用的話，需要設(shè)置為true。  

比如，對(duì)于decimal數(shù)據(jù)類型的兼容處理，不設(shè)置true時(shí)，經(jīng)常會(huì)報(bào)類似如下的錯(cuò)誤：

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)...

此時(shí)我們需要將spark.sql.parquet.writeLegacyFormat設(shè)置為true來解決上述的異常問題。

但如果同時(shí)設(shè)置spark.sql.hive.convertMetastoreParquet為false時(shí)，要注意一些數(shù)據(jù)類型以及精度的處理，比如對(duì)于decimal類型的處理。通過一個(gè)例子復(fù)原一下當(dāng)時(shí)的場(chǎng)景：

1.創(chuàng)建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中的數(shù)據(jù)處理后保存到testdb.test_decimal中

// 這里為了展示方便，直接查詢testdb.item中的數(shù)據(jù)// 注意: 字段fee_rate的類型為decimal(10,6)select no, fee_rate from testdb.item  where dt=20190528;
// testdb.item中數(shù)據(jù)示例如下+-------------------+----------------+|                 no|       fee_rate|+-------------------+----------------+|                  1|        0.000000||                  2|        0.000000||                  3|        0.000000|+-------------------+----------------+
3.將testdb.item中的數(shù)據(jù)保存到testdb.test_decimal中  
// tmp是上述查詢testdb.item獲得的臨時(shí)表// 以parquet格式保存到test_decimal的20200529分區(qū)中save overwrite tmp as parquet.`/data/test_decimal/dt=20200529`; msck repair TABLE testdb.item;

上述1-3都能成功執(zhí)行，數(shù)據(jù)也能保存到testdb.test_decimal中，但是當(dāng)查詢testdb.test_decimal中的數(shù)據(jù)時(shí)，比如執(zhí)行sql：

select * from testdb.test_decimal where dt = 20200529;
會(huì)報(bào)如下空指針的異常：
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)  ...

究其原因是因?yàn)榘凑丈鲜鰞蓚€(gè)參數(shù)的配置，testdb.item中fee_rate字段類型為decimal(10,6)，數(shù)據(jù)為0.000000，經(jīng)過一系列處理0.000000最終會(huì)被處理為0，看下邊最終導(dǎo)致空指針異常的部分，就會(huì)一目了然。

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            // 對(duì)應(yīng)這里即maxPrecision和maxScale分別為10，則maxIntDigits為0            int maxIntDigits = maxPrecision - maxScale;                        // bd對(duì)應(yīng)0。對(duì)于0而言，precision為1，scale為0            // 處理之后 intDigits為1            int intDigits = bd.precision() - bd.scale();            return intDigits > maxIntDigits ? null : bd;        }}

解決辦法也很簡(jiǎn)單，就是將testdb.test_decimal中的fee_rate數(shù)據(jù)類型和依賴的表testdb.item中的fee_rate保持完全一致，即也為decimal(10,6)。

這個(gè)現(xiàn)象在實(shí)際應(yīng)用環(huán)境中經(jīng)常遇到，通用的解決辦法就是將要保存的表中的數(shù)據(jù)類型與依賴的表（物理表或者臨時(shí)表）的字段類型保持完全一致。

看完上述內(nèi)容，你們掌握如何進(jìn)行SparkSQL與Hive metastore Parquet轉(zhuǎn)換的分析的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！

向AI問一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手