Merge branch '1.8.0_dev_optimizeFormatWithKafka' into 'v1.8.0_dev'

1.8.0 dev optimize format with kafka

1. 对所有kafka插件的序列化和反序列化做了统筹管理和优化
2. 对kafka的重复代码进行重构
3. 抽象kafka-base作为所有kafka插件的基础

代码的优化改动较大

See merge request !180

Author: zoudaokoulife

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -48,7 +48,7 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends MetricOutputFormat {
+public class CassandraOutputFormat extends DtRichOutputFormat {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.sink.console;
 
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import com.dtstack.flink.sql.sink.console.table.TablePrintUtil;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -37,7 +37,7 @@
  *
  * @author xuqianjin
  */
-public class ConsoleOutputFormat extends MetricOutputFormat {
+public class ConsoleOutputFormat extends DtRichOutputFormat {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConsoleOutputFormat.class);
 

core/pom.xml

@@ -116,6 +116,24 @@
             <version>${flink.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-avro</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-csv</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-json</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
 
     </dependencies>
 

core/src/main/java/com/dtstack/flink/sql/source/AbsDeserialization.java renamed to core/src/main/java/com/dtstack/flink/sql/format/DeserializationMetricWrapper.java

@@ -16,45 +16,51 @@
  * limitations under the License.
  */
 
-package com.dtstack.flink.sql.source;
+package com.dtstack.flink.sql.format;
 
 import com.dtstack.flink.sql.metric.MetricConstant;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+
 /**
- * add metric for source, customer Deserialization which want add metric need to extends this abs class
- * Date: 2018/10/19
- * Company: www.dtstack.com
- *
- * @author xuchao
+ * add metric for source
+ * <p>
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/12/24
  */
+public class DeserializationMetricWrapper extends AbstractDeserializationSchema<Row> {
 
-public abstract class AbsDeserialization<T> extends AbstractDeserializationSchema<T> {
-    private static final Logger LOG = LoggerFactory.getLogger(AbsDeserialization.class);
-
-    private static final long serialVersionUID = 2176278128811784415L;
+    private static final Logger LOG = LoggerFactory.getLogger(DeserializationMetricWrapper.class);
 
     private static int dataPrintFrequency = 1000;
 
-    protected JsonDataParser jsonDataParser;
+    private DeserializationSchema<Row> deserializationSchema;
 
     private transient RuntimeContext runtimeContext;
 
     protected transient Counter dirtyDataCounter;
 
-    //tps ransactions Per Second
+    /**
+     * tps ransactions Per Second
+     */
     protected transient Counter numInRecord;
 
     protected transient Meter numInRate;
 
-    //rps Record Per Second: deserialize data and out record num
+    /**
+     * rps Record Per Second: deserialize data and out record num
+     */
     protected transient Counter numInResolveRecord;
 
     protected transient Meter numInResolveRate;
@@ -63,37 +69,37 @@ public abstract class AbsDeserialization<T> extends AbstractDeserializationSchem
 
     protected transient Meter numInBytesRate;
 
-    public RuntimeContext getRuntimeContext() {
-        return runtimeContext;
-    }
-
-    public void setRuntimeContext(RuntimeContext runtimeContext) {
-        this.runtimeContext = runtimeContext;
+    public DeserializationMetricWrapper(TypeInformation<Row> typeInfo, DeserializationSchema<Row> deserializationSchema) {
+        super(typeInfo);
+        this.deserializationSchema = deserializationSchema;
     }
 
-    public void initMetric(){
+    public void initMetric() {
         dirtyDataCounter = runtimeContext.getMetricGroup().counter(MetricConstant.DT_DIRTY_DATA_COUNTER);
 
         numInRecord = runtimeContext.getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_IN_COUNTER);
-        numInRate = runtimeContext.getMetricGroup().meter( MetricConstant.DT_NUM_RECORDS_IN_RATE, new MeterView(numInRecord, 20));
+        numInRate = runtimeContext.getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_IN_RATE, new MeterView(numInRecord, 20));
 
         numInBytes = runtimeContext.getMetricGroup().counter(MetricConstant.DT_NUM_BYTES_IN_COUNTER);
-        numInBytesRate = runtimeContext.getMetricGroup().meter(MetricConstant.DT_NUM_BYTES_IN_RATE , new MeterView(numInBytes, 20));
+        numInBytesRate = runtimeContext.getMetricGroup().meter(MetricConstant.DT_NUM_BYTES_IN_RATE, new MeterView(numInBytes, 20));
 
         numInResolveRecord = runtimeContext.getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_RESOVED_IN_COUNTER);
         numInResolveRate = runtimeContext.getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_RESOVED_IN_RATE, new MeterView(numInResolveRecord, 20));
     }
 
-    protected T parseSourceData (byte[] message) {
+    @Override
+    public Row deserialize(byte[] message) throws IOException {
         try {
             if (numInRecord.getCount() % dataPrintFrequency == 0) {
                 LOG.info("receive source data:" + new String(message, "UTF-8"));
             }
             numInRecord.inc();
             numInBytes.inc(message.length);
-            Row row = jsonDataParser.parseData(message);
+            beforeDeserialize();
+            Row row = deserializationSchema.deserialize(message);
+            afterDeserialize();
             numInResolveRecord.inc();
-            return (T) row;
+            return row;
         } catch (Exception e) {
             //add metric of dirty data
             if (dirtyDataCounter.getCount() % dataPrintFrequency == 0) {
@@ -105,5 +111,17 @@ protected T parseSourceData (byte[] message) {
         }
     }
 
+    protected void beforeDeserialize() throws IOException {
+    }
 
+    protected void afterDeserialize() throws IOException {
+    }
+
+    public RuntimeContext getRuntimeContext() {
+        return runtimeContext;
+    }
+
+    public void setRuntimeContext(RuntimeContext runtimeContext) {
+        this.runtimeContext = runtimeContext;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/format/FormatType.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.format;
+
+/**
+ * indicate source table input data format type
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/12/24
+ */
+public enum FormatType {
+    //Indicates that the data is in nest json format(default)
+    DT_NEST,
+    //Indicates that the data is in json format
+    JSON,
+    //Indicates that the data is in avro format
+    AVRO,
+    //Indicates that the data is in csv format
+    CSV
+}

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.format;
+
+import com.dtstack.flink.sql.metric.MetricConstant;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.MeterView;
+import org.apache.flink.types.Row;
+
+
+/**
+ * add metric for source
+ * <p>
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/12/24
+ */
+public class SerializationMetricWrapper implements SerializationSchema<Row> {
+
+    private SerializationSchema<Row> serializationSchema;
+
+    private transient RuntimeContext runtimeContext;
+
+    protected transient Counter dtNumRecordsOut;
+
+    protected transient Meter dtNumRecordsOutRate;
+
+
+    public SerializationMetricWrapper(SerializationSchema<Row> serializationSchema) {
+        this.serializationSchema = serializationSchema;
+    }
+
+    public void initMetric() {
+        dtNumRecordsOut = runtimeContext.getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        dtNumRecordsOutRate = runtimeContext.getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(dtNumRecordsOut, 20));
+    }
+
+    @Override
+    public byte[] serialize(Row element) {
+        beforeSerialize();
+        byte[] row = serializationSchema.serialize(element);
+        afterSerialize();
+        return row;
+    }
+
+    protected void beforeSerialize() {
+    }
+
+    protected void afterSerialize() {
+        dtNumRecordsOut.inc();
+    }
+
+    public RuntimeContext getRuntimeContext() {
+        return runtimeContext;
+    }
+
+    public void setRuntimeContext(RuntimeContext runtimeContext) {
+        this.runtimeContext = runtimeContext;
+    }
+
+}

core/src/main/java/com/dtstack/flink/sql/source/JsonDataParser.java renamed to core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -16,11 +16,12 @@
  * limitations under the License.
  */
 
-package com.dtstack.flink.sql.source;
+package com.dtstack.flink.sql.format.dtnest;
 
 import com.dtstack.flink.sql.table.TableInfo;
 import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
+import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -32,7 +33,6 @@
 import org.apache.flink.types.Row;
 
 import java.io.IOException;
-import java.io.Serializable;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -41,13 +41,14 @@
 import java.util.Map;
 
 /**
- *   source data parse to json format
+ * source data parse to json format
  *
  * Date: 2019/12/12
  * Company: www.dtstack.com
+ *
  * @author maqi
  */
-public class JsonDataParser implements Serializable {
+public class DtNestRowDeserializationSchema extends AbstractDeserializationSchema<Row> {
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -58,17 +59,17 @@ public class JsonDataParser implements Serializable {
     private final TypeInformation<?>[] fieldTypes;
     private List<TableInfo.FieldExtraInfo> fieldExtraInfos;
 
-    public JsonDataParser(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos) {
+    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos) {
         this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
         this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
         this.rowAndFieldMapping = rowAndFieldMapping;
         this.fieldExtraInfos = fieldExtraInfos;
     }
 
-
-    public Row parseData(byte[] data) throws IOException {
-        JsonNode root = objectMapper.readTree(data);
-        parseTree(root, null);
+    @Override
+    public Row deserialize(byte[] message) throws IOException {
+        JsonNode root = objectMapper.readTree(message);
+        this.parseTree(root, null);
         Row row = new Row(fieldNames.length);
 
         try {
@@ -176,4 +177,6 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
             }
         }
     }
+
+
 }