最新kafka版本支持嵌套json和数组类型字段解析

Author: 修竹

README.md

@@ -17,7 +17,7 @@
   2.修复yarnPer模式提交失败的异常。
 
 # 已支持
-  * 源表：kafka 0.9，1.x版本
+  * 源表：kafka 0.9，1.x及以上版本
   * 维表：mysql，SQlServer,oracle,hbase，mongo，redis,cassandra,serversocket
   * 结果表：mysql，SQlServer,oracle,hbase，elasticsearch5.x，mongo，redis,cassandra,console
 

docs/kafkaSource.md

@@ -21,7 +21,7 @@ CREATE TABLE tableName(
 ```
 
 ## 2.支持的版本
-    kafka08,kafka09,kafka10,kafka11    
+    kafka08,kafka09,kafka10,kafka11及以上版本    
  **kafka读取和写入的版本必须一致，否则会有兼容性错误。**
 
 ## 3.表结构定义

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

@@ -30,6 +30,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 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.ArrayNode;
 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;
@@ -157,21 +158,27 @@ public Row deserialize(byte[] message) throws IOException {
 
     public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
-        JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
-        if(node == null){
-            return null;
-        }
-
-        JsonNodeType nodeType = node.getNodeType();
-
-        if (nodeType==JsonNodeType.ARRAY){
-            throw new IllegalStateException("Unsupported  type information  array .") ;
-        }
-
-        return node;
+        return nodeAndJsonNodeMapping.get(nodeMappingKey);
     }
 
     private void parseTree(JsonNode jsonNode, String prefix){
+        if (jsonNode.isArray()) {
+            ArrayNode array = (ArrayNode) jsonNode;
+            for (int i = 0; i < array.size(); i++) {
+                JsonNode child = array.get(i);
+                String nodeKey = getNodeKey(prefix, i);
+
+                if (child.isValueNode()) {
+                    nodeAndJsonNodeMapping.put(nodeKey, child);
+                } else {
+                    if (rowAndFieldMapping.containsValue(nodeKey)) {
+                        nodeAndJsonNodeMapping.put(nodeKey, child);
+                    }
+                    parseTree(child, nodeKey);
+                }
+            }
+            return;
+        }
 
         Iterator<String> iterator = jsonNode.fieldNames();
         while (iterator.hasNext()){
@@ -182,7 +189,7 @@ 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()));
+                parseTree(child, nodeKey);
             }else {
                 parseTree(child, nodeKey);
             }
@@ -197,6 +204,13 @@ private String getNodeKey(String prefix, String nodeName){
         return prefix + "." + nodeName;
     }
 
+    private String getNodeKey(String prefix, int i) {
+        if (Strings.isNullOrEmpty(prefix)) {
+            return "[" + i + "]";
+        }
+        return prefix + "[" + i + "]";
+    }
+
     public void setFetcher(AbstractFetcher<Row, ?> fetcher) {
         this.fetcher = fetcher;
     }

kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -40,7 +40,7 @@ public class KafkaSourceParser extends AbsSourceParser {
 
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
-    private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
+    private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
 
     static {
         keyPatternMap.put(KAFKA_NEST_FIELD_KEY, kafkaNestFieldKeyPattern);