【原创】大叔问题定位分享(15)spark写parquet数据报错ParquetEncodingException: empty fields are illegal, the field should be ommited completely instead

spark 2.1.1

spark里执行sql报错

insert overwrite table test_parquet_table select * from dummy

报错如下:

org.apache.spark.SparkException: Task failed while writing rows.
at org.apache.spark.sql.hive.SparkHiveDynamicPartitionWriterContainer.writeToFile(hiveWriterContainers.scala:333)
at org.apache.spark.sql.hive.execution.InsertIntoHiveTable$$anonfun$saveAsHiveFile$3.apply(InsertIntoHiveTable.scala:210)
at org.apache.spark.sql.hive.execution.InsertIntoHiveTable$$anonfun$saveAsHiveFile$3.apply(InsertIntoHiveTable.scala:210)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: Parquet record is malformed: empty fields are illegal, the field should be ommited completely instead
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.write(DataWritableWriter.java:64)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriteSupport.write(DataWritableWriteSupport.java:59)
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriteSupport.write(DataWritableWriteSupport.java:31)
at parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:121)
at parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:123)
at parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:42)
at org.apache.hadoop.hive.ql.io.parquet.write.ParquetRecordWriterWrapper.write(ParquetRecordWriterWrapper.java:111)
at org.apache.hadoop.hive.ql.io.parquet.write.ParquetRecordWriterWrapper.write(ParquetRecordWriterWrapper.java:124)
at org.apache.spark.sql.hive.SparkHiveDynamicPartitionWriterContainer.writeToFile(hiveWriterContainers.scala:321)
... 8 more
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)
... 16 more

跟进代码

org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter

    private void writeMap(Object value, MapObjectInspector inspector, GroupType type) {
GroupType repeatedType = type.getType(0).asGroupType();
this.recordConsumer.startGroup();
this.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(Iterator i$ = mapValues.entrySet().iterator(); i$.hasNext(); this.recordConsumer.endGroup()) {
Entry<?, ?> keyValue = (Entry)i$.next();
this.recordConsumer.startGroup();
if (keyValue != null) {
Object keyElement = keyValue.getKey();
this.recordConsumer.startField(keyName, 0);
this.writeValue(keyElement, keyInspector, keyType);
this.recordConsumer.endField(keyName, 0);
Object valueElement = keyValue.getValue();
if (valueElement != null) {
this.recordConsumer.startField(valueName, 1);
this.writeValue(valueElement, valueInspector, valuetype);
this.recordConsumer.endField(valueName, 1);
}
}
} this.recordConsumer.endField(repeatedType.getName(), 0);
this.recordConsumer.endGroup();
} private void writeValue(Object value, ObjectInspector inspector, Type type) {
if (type.isPrimitive()) {
this.checkInspectorCategory(inspector, Category.PRIMITIVE);
this.writePrimitive(value, (PrimitiveObjectInspector)inspector);
} else {
GroupType groupType = type.asGroupType();
OriginalType originalType = type.getOriginalType();
if (originalType != null && originalType.equals(OriginalType.LIST)) {
this.checkInspectorCategory(inspector, Category.LIST);
this.writeArray(value, (ListObjectInspector)inspector, groupType);
} else if (originalType != null && originalType.equals(OriginalType.MAP)) {
this.checkInspectorCategory(inspector, Category.MAP);
this.writeMap(value, (MapObjectInspector)inspector, groupType);
} else {
this.checkInspectorCategory(inspector, Category.STRUCT);
this.writeGroup(value, (StructObjectInspector)inspector, groupType);
}
} } private void writePrimitive(Object value, PrimitiveObjectInspector inspector) {
if (value != null) {
switch(inspector.getPrimitiveCategory()) {
case VOID:
return;
case DOUBLE:
this.recordConsumer.addDouble(((DoubleObjectInspector)inspector).get(value));
break;
case BOOLEAN:
this.recordConsumer.addBoolean(((BooleanObjectInspector)inspector).get(value));
break;
case FLOAT:
this.recordConsumer.addFloat(((FloatObjectInspector)inspector).get(value));
break;
case BYTE:
this.recordConsumer.addInteger(((ByteObjectInspector)inspector).get(value));
break;
case INT:
this.recordConsumer.addInteger(((IntObjectInspector)inspector).get(value));
break;
case LONG:
this.recordConsumer.addLong(((LongObjectInspector)inspector).get(value));
break;
case SHORT:
this.recordConsumer.addInteger(((ShortObjectInspector)inspector).get(value));
break;
case STRING:
String v = ((StringObjectInspector)inspector).getPrimitiveJavaObject(value);
this.recordConsumer.addBinary(Binary.fromString(v));
break;
case CHAR:
String vChar = ((HiveCharObjectInspector)inspector).getPrimitiveJavaObject(value).getStrippedValue();
this.recordConsumer.addBinary(Binary.fromString(vChar));
break;
case VARCHAR:
String vVarchar = ((HiveVarcharObjectInspector)inspector).getPrimitiveJavaObject(value).getValue();
this.recordConsumer.addBinary(Binary.fromString(vVarchar));
break;
case BINARY:
byte[] vBinary = ((BinaryObjectInspector)inspector).getPrimitiveJavaObject(value);
this.recordConsumer.addBinary(Binary.fromByteArray(vBinary));
break;
case TIMESTAMP:
Timestamp ts = ((TimestampObjectInspector)inspector).getPrimitiveJavaObject(value);
this.recordConsumer.addBinary(NanoTimeUtils.getNanoTime(ts, false).toBinary());
break;
case DECIMAL:
HiveDecimal vDecimal = (HiveDecimal)inspector.getPrimitiveJavaObject(value);
DecimalTypeInfo decTypeInfo = (DecimalTypeInfo)inspector.getTypeInfo();
this.recordConsumer.addBinary(this.decimalToBinary(vDecimal, decTypeInfo));
break;
case DATE:
Date vDate = ((DateObjectInspector)inspector).getPrimitiveJavaObject(value);
this.recordConsumer.addInteger(DateWritable.dateToDays(vDate));
break;
default:
throw new IllegalArgumentException("Unsupported primitive data type: " + inspector.getPrimitiveCategory());
} }
}

parquet.io.MessageColumnIO.MessageColumnIORecordConsumer

        public void startField(String field, int index) {
try {
if (MessageColumnIO.DEBUG) {
this.log("startField(" + field + ", " + index + ")");
} this.currentColumnIO = ((GroupColumnIO)this.currentColumnIO).getChild(index);
this.emptyField = true;
if (MessageColumnIO.DEBUG) {
this.printState();
} } catch (RuntimeException var4) {
throw new ParquetEncodingException("error starting field " + field + " at " + index, var4);
}
} public void endField(String field, int index) {
if (MessageColumnIO.DEBUG) {
this.log("endField(" + field + ", " + index + ")");
} this.currentColumnIO = this.currentColumnIO.getParent();
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();
} }
} public void addInteger(int value) {
if (MessageColumnIO.DEBUG) {
this.log("addInt(" + value + ")");
} this.emptyField = false;
this.getColumnWriter().write(value, this.r[this.currentLevel], this.currentColumnIO.getDefinitionLevel());
this.setRepetitionLevel();
if (MessageColumnIO.DEBUG) {
this.printState();
} }

DataWritableWriter报错的关键代码是这几行

                Object keyElement = keyValue.getKey();
this.recordConsumer.startField(keyName, 0);
this.writeValue(keyElement, keyInspector, keyType);
this.recordConsumer.endField(keyName, 0);

代码流程梳理如下:

DataWritableWriter.writeMap

MessageColumnIORecordConsumer.startField

注释:this.emptyField = true;

迭代entry

处理key

Object keyElement = keyValue.getKey();

MessageColumnIORecordConsumer.startField

DataWritableWriter.writeValue

DataWritableWriter.isPrimitive

DataWritableWriter.writePrimitive

1)if (value == null) 或是Void

注释:this.emptyField依旧为true

2)if (value != null) MessageColumnIORecordConsumer.addInteger

注释:this.emptyField = false;

MessageColumnIORecordConsumer.endField

MessageColumnIORecordConsumer.endField

注释:if (this.emptyField) {throw new ParquetEncodingException("empty fields are illegal, the field should be ommited completely instead");}

当map<?,?>或array<?>类型的列插入空集合或者map中存在key为null的情形时,就会触发这个错误,

后来发现官方已经有讨论:https://issues.apache.org/jira/browse/HIVE-11625

要避免这个问题有两种方式:

1 改用hive执行sql;

2 增加udf函数filter_map,当map为空集合时置为null,当map不为空集合时过滤掉map值中所有key为null的entry

spark.udf.register("filter_map", ((map : Map[String, String]) => {if (map != null && !map.isEmpty) map.filter(_._1 != null) else null}))
上一篇:【原创】大叔问题定位分享(5)Kafka客户端报错SocketException: Too many open files 打开的文件过多


下一篇:【原创】大叔问题定位分享(29)datanode启动报错:50020端口被占用