reslove conflict

todd5167 · todd5167 · commit 7f87bd7cbe22 · 2019-07-30T11:42:00.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/Main.java b/core/src/main/java/com/dtstack/flink/sql/Main.java
@@ -205,6 +205,9 @@ public static void main(String[] args) throws Exception {
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);
                     }else{
                         tableEnv.sqlUpdate(result.getExecSql());
+                        if(LOG.isInfoEnabled()){
+                            LOG.info("exec sql: " + result.getExecSql());
+                        }
                     }
                 }
             }
@@ -287,6 +290,9 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
 
                 Table regTable = tableEnv.fromDataStream(adaptStream, fields);
                 tableEnv.registerTable(tableInfo.getName(), regTable);
+                if(LOG.isInfoEnabled()){
+                    LOG.info("registe table {} success.", tableInfo.getName());
+                }
                 registerTableCache.put(tableInfo.getName(), regTable);
                 classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));
             } else if (tableInfo instanceof TargetTableInfo) {
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
@@ -50,6 +50,8 @@
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.*;
 
@@ -64,6 +66,8 @@
 
 public class SideSqlExec {
 
+    private static final Logger LOG = LoggerFactory.getLogger(SideSqlExec.class);
+
     private String localSqlPluginPath = null;
 
     private String tmpFields = null;
@@ -102,6 +106,9 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
                 if(pollSqlNode.getKind() == INSERT){
                     tableEnv.sqlUpdate(pollSqlNode.toString());
+                    if(LOG.isInfoEnabled()){
+                        LOG.info("exec sql: " + pollSqlNode.toString());
+                    }
                 }else if(pollSqlNode.getKind() == AS){
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
@@ -531,6 +538,9 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
+                    if(LOG.isInfoEnabled()){
+                        LOG.info("Register Table {} by {}", aliasInfo.getAlias(), aliasInfo.getName());
+                    }
                     localTableCache.put(aliasInfo.getAlias(), table);
                 } else if (pollSqlNode.getKind() == SELECT){
                     Table table = tableEnv.sqlQuery(pollObj.toString());
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java b/core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java
@@ -101,16 +101,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
         }
 
         String cleanupModeStr = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY);
-        if(cleanupModeStr != null){//设置在cancel job情况下checkpoint是否被保存
-            if("true".equalsIgnoreCase(cleanupModeStr)){
-                env.getCheckpointConfig().enableExternalizedCheckpoints(
-                        CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
-            }else if("false".equalsIgnoreCase(cleanupModeStr)){
-                env.getCheckpointConfig().enableExternalizedCheckpoints(
-                        CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-            }else{
-                throw new RuntimeException("not support value of cleanup mode :" + cleanupModeStr);
-            }
+        if ("true".equalsIgnoreCase(cleanupModeStr)){
+            env.getCheckpointConfig().enableExternalizedCheckpoints(
+                    CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
+        } else if("false".equalsIgnoreCase(cleanupModeStr) || cleanupModeStr == null){
+            env.getCheckpointConfig().enableExternalizedCheckpoints(
+                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        } else{
+            throw new RuntimeException("not support value of cleanup mode :" + cleanupModeStr);
         }
 
         String backendPath = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_DATAURI_KEY);
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java
@@ -64,6 +64,8 @@ public class HbaseOutputFormat extends MetricOutputFormat {
     public final SimpleDateFormat ROWKEY_DATE_FORMAT = new SimpleDateFormat("yyyyMMddHHmmss");
     public final SimpleDateFormat FIELD_DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
+    private static int rowLenth = 1000;
+
     @Override
     public void configure(Configuration parameters) {
         LOG.warn("---configure---");
@@ -133,6 +135,10 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         }
 
         table.put(put);
+
+        if (outRecords.getCount()%rowLenth == 0){
+            LOG.info(record.toString());
+        }
         outRecords.inc();
 
     }
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java
@@ -62,6 +62,8 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
+    private static int rowLenth = 1000;
+
     private final ObjectMapper objectMapper = new ObjectMapper();
 
     /** Type information describing the result type. */
@@ -109,10 +111,14 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
+            JsonNode root = objectMapper.readTree(message);
+            if (numInRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
+
             numInRecord.inc();
             numInBytes.inc(message.length);
 
-            JsonNode root = objectMapper.readTree(message);
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -137,6 +143,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info("dirtyData: " + new String(message));
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -25,6 +25,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -39,6 +41,8 @@
 
 public class KafkaSourceParser extends AbsSourceParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceParser.class);
+
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
     private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)$");
@@ -59,6 +63,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java
@@ -62,6 +62,8 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
+    private static int rowLenth = 1000;
+
     private final ObjectMapper objectMapper = new ObjectMapper();
 
     /** Type information describing the result type. */
@@ -108,10 +110,15 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
+            JsonNode root = objectMapper.readTree(message);
+
+            if (numInRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
+
             numInRecord.inc();
             numInBytes.inc(message.length);
 
-            JsonNode root = objectMapper.readTree(message);
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -136,6 +143,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info("dirtyData: " + new String(message));
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -24,6 +24,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -38,6 +40,8 @@
 
 public class KafkaSourceParser extends AbsSourceParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceParser.class);
+
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
     private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)$");
@@ -63,6 +67,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java
@@ -64,6 +64,8 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
+    private static int rowLenth = 1000;
+
     private final ObjectMapper objectMapper = new ObjectMapper();
 
     /** Type information describing the result type. */
@@ -111,10 +113,15 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
+            JsonNode root = objectMapper.readTree(message);
+
+            if (numInRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
+
             numInRecord.inc();
             numInBytes.inc(message.length);
 
-            JsonNode root = objectMapper.readTree(message);
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -139,6 +146,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info("dirtyData: " + new String(message));
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -24,6 +24,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -38,6 +40,8 @@
 
 public class KafkaSourceParser extends AbsSourceParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceParser.class);
+
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
     private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)$");
@@ -63,6 +67,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override
diff --git a/mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java b/mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java
@@ -63,6 +63,8 @@ public class MongoOutputFormat extends MetricOutputFormat {
 
     private static String PK = "_ID";
 
+    private static int rowLenth = 1000;
+
     public final SimpleDateFormat ROWKEY_DATE_FORMAT = new SimpleDateFormat("yyyyMMddHHmmss");
 
     @Override
@@ -107,6 +109,10 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         } else {
             dbCollection.insertOne(doc);
         }
+
+        if (outRecords.getCount()%rowLenth == 0){
+            LOG.info(record.toString());
+        }
         outRecords.inc();
     }
 
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java
@@ -67,11 +67,10 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
 
     private Connection dbConn;
     private PreparedStatement upload;
-
     private AtomicInteger batchCount = new AtomicInteger(0);
-
     private transient ScheduledThreadPoolExecutor timerService;
 
+
     //index field
     private Map<String, List<String>> realIndexes = Maps.newHashMap();
     //full field
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java
@@ -24,12 +24,15 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import redis.clients.jedis.*;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.*;
 
 public class RedisOutputFormat extends MetricOutputFormat {
+    private static final Logger LOG = LoggerFactory.getLogger(RedisOutputFormat.class);
 
     private String url;
 
@@ -65,6 +68,8 @@ public class RedisOutputFormat extends MetricOutputFormat {
 
     private GenericObjectPoolConfig poolConfig;
 
+    private static int rowLenth = 1000;
+
     private RedisOutputFormat(){
     }
     @Override
@@ -168,6 +173,10 @@ public void writeRecord(Tuple2 record) throws IOException {
             key.append(tableName).append(":").append(perKey).append(":").append(fieldNames[i]);
             jedis.set(key.toString(), row.getField(i).toString());
         }
+
+        if (outRecords.getCount()%rowLenth == 0){
+            LOG.info(record.toString());
+        }
         outRecords.inc();
     }
 

Original file line number	Diff line number	Diff line change
`@@ -205,6 +205,9 @@ public static void main(String[] args) throws Exception {`
`205`	`205`	`sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);`
`206`	`206`	`}else{`
`207`	`207`	`tableEnv.sqlUpdate(result.getExecSql());`
	`208`	`+ if(LOG.isInfoEnabled()){`
	`209`	`+ LOG.info("exec sql: " + result.getExecSql());`
	`210`	`+ }`
`208`	`211`	`}`
`209`	`212`	`}`
`210`	`213`	`}`
`@@ -287,6 +290,9 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en`
`287`	`290`
`288`	`291`	`Table regTable = tableEnv.fromDataStream(adaptStream, fields);`
`289`	`292`	`tableEnv.registerTable(tableInfo.getName(), regTable);`
	`293`	`+ if(LOG.isInfoEnabled()){`
	`294`	`+ LOG.info("registe table {} success.", tableInfo.getName());`
	`295`	`+ }`
`290`	`296`	`registerTableCache.put(tableInfo.getName(), regTable);`
`291`	`297`	`classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));`
`292`	`298`	`} else if (tableInfo instanceof TargetTableInfo) {`