add timestamp

Author: todd5167

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

@@ -23,11 +23,14 @@
 
 import com.dtstack.flink.sql.source.AbsDeserialization;
 import com.dtstack.flink.sql.source.kafka.metric.KafkaTopicPartitionLagMetric;
+import com.dtstack.flink.sql.table.TableInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
+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.JsonNodeType;
@@ -42,8 +45,13 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.lang.reflect.Field;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -64,7 +72,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
-    private static int rowLenth = 1000;
+    private static int dirtyDataFrequency = 1000;
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -115,7 +123,7 @@ public Row deserialize(byte[] message) throws IOException {
         try {
             JsonNode root = objectMapper.readTree(message);
 
-            if (numInRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount() % dirtyDataFrequency == 0) {
                 LOG.info(root.toString());
             }
 
@@ -137,17 +145,19 @@ public Row deserialize(byte[] message) throws IOException {
                     }
                 } else {
                     // Read the value as specified type
-                    Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
+
+                    Object value = convert(node, fieldTypes[i]);
                     row.setField(i, value);
                 }
             }
 
             numInResolveRecord.inc();
             return row;
-        } catch (Throwable t) {
+        } catch (Exception e) {
             //add metric of dirty data
-            if (dirtyDataCounter.getCount()%rowLenth == 0){
+            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
                 LOG.info("dirtyData: " + new String(message));
+                LOG.error("" , e);
             }
             dirtyDataCounter.inc();
             return null;
@@ -245,4 +255,29 @@ protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exceptio
     private static String partitionLagMetricName(TopicPartition tp) {
         return tp + ".records-lag";
     }
+
+    private Object convert(JsonNode node, TypeInformation<?> info) {
+        if (info.getTypeClass().equals(Types.BOOLEAN.getTypeClass())) {
+            return node.asBoolean();
+        } else if (info.getTypeClass().equals(Types.STRING.getTypeClass())) {
+            return node.asText();
+        }  else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
+            return Date.valueOf(node.asText());
+        } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
+            // local zone
+            return Time.valueOf(node.asText());
+        } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
+            // local zone
+            return Timestamp.valueOf(node.asText());
+        }  else {
+            // for types that were specified without JSON schema
+            // e.g. POJOs
+            try {
+                return objectMapper.treeToValue(node, info.getTypeClass());
+            } catch (JsonProcessingException e) {
+                throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
+            }
+        }
+    }
+
 }