danny0405 commented on a change in pull request #10060: [FLINK-14546]
[flink-json] Support map type in flink-json
URL: https://github.com/apache/flink/pull/10060#discussion_r341469710
##########
File path:
flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
##########
@@ -242,11 +247,33 @@ private DeserializationRuntimeConverter
wrapIntoNullableConverter(Deserializatio
return
Optional.of(createObjectArrayConverter(((BasicArrayTypeInfo)
typeInfo).getComponentInfo()));
} else if (isPrimitiveByteArray(typeInfo)) {
return Optional.of(createByteArrayConverter());
+ } else if (typeInfo instanceof MapTypeInfo) {
+ MapTypeInfo<?, ?> mapTypeInfo = (MapTypeInfo<?, ?>)
typeInfo;
+ return
Optional.of(createMapConverter(mapTypeInfo.getKeyTypeInfo(),
mapTypeInfo.getValueTypeInfo()));
} else {
return Optional.empty();
}
}
+ private DeserializationRuntimeConverter
createMapConverter(TypeInformation keyType, TypeInformation valueType) {
+ DeserializationRuntimeConverter valueConverter =
createConverter(valueType);
+ DeserializationRuntimeConverter keyConverter =
createConverter(keyType);
+
+ return (mapper, jsonNode) -> {
+ // ObjectNode stores Map<String, JsonNode> internally,
so we can only get String keys.
+ Iterator<String> fieldNames = jsonNode.fieldNames();
+ Map<Object, Object> result = new HashMap<>();
+ while (fieldNames.hasNext()) {
+ String stringKey = fieldNames.next();
+ JsonNode keyNode = TextNode.valueOf(stringKey);
Review comment:
We can use the lambda:
```java
fieldNames.stream().map() ...
```
to construct the result Map.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
