SparkSQL與HivemetastoreParquet轉(zhuǎn)換

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

創(chuàng)新互聯(lián)建站專注于漳浦網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。熱誠為您提供漳浦營銷型網(wǎng)站建設(shè)，漳浦網(wǎng)站制作、漳浦網(wǎng)頁設(shè)計(jì)、漳浦網(wǎng)站官網(wǎng)定制、小程序制作服務(wù)，打造漳浦網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供漳浦網(wǎng)站排名全網(wǎng)營銷落地服務(wù)。

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

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

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

由于上面的原因，在將Hive metastore parquet轉(zhuǎn)化為Spark SQL parquet時，需要兼容處理一下Hive和Parquet的schema，即需要對二者的結(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)在以下兩個方面：

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

（2）只出現(xiàn)在Hive元數(shù)據(jù)里的字段將會被視為nullable，并處理到兼容后的schema中

關(guān)于schema（或者說元數(shù)據(jù)metastore），Spark SQL在處理Parquet表時，同樣為了更好的性能，會緩存Parquet的元數(shù)據(jù)信息。此時，如果我們直接通過Hive或者其他工具對該P(yáng)arquet表進(jìn)行修改導(dǎo)致了元數(shù)據(jù)的變化，那么Spark SQL緩存的元數(shù)據(jù)并不能同步更新，此時需要手動刷新Spark SQL緩存的元數(shù)據(jù)，來確保元數(shù)據(jù)的一致性，方式如下：

// 第一種方式應(yīng)用的比較多 1. sparkSession.catalog.refreshTable(s"${dbName.tableName}") 2. sparkSession.catalog.refreshByPath(s"${path}")

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

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

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

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設(shè)置為true來解決上述的異常問題。

但如果同時設(shè)置spark.sql.hive.convertMetastoreParquet為false時，要注意一些數(shù)據(jù)類型以及精度的處理，比如對于decimal類型的處理。通過一個例子復(fù)原一下當(dā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獲得的臨時表 // 以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ù)時，比如執(zhí)行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)

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

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

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

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

網(wǎng)頁名稱：SparkSQL與HivemetastoreParquet轉(zhuǎn)換
標(biāo)題網(wǎng)址：http://muchs.cn/article22/cjhcjc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供微信小程序、品牌網(wǎng)站建設(shè)、App開發(fā)、定制網(wǎng)站、網(wǎng)站營銷、電子商務(wù)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容