Spark存储Parquet数据到Hive时如何对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 PARQUETLOCATION '/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 PARQUETLOCATION '/home/spark/test/tmp/t2';

Caused by: parquet.io.ParquetEncodingException: empty fields are illegal, the field should be ommited completely insteadat 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

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();        }    }}

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等数据类型做的处理，这里不在贴出    ....

-- 这种方式本质上还是用ParquetFileFormat，并且是内部表，生产中不建议直接使用这种方式CREATE TABLE `test`(`id` STRING,`map_col` MAP<STRING, STRING>,`arr_col` ARRAY<STRING>,`struct_col` STRUCT<A:STRING,B:STRING>)USING parquetOPTIONS(`serialization.format` '1');