利用Spark往Hive中存儲parquet數據，針對一些復雜數據類型如map、array、struct的處理遇到的問題？
為了更好的說明導致問題的原因、現象以及解決方案，首先看下述示例：
-- 創建存儲格式為parquet的Hive非分區表 CREATE EXTERNAL TABLE `t1`( `id` STRING, `map_col` MAP<STRING, STRING>, `arr_col` ARRAY<STRING>, `struct_col` STRUCT<A:STRING,B:STRING>) STORED AS PARQUET LOCATION '/home/spark/test/tmp/t1'; -- 創建存儲格式為parquet的Hive分區表 CREATE EXTERNAL TABLE `t2`( `id` STRING, `map_col` MAP<STRING, STRING>, `arr_col` ARRAY<STRING>, `struct_col` STRUCT<A:STRING,B:STRING>) PARTITIONED BY (`dt` STRING) STORED AS PARQUET LOCATION '/home/spark/test/tmp/t2';
分別向t1、t2執行insert into（insert overwrite..select也會導致下列問題）語句，列map_col都存儲為空map：
insert into table t1 values(1,map(),array('1,1,1'),named_struct('A','1','B','1'));
insert into table t2 partition(dt='20200101') values(1,map(),array('1,1,1'),named_struct('A','1','B','1'));
t1表正常執行，但對t2執行上述insert語句時，報如下異常：
Caused by: parquet.io.ParquetEncodingException: empty fields are illegal, the field should be ommited completely instead
at parquet.io.MessageColumnIO$MessageColumnIORecordConsumer.endField(MessageColumnIO.java:244)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writeMap(DataWritableWriter.java:241)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writeValue(DataWritableWriter.java:116)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writeGroupFields(DataWritableWriter.java:89)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.write(DataWritableWriter.java:60)
... 23 more
t1和t2從建表看唯一的區別就是t1不是分區表而t2是分區表，僅僅從報錯信息是無法看出表分區產生這種問題的原因，看看源碼是做了哪些不同的處理（這里為了方便，筆者這里直接給出分析這個問題的源碼思路圖）：
t1底層存儲指定的是ParquetFilemat，t2底層存儲指定的是HiveFileFormat。這里主要分析一下存儲空map到t2時，為什么出問題，以及如何處理，看幾個核心的代碼（具體的可以參考上述源碼圖）：
從拋出的異常信息empty fields are illegal，關鍵看empty fields在哪里拋出，做了哪些處理，這要看MessageColumnIO中startField和endField是做了哪些處理：
public void startField(String field, int index) {
try {
if (MessageColumnIO.DEBUG) {
this.log("startField(" + field + ", " + index + ")");
}
this.currentColumnIO = ((GroupColumnIO)this.currentColumnIO).getChild(index);
//MessageColumnIO中，startField方法中首先會將emptyField設置為true
this.emptyField = true;
if (MessageColumnIO.DEBUG) {
this.printState();
}
} catch (RuntimeException var4) {
throw new ParquetEncodingException("error starting field " + field + " at " + index, var4);
}
}
//endField方法中會針對emptyField是否為true來決定是否拋出異常
public void endField(String field, int index) {
if (MessageColumnIO.DEBUG) {
this.log("endField(" + field + ", " + index + ")");
}
this.currentColumnIO = this.currentColumnIO.getParent();
//如果到這里仍為true，則拋異常
if (this.emptyField) {
throw new ParquetEncodingException("empty fields are illegal, the field should be ommited completely instead");
} else {
this.fieldsWritten[this.currentLevel].markWritten(index);
this.r[this.currentLevel] = this.currentLevel == 0 ? 0 : this.r[this.currentLevel - 1];
if (MessageColumnIO.DEBUG) {
this.printState();
}
}
}
針對map做處理的一些源碼：
private void writeMap(final Object value, final MapObjectInspector inspector, final GroupType type) {
// Get the internal map structure (MAP_KEY_VALUE)
GroupType repeatedType = type.getType(0).asGroupType();
recordConsumer.startGroup();
recordConsumer.startField(repeatedType.getName(), 0);
Map<?, ?> mapValues = inspector.getMap(value);
Type keyType = repeatedType.getType(0);
String keyName = keyType.getName();
ObjectInspector keyInspector = inspector.getMapKeyObjectInspector();
Type valuetype = repeatedType.getType(1);
String valueName = valuetype.getName();
ObjectInspector valueInspector = inspector.getMapValueObjectInspector();
for (Map.Entry<?, ?> keyValue : mapValues.entrySet()) {
recordConsumer.startGroup();
if (keyValue != null) {
// write key element
Object keyElement = keyValue.getKey();
//recordConsumer此處對應的是MessageColumnIO中的MessageColumnIORecordConsumer
//查看其中的startField和endField的處理
recordConsumer.startField(keyName, 0);
//查看writeValue中對原始數據類型的處理，如int、boolean、varchar
writeValue(keyElement, keyInspector, keyType);
recordConsumer.endField(keyName, 0);
// write value element
Object valueElement = keyValue.getValue();
if (valueElement != null) {
//同上
recordConsumer.startField(valueName, 1);
writeValue(valueElement, valueInspector, valuetype);
recordConsumer.endField(valueName, 1);
}
}
recordConsumer.endGroup();
}
recordConsumer.endField(repeatedType.getName(), 0);
recordConsumer.endGroup();
}
private void writePrimitive(final Object value, final PrimitiveObjectInspector inspector) {
//value為null，則return
if (value == null) {
return;
}
switch (inspector.getPrimitiveCategory()) {
//PrimitiveCategory為VOID，則return
case VOID:
return;
case DOUBLE:
recordConsumer.addDouble(((DoubleObjectInspector) inspector).get(value));
break;
//下面是對double、boolean、float、byte、int等數據類型做的處理，這里不在貼出
....
可以看到在startFiled中首先對emptyField設置為true，只有在結束時比如endField方法中將emptyField設置為false，才不會拋出上述異常。而存儲字段類型為map時，有幾種情況會導致這種異常的發生，比如map為空或者map的key為null。
這里只是以map為例，對於array、struct都有類似問題，看源碼HiveFileFormat -> DataWritableWriter對這三者處理方式類似。類似的問題，在Hive的issue中https://issues.apache.org/jira/browse/HIVE-11625也有討論。
分析出問題解決就比較簡單了，以存儲map類型字段為例：
1. 如果無法改變建表schema，或者存儲時底層用的就是HiveFileFormat
如果無法確定存儲的map字段是否為空，存儲之前判斷一下map是否為空，可以寫個udf或者用size判斷一下，同時要保證key不能為null
2. 建表時使用Spark的DataSource表
-- 這種方式本質上還是用ParquetFileFormat，並且是內部表，生產中不建議直接使用這種方式 CREATE TABLE `test`( `id` STRING, `map_col` MAP<STRING, STRING>, `arr_col` ARRAY<STRING>, `struct_col` STRUCT<A:STRING,B:STRING>) USING parquet OPTIONS(`serialization.format` '1');
3. 存儲時指定ParquetFileFormat
比如，ds.write.format("parquet").save("/tmp/test")其實像這類問題，相信很多人都遇到過並且解決了。這里是為了給出當遇到問題時，解決的一種思路。不僅要知道如何解決，更要知道發生問題是什么原因導致的、如何避免這種問題、解決了問題是怎么解決的（為什么這種方式能解決，有沒有更優的方法）等。
近期文章：
Spark SQL解析查詢parquet格式Hive表獲取分區字段和查詢條件
關注微信公眾號：大數據學習與分享，獲取更對技術干貨