add log

yanxi0227 · yanxi0227 · commit 5536ee7b54f8 · 2019-07-27T11:30:23.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
@@ -173,9 +173,7 @@ public static void main(String[] args) throws Exception {
         }
 
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
-            if(LOG.isInfoEnabled()){
-                LOG.info("exe-sql:\n" + result.getExecSql());
-            }
+            LOG.info("exe-sql:\n" + result.getExecSql());
 
             boolean isSide = false;
 
@@ -205,6 +203,7 @@ public static void main(String[] args) throws Exception {
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);
                     }else{
                         tableEnv.sqlUpdate(result.getExecSql());
+                        LOG.info("exec sql: " + result.getExecSql());
                     }
                 }
             }
@@ -287,6 +286,7 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
 
                 Table regTable = tableEnv.fromDataStream(adaptStream, fields);
                 tableEnv.registerTable(tableInfo.getName(), regTable);
+                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,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
                 if(pollSqlNode.getKind() == INSERT){
                     tableEnv.sqlUpdate(pollSqlNode.toString());
+                    LOG.info("exec sql: " + pollSqlNode.toString());
                 }else if(pollSqlNode.getKind() == AS){
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
@@ -531,6 +536,7 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
+                    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/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. */
@@ -113,6 +115,10 @@ public Row deserialize(byte[] message) throws IOException {
             numInBytes.inc(message.length);
 
             JsonNode root = objectMapper.readTree(message);
+
+            if (numInResolveRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
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,7 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        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. */
@@ -112,6 +114,10 @@ public Row deserialize(byte[] message) throws IOException {
             numInBytes.inc(message.length);
 
             JsonNode root = objectMapper.readTree(message);
+
+            if (numInResolveRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
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. */
@@ -115,6 +117,10 @@ public Row deserialize(byte[] message) throws IOException {
             numInBytes.inc(message.length);
 
             JsonNode root = objectMapper.readTree(message);
+
+            if (numInResolveRecord.getCount()%rowLenth == 0){
+                LOG.info(root.toString());
+            }
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
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
@@ -61,6 +61,7 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private PreparedStatement upload;
 
     private int batchCount = 0;
+    private static int rowLenth = 1000;
 
     //index field
     private Map<String, List<String>> realIndexes = Maps.newHashMap();
@@ -140,6 +141,9 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         try {
             if (retract) {
                 insertWrite(row);
+                if (outRecords.getCount()%rowLenth == 0){
+                    LOG.info(row.toString());
+                }
                 outRecords.inc();
             } else {
                 //do nothing
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
`@@ -173,9 +173,7 @@ public static void main(String[] args) throws Exception {`
`173`	`173`	`}`
`174`	`174`
`175`	`175`	`for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {`
`176`		`- if(LOG.isInfoEnabled()){`
`177`		`- LOG.info("exe-sql:\n" + result.getExecSql());`
`178`		`- }`
	`176`	`+ LOG.info("exe-sql:\n" + result.getExecSql());`
`179`	`177`
`180`	`178`	`boolean isSide = false;`
`181`	`179`
`@@ -205,6 +203,7 @@ public static void main(String[] args) throws Exception {`
`205`	`203`	`sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);`
`206`	`204`	`}else{`
`207`	`205`	`tableEnv.sqlUpdate(result.getExecSql());`
	`206`	`+ LOG.info("exec sql: " + result.getExecSql());`
`208`	`207`	`}`
`209`	`208`	`}`
`210`	`209`	`}`
`@@ -287,6 +286,7 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en`
`287`	`286`
`288`	`287`	`Table regTable = tableEnv.fromDataStream(adaptStream, fields);`
`289`	`288`	`tableEnv.registerTable(tableInfo.getName(), regTable);`
	`289`	`+ LOG.info("registe table {} success.", tableInfo.getName());`
`290`	`290`	`registerTableCache.put(tableInfo.getName(), regTable);`
`291`	`291`	`classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));`
`292`	`292`	`} else if (tableInfo instanceof TargetTableInfo) {`