You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2019/09/03 16:27:28 UTC
[GitHub] [nifi] tpalfy commented on a change in pull request #3684:
NIFI-6295: Refactored NiFiRecordSerDe to handle nested complex types
tpalfy commented on a change in pull request #3684: NIFI-6295: Refactored NiFiRecordSerDe to handle nested complex types
URL: https://github.com/apache/nifi/pull/3684#discussion_r320363525
##########
File path: nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/hive/streaming/NiFiRecordSerDe.java
##########
@@ -187,152 +173,166 @@ private Object extractCurrentField(Record record, RecordField field, TypeInfo fi
}
switch (primitiveCategory) {
case BYTE:
- Integer bIntValue = record.getAsInt(fieldName);
- val = bIntValue == null ? null : bIntValue.byteValue();
+ Integer bIntValue = DataTypeUtils.toInteger(fieldValue, field.getFieldName());
+ val = bIntValue.byteValue();
break;
case SHORT:
- Integer sIntValue = record.getAsInt(fieldName);
- val = sIntValue == null ? null : sIntValue.shortValue();
+ Integer sIntValue = DataTypeUtils.toInteger(fieldValue, field.getFieldName());
+ val = sIntValue.shortValue();
break;
case INT:
- val = record.getAsInt(fieldName);
+ val = DataTypeUtils.toInteger(fieldValue, field.getFieldName());
break;
case LONG:
- val = record.getAsLong(fieldName);
+ val = DataTypeUtils.toLong(fieldValue, field.getFieldName());
break;
case BOOLEAN:
- val = record.getAsBoolean(fieldName);
+ val = DataTypeUtils.toBoolean(fieldValue, field.getFieldName());
break;
case FLOAT:
- val = record.getAsFloat(fieldName);
+ val = DataTypeUtils.toFloat(fieldValue, field.getFieldName());
break;
case DOUBLE:
- val = record.getAsDouble(fieldName);
+ val = DataTypeUtils.toDouble(fieldValue, field.getFieldName());
break;
case STRING:
case VARCHAR:
case CHAR:
- val = record.getAsString(fieldName);
+ val = DataTypeUtils.toString(fieldValue, field.getFieldName());
break;
case BINARY:
- Object[] array = record.getAsArray(fieldName);
- if (array == null) {
- return null;
+ final ArrayDataType arrayDataType;
+ if(fieldValue instanceof String) {
+ // Treat this as an array of bytes
+ arrayDataType = (ArrayDataType) RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType());
+ } else {
+ arrayDataType = (ArrayDataType) field.getDataType();
}
+ Object[] array = DataTypeUtils.toArray(fieldValue, field.getFieldName(), arrayDataType.getElementType());
val = AvroTypeUtil.convertByteArray(array).array();
break;
case DATE:
- Date d = record.getAsDate(fieldName, field.getDataType().getFormat());
- if(d != null) {
- org.apache.hadoop.hive.common.type.Date hiveDate = new org.apache.hadoop.hive.common.type.Date();
- hiveDate.setTimeInMillis(d.getTime());
- val = hiveDate;
- } else {
- val = null;
- }
+ Date d = DataTypeUtils.toDate(fieldValue, () -> DataTypeUtils.getDateFormat(field.getDataType().getFormat()), field.getFieldName());
+ org.apache.hadoop.hive.common.type.Date hiveDate = new org.apache.hadoop.hive.common.type.Date();
+ hiveDate.setTimeInMillis(d.getTime());
+ val = hiveDate;
break;
// ORC doesn't currently handle TIMESTAMPLOCALTZ
case TIMESTAMP:
- Timestamp ts = DataTypeUtils.toTimestamp(record.getValue(fieldName), () -> DataTypeUtils.getDateFormat(field.getDataType().getFormat()), fieldName);
- if(ts != null) {
- // Convert to Hive's Timestamp type
- org.apache.hadoop.hive.common.type.Timestamp hivetimestamp = new org.apache.hadoop.hive.common.type.Timestamp();
- hivetimestamp.setTimeInMillis(ts.getTime(), ts.getNanos());
- val = hivetimestamp;
- } else {
- val = null;
- }
+ Timestamp ts = DataTypeUtils.toTimestamp(fieldValue, () -> DataTypeUtils.getDateFormat(field.getDataType().getFormat()), field.getFieldName());;
+ // Convert to Hive's Timestamp type
+ org.apache.hadoop.hive.common.type.Timestamp hivetimestamp = new org.apache.hadoop.hive.common.type.Timestamp();
+ hivetimestamp.setTimeInMillis(ts.getTime(), ts.getNanos());
+ val = hivetimestamp;
break;
case DECIMAL:
- Double value = record.getAsDouble(fieldName);
- val = value == null ? null : HiveDecimal.create(value);
+ if(fieldValue instanceof BigDecimal){
+ val = HiveDecimal.create((BigDecimal) fieldValue);
+ } else if (fieldValue instanceof Double){
+ val = HiveDecimal.create((Double)fieldValue);
+ } else if (fieldValue instanceof Number) {
+ val = HiveDecimal.create(((Number)fieldValue).doubleValue());
+ } else {
+ val = HiveDecimal.create(DataTypeUtils.toDouble(fieldValue, field.getDataType().getFormat()));
+ }
break;
default:
- throw new IllegalArgumentException("Field " + fieldName + " cannot be converted to type: " + primitiveCategory.name());
+ throw new IllegalArgumentException("Field " + field.getFieldName() + " cannot be converted to type: " + primitiveCategory.name());
}
break;
case LIST:
- Object[] value = record.getAsArray(fieldName);
- val = value == null ? null : Arrays.asList(value);
+ Object[] value = (Object[])fieldValue;
+ ListTypeInfo listTypeInfo = (ListTypeInfo)fieldTypeInfo;
+ TypeInfo nestedType = listTypeInfo.getListElementTypeInfo();
+ List<Object> converted = new ArrayList<>(value.length);
+ for(int i=0; i<value.length; i++){
+ converted.add(extractCurrentField(value[i], field, nestedType));
Review comment:
Not sure, but If I understand correctly, `field` in this case has the an array datatype but what we wan't here is to pass a field that has the datatype of the elements in the array. Similar to how the MAP case is handled.
It bothered me a little bit that in fact we want to pass 2 simple things and not really a full `RecordField`: the _name_ of the field and it's _datatype_.
I think a very reasonable approach would be to have this `extractCurrentField` get that 2 arguments instead of the single `RecordField`, like this:
```java
@SuppressWarnings("unchecked")
private Object extractCurrentField(final Object fieldValue, String fieldName, DataType fieldDataType, final TypeInfo fieldTypeInfo) throws SerDeException {
```
and the LIST section could look something like this:
```java
case LIST:
Object[] value = (Object[])fieldValue;
ListTypeInfo listTypeInfo = (ListTypeInfo)fieldTypeInfo;
TypeInfo nestedType = listTypeInfo.getListElementTypeInfo();
List<Object> converted = new ArrayList<>(value.length);
for(int i=0; i<value.length; i++){
converted.add(extractCurrentField(value[i], fieldName + "_" + i, ((ArrayDataType) fieldDataType).getElementType(), nestedType));
}
val = converted;
break;
```
(It would also save us the creation of the key and value `RecordField` objects in the MAP case.)
Also a refactor suggestion: the STRUCT case has very much in common with the `deserialize` method. The common part could be refactored, like this:
```java
@Override
public Object deserialize(Writable writable) throws SerDeException {
ObjectWritable t = (ObjectWritable) writable;
Record record = (Record) t.get();
List<Object> result = deserialize(record, schema);
stats.setRowCount(stats.getRowCount() + 1);
return result;
}
private List<Object> deserialize(Record record, StructTypeInfo schema) throws SerDeException {
List<Object> result = new ArrayList<>(Collections.nCopies(schema.getAllStructFieldNames().size(), null));
try {
RecordSchema recordSchema = record.getSchema();
for (RecordField field : recordSchema.getFields()) {
populateRecord(result, record.getValue(field), field, schema);
}
} catch(SerDeException se) {
log.error("Error [{}] parsing Record [{}].", new Object[]{se.toString(), record}, se);
throw se;
} catch (Exception e) {
log.error("Error [{}] parsing Record [{}].", new Object[]{e.toString(), record}, e);
throw new SerDeException(e);
}
return result;
}
...
case STRUCT:
Record nestedRecord = (Record) fieldValue;
StructTypeInfo s = (StructTypeInfo) fieldTypeInfo;
List<Object> result = deserialize(nestedRecord, s);
val = result;
break;
```
The whole `extractCurrentField` method would look like this:
```java
@SuppressWarnings("unchecked")
private Object extractCurrentField(final Object fieldValue, String fieldName, DataType fieldDataType, final TypeInfo fieldTypeInfo) throws SerDeException {
if(fieldValue == null){
return null;
}
Object val;
switch (fieldTypeInfo.getCategory()) {
case PRIMITIVE:
PrimitiveObjectInspector.PrimitiveCategory primitiveCategory = PrimitiveObjectInspector.PrimitiveCategory.UNKNOWN;
if (fieldTypeInfo instanceof PrimitiveTypeInfo) {
primitiveCategory = ((PrimitiveTypeInfo) fieldTypeInfo).getPrimitiveCategory();
}
switch (primitiveCategory) {
case BYTE:
Integer bIntValue = DataTypeUtils.toInteger(fieldValue, fieldName);
val = bIntValue.byteValue();
break;
case SHORT:
Integer sIntValue = DataTypeUtils.toInteger(fieldValue, fieldName);
val = sIntValue.shortValue();
break;
case INT:
val = DataTypeUtils.toInteger(fieldValue, fieldName);
break;
case LONG:
val = DataTypeUtils.toLong(fieldValue, fieldName);
break;
case BOOLEAN:
val = DataTypeUtils.toBoolean(fieldValue, fieldName);
break;
case FLOAT:
val = DataTypeUtils.toFloat(fieldValue, fieldName);
break;
case DOUBLE:
val = DataTypeUtils.toDouble(fieldValue, fieldName);
break;
case STRING:
case VARCHAR:
case CHAR:
val = DataTypeUtils.toString(fieldValue, fieldName);
break;
case BINARY:
final ArrayDataType arrayDataType;
if(fieldValue instanceof String) {
// Treat this as an array of bytes
arrayDataType = (ArrayDataType) RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType());
} else {
arrayDataType = (ArrayDataType) fieldDataType;
}
Object[] array = DataTypeUtils.toArray(fieldValue, fieldName, arrayDataType.getElementType());
val = AvroTypeUtil.convertByteArray(array).array();
break;
case DATE:
Date d = DataTypeUtils.toDate(fieldValue, () -> DataTypeUtils.getDateFormat(fieldDataType.getFormat()), fieldName);
org.apache.hadoop.hive.common.type.Date hiveDate = new org.apache.hadoop.hive.common.type.Date();
hiveDate.setTimeInMillis(d.getTime());
val = hiveDate;
break;
// ORC doesn't currently handle TIMESTAMPLOCALTZ
case TIMESTAMP:
Timestamp ts = DataTypeUtils.toTimestamp(fieldValue, () -> DataTypeUtils.getDateFormat(fieldDataType.getFormat()), fieldName);;
// Convert to Hive's Timestamp type
org.apache.hadoop.hive.common.type.Timestamp hivetimestamp = new org.apache.hadoop.hive.common.type.Timestamp();
hivetimestamp.setTimeInMillis(ts.getTime(), ts.getNanos());
val = hivetimestamp;
break;
case DECIMAL:
if(fieldValue instanceof BigDecimal){
val = HiveDecimal.create((BigDecimal) fieldValue);
} else if (fieldValue instanceof Double){
val = HiveDecimal.create((Double)fieldValue);
} else if (fieldValue instanceof Number) {
val = HiveDecimal.create(((Number)fieldValue).doubleValue());
} else {
val = HiveDecimal.create(DataTypeUtils.toDouble(fieldValue, fieldDataType.getFormat()));
}
break;
default:
throw new IllegalArgumentException("Field " + fieldName + " cannot be converted to type: " + primitiveCategory.name());
}
break;
case LIST:
Object[] value = (Object[])fieldValue;
ListTypeInfo listTypeInfo = (ListTypeInfo)fieldTypeInfo;
TypeInfo nestedType = listTypeInfo.getListElementTypeInfo();
List<Object> converted = new ArrayList<>(value.length);
for(int i=0; i<value.length; i++){
converted.add(extractCurrentField(value[i], fieldName + "_" + i, ((ArrayDataType) fieldDataType).getElementType(), nestedType));
}
val = converted;
break;
case MAP:
//in nifi all maps are <String,?> so use that
Map<String, Object> valueMap = (Map<String,Object>)fieldValue;
MapTypeInfo mapTypeInfo = (MapTypeInfo)fieldTypeInfo;
Map<Object, Object> convertedMap = new HashMap<>(valueMap.size());
//get a key record field, nifi map keys are always string. synthesize new
//record fields for the map field key and value.
for (Map.Entry<String, Object> entry: valueMap.entrySet()) {
convertedMap.put(
extractCurrentField(entry.getKey(), fieldName + ".key", RecordFieldType.STRING.getDataType(), mapTypeInfo.getMapKeyTypeInfo()),
extractCurrentField(entry.getValue(), fieldName + ".value", ((MapDataType) fieldDataType).getValueType(), mapTypeInfo.getMapValueTypeInfo())
);
}
val = convertedMap;
break;
case STRUCT:
Record nestedRecord = (Record) fieldValue;
StructTypeInfo s = (StructTypeInfo) fieldTypeInfo;
List<Object> result = deserialize(nestedRecord, s);
val = result;
break;
default:
log.error("Unknown type found: " + fieldTypeInfo + "for field of type: " + fieldDataType.toString());
return null;
}
return val;
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services