json array 的读取为text

Author: zoudaokoulife

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -31,6 +31,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode;
 import org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread;
 import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
 import org.apache.flink.types.Row;
@@ -174,6 +175,8 @@ private void parseTree(JsonNode jsonNode, String prefix){
 
             if (child.isValueNode()){
                 nodeAndJsonNodeMapping.put(nodeKey, child);
+            } else if(child.isArray()){
+                nodeAndJsonNodeMapping.put(nodeKey, new TextNode(child.toString()));
             }else {
                 parseTree(child, nodeKey);
             }

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -31,6 +31,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode;
 import org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread;
 import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
 import org.apache.flink.types.Row;
@@ -174,6 +175,8 @@ private void parseTree(JsonNode jsonNode, String prefix){
 
             if (child.isValueNode()){
                 nodeAndJsonNodeMapping.put(nodeKey, child);
+            }else if(child.isArray()){
+                nodeAndJsonNodeMapping.put(nodeKey, new TextNode(child.toString()));
             }else {
                 parseTree(child, nodeKey);
             }

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -31,6 +31,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode;
 import org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread;
 import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
 import org.apache.flink.types.Row;
@@ -175,6 +176,8 @@ private void parseTree(JsonNode jsonNode, String prefix){
 
             if (child.isValueNode()){
                 nodeAndJsonNodeMapping.put(nodeKey, child);
+            }else if(child.isArray()){
+                nodeAndJsonNodeMapping.put(nodeKey, new TextNode(child.toString()));
             }else {
                 parseTree(child, nodeKey);
             }