更新代码

Author: dengdejiushini

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -18,8 +18,18 @@
 
 package com.dtstack.flink.sql.exec;
 
-import ch.qos.logback.classic.Level;
-import ch.qos.logback.classic.LoggerContext;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StreamQueryConfig;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.types.Row;
+
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.config.CalciteConfig;
 import com.dtstack.flink.sql.enums.ClusterMode;
@@ -55,17 +65,6 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.table.api.StreamQueryConfig;
-import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.java.StreamTableEnvironment;
-import org.apache.flink.table.sinks.TableSink;
-import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -110,7 +109,6 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
         String remoteSqlPluginPath = options.getRemoteSqlPluginPath();
         String pluginLoadMode = options.getPluginLoadMode();
         String deployMode = options.getMode();
-        String logLevel = options.getLogLevel();
 
         Preconditions.checkArgument(checkRemoteSqlPluginPath(remoteSqlPluginPath, deployMode, pluginLoadMode),
                 "Non-local mode or shipfile deployment mode, remoteSqlPluginPath is required");
@@ -128,7 +126,6 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
                 .setDeployMode(deployMode)
                 .setConfProp(confProperties)
                 .setJarUrlList(jarURList)
-                .setLogLevel(logLevel)
                 .build();
 
     }
@@ -154,8 +151,6 @@ public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInf
         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
         StreamQueryConfig streamQueryConfig = StreamEnvConfigManager.getStreamQueryConfig(tableEnv, paramsInfo.getConfProp());
 
-        setLogLevel(paramsInfo.getLogLevel());
-
         SqlParser.setLocalSqlPluginRoot(paramsInfo.getLocalSqlPluginPath());
         SqlTree sqlTree = SqlParser.parseSql(paramsInfo.getSql());
 
@@ -348,11 +343,5 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         return env;
     }
 
-    private static void setLogLevel(String level){
-        LoggerContext loggerContext= (LoggerContext) LoggerFactory.getILoggerFactory();
-        //设置全局日志级别
-        ch.qos.logback.classic.Logger logger = loggerContext.getLogger("root");
-        logger.setLevel(Level.toLevel(level, Level.INFO));
-    }
 
 }

core/src/main/java/com/dtstack/flink/sql/exec/ParamsInfo.java

@@ -38,19 +38,17 @@ public class ParamsInfo {
     private String remoteSqlPluginPath;
     private String pluginLoadMode;
     private String deployMode;
-    private String logLevel;
     private Properties confProp;
 
     public ParamsInfo(String sql, String name, List<URL> jarUrlList, String localSqlPluginPath,
-                      String remoteSqlPluginPath, String pluginLoadMode, String deployMode, String logLevel, Properties confProp) {
+                      String remoteSqlPluginPath, String pluginLoadMode, String deployMode, Properties confProp) {
         this.sql = sql;
         this.name = name;
         this.jarUrlList = jarUrlList;
         this.localSqlPluginPath = localSqlPluginPath;
         this.remoteSqlPluginPath = remoteSqlPluginPath;
         this.pluginLoadMode = pluginLoadMode;
         this.deployMode = deployMode;
-        this.logLevel = logLevel;
         this.confProp = confProp;
     }
 
@@ -86,9 +84,6 @@ public Properties getConfProp() {
         return confProp;
     }
 
-    public String getLogLevel() {
-        return logLevel;
-    }
 
     @Override
     public String toString() {
@@ -100,7 +95,6 @@ public String toString() {
                 ", remoteSqlPluginPath='" + remoteSqlPluginPath + '\'' +
                 ", pluginLoadMode='" + pluginLoadMode + '\'' +
                 ", deployMode='" + deployMode + '\'' +
-                ", logLevel='" + logLevel + '\'' +
                 ", confProp=" + confProp +
                 '}';
     }
@@ -160,10 +154,6 @@ public ParamsInfo.Builder setDeployMode(String deployMode) {
             return this;
         }
 
-        public ParamsInfo.Builder setLogLevel(String logLevel) {
-            this.logLevel = logLevel;
-            return this;
-        }
 
         public ParamsInfo.Builder setConfProp(Properties confProp) {
             this.confProp = confProp;
@@ -172,7 +162,7 @@ public ParamsInfo.Builder setConfProp(Properties confProp) {
 
         public ParamsInfo build() {
             return new ParamsInfo(sql, name, jarUrlList, localSqlPluginPath,
-                    remoteSqlPluginPath, pluginLoadMode, deployMode, logLevel, confProp);
+                    remoteSqlPluginPath, pluginLoadMode, deployMode, confProp);
         }
     }
 }

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -69,8 +69,6 @@ public class Options {
     @OptionRequired(description = "plugin load mode, by classpath or shipfile")
     private String pluginLoadMode = EPluginLoadMode.CLASSPATH.name();
 
-    private String logLevel;
-
     public String getMode() {
         return mode;
     }
@@ -175,11 +173,4 @@ public void setPluginLoadMode(String pluginLoadMode) {
         this.pluginLoadMode = pluginLoadMode;
     }
 
-    public String getLogLevel() {
-        return logLevel;
-    }
-
-    public void setLogLevel(String logLevel) {
-        this.logLevel = logLevel;
-    }
 }

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -18,19 +18,12 @@
 
 package com.dtstack.flink.sql.sink.kudu;
 
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
-
-import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
-import org.apache.kudu.client.AsyncKuduClient;
-import org.apache.kudu.client.AsyncKuduSession;
-import org.apache.kudu.client.KuduClient;
-import org.apache.kudu.client.KuduException;
-import org.apache.kudu.client.KuduTable;
-import org.apache.kudu.client.Operation;
-import org.apache.kudu.client.PartialRow;
+import org.apache.kudu.client.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java

@@ -24,6 +24,7 @@
 import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;