opt all over

Author: jiemotongxue

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

@@ -18,11 +18,13 @@
 
 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;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 /**
@@ -34,20 +36,24 @@
  */
 public class SerializationMetricWrapper implements SerializationSchema<Row> {
 
-    private static final Logger LOG = LoggerFactory.getLogger(SerializationMetricWrapper.class);
-
     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();
@@ -60,6 +66,7 @@ protected void beforeSerialize() {
     }
 
     protected void afterSerialize() {
+        dtNumRecordsOut.inc();
     }
 
     public RuntimeContext getRuntimeContext() {

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

@@ -20,6 +20,8 @@
 
 package com.dtstack.flink.sql.table;
 
+import com.dtstack.flink.sql.format.FormatType;
+
 /**
  * Reason:
  * Date: 2018/6/25
@@ -33,7 +35,7 @@ public abstract class TargetTableInfo extends TableInfo {
 
     public static final String SINK_DATA_TYPE = "sinkdatatype";
 
-    private String sinkDataType = "json";
+    private String sinkDataType = FormatType.JSON.name();
 
     public String getSinkDataType() {
         return sinkDataType;

core/src/main/java/com/dtstack/flink/sql/table/TargetTableInfo.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.sink.kafka;
+
+import com.dtstack.flink.sql.format.FormatType;
+import com.dtstack.flink.sql.format.SerializationMetricWrapper;
+import com.dtstack.flink.sql.sink.kafka.table.KafkaSinkTableInfo;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.formats.avro.AvroRowSerializationSchema;
+import org.apache.flink.formats.csv.CsvRowSerializationSchema;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.types.Row;
+
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * company: www.dtstack.com
+ *
+ * @author: toutian
+ * create: 2019/12/26
+ */
+public abstract class AbstractKafkaProducerFactory {
+
+    public abstract RichSinkFunction<Row> createKafkaProducer(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Row> typeInformation, Properties properties, Optional<FlinkKafkaPartitioner<Row>> partitioner);
+
+    protected SerializationMetricWrapper createSerializationMetricWrapper(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Row> typeInformation) {
+        return new SerializationMetricWrapper(createSerializationSchema(kafkaSinkTableInfo, typeInformation));
+    }
+
+    private SerializationSchema<Row> createSerializationSchema(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Row> typeInformation) {
+        SerializationSchema<Row> serializationSchema = null;
+        if (FormatType.JSON.name().equalsIgnoreCase(kafkaSinkTableInfo.getSinkDataType())) {
+            if (StringUtils.isNotBlank(kafkaSinkTableInfo.getSchemaString())) {
+                serializationSchema = new JsonRowSerializationSchema(kafkaSinkTableInfo.getSchemaString());
+            } else if (typeInformation != null && typeInformation.getArity() != 0) {
+                serializationSchema = new JsonRowSerializationSchema(typeInformation);
+            } else {
+                throw new IllegalArgumentException("sinkDataType:" + FormatType.JSON.name() + " must set schemaString（JSON Schema）or TypeInformation<Row>");
+            }
+        } else if (FormatType.CSV.name().equalsIgnoreCase(kafkaSinkTableInfo.getSinkDataType())) {
+            if (StringUtils.isBlank(kafkaSinkTableInfo.getFieldDelimiter())) {
+                throw new IllegalArgumentException("sinkDataType:" + FormatType.CSV.name() + " must set fieldDelimiter");
+            }
+            final CsvRowSerializationSchema.Builder serSchemaBuilder = new CsvRowSerializationSchema.Builder(typeInformation);
+            serSchemaBuilder.setFieldDelimiter(kafkaSinkTableInfo.getFieldDelimiter().toCharArray()[0]);
+            serializationSchema = serSchemaBuilder.build();
+        } else if (FormatType.AVRO.name().equalsIgnoreCase(kafkaSinkTableInfo.getSinkDataType())) {
+            if (StringUtils.isBlank(kafkaSinkTableInfo.getSchemaString())) {
+                throw new IllegalArgumentException("sinkDataType:" + FormatType.AVRO.name() + " must set schemaString");
+            }
+            serializationSchema = new AvroRowSerializationSchema(kafkaSinkTableInfo.getSchemaString());
+        }
+
+        if (null == serializationSchema) {
+            throw new UnsupportedOperationException("FormatType:" + kafkaSinkTableInfo.getSinkDataType());
+        }
+        return serializationSchema;
+    }
+
+}