Merge branch '1.8_code_opt' into 'v1.8.0_dev'

1.8 代码优化以及增加autoWaterMarkinterval参数



See merge request !175

Author: toutian

README.md

@@ -48,6 +48,17 @@
 * Java: JDK8及以上
 * Flink集群: 1.4,1.5,1.8（单机模式不需要安装Flink集群）
 * 操作系统：理论上不限
+* kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数，配置案例:
+```
+fs.hdfs.hadoopconf: /Users/maqi/tmp/hadoopconf/hadoop_250
+security.kerberos.login.use-ticket-cache: true
+security.kerberos.login.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/yanxi.keytab
+security.kerberos.login.principal: yanxi@DTSTACK.COM
+security.kerberos.login.contexts: Client,KafkaClient
+zookeeper.sasl.service-name: zookeeper
+zookeeper.sasl.login-context-name: Client
+
+```
 
 ### 1.3 打包
 

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -19,11 +19,15 @@
 package com.dtstack.flink.sql.classloader;
 
 import com.dtstack.flink.sql.util.PluginUtil;
+import com.dtstack.flink.sql.util.ReflectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.net.URL;
+import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -91,4 +95,26 @@ public static List<URL> getClassPath() {
         }
         return classPaths;
     }
+
+
+
+    public static URLClassLoader loadExtraJar(List<URL> jarURLList, URLClassLoader classLoader)
+            throws  IllegalAccessException, InvocationTargetException {
+        for(URL url : jarURLList){
+            if(url.toString().endsWith(".jar")){
+                urlClassLoaderAddUrl(classLoader, url);
+            }
+        }
+        return classLoader;
+    }
+
+    private static void urlClassLoaderAddUrl(URLClassLoader classLoader, URL url) throws InvocationTargetException, IllegalAccessException {
+        Method method = ReflectionUtils.getDeclaredMethod(classLoader, "addURL", URL.class);
+
+        if (method == null) {
+            throw new RuntimeException("can't not find declared method addURL, curr classLoader is " + classLoader.getClass());
+        }
+        method.setAccessible(true);
+        method.invoke(classLoader, url);
+    }
 }

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -53,9 +53,10 @@ public class ConfigConstrant {
     public static final String SQL_BUFFER_TIMEOUT_MILLIS = "sql.buffer.timeout.millis";
 
     public static final String FLINK_TIME_CHARACTERISTIC_KEY = "time.characteristic";
+    // default 200ms
+    public static final String AUTO_WATERMARK_INTERVAL_KEY = "autoWatermarkInterval";
 
     public static final String SQL_TTL_MINTIME = "sql.ttl.min";
-
     public static final String SQL_TTL_MAXTIME = "sql.ttl.max";
 
     public static final String STATE_BACKEND_KEY = "state.backend";

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -24,7 +24,14 @@
  */
 public enum ClusterMode {
 
-    local(0),standalone(1),yarn(2),yarnPer(3);
+    //run in local
+    local(0),
+    //submit job to standalone cluster
+    standalone(1),
+    //submit job to flink-session which is already run on yarn
+    yarn(2),
+    //submit job to yarn cluster as an application
+    yarnPer(3);
 
     private int type;
 

core/src/main/java/com/dtstack/flink/sql/enums/ClusterMode.java

@@ -0,0 +1,134 @@
+/*
+ * 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.function;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.functions.AggregateFunction;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ *  自定义函数管理类
+ * Reason:
+ * Date: 2017/2/21
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class FunctionManager {
+    private static final Logger logger = LoggerFactory.getLogger(FunctionManager.class);
+
+    /**
+     * TABLE|SCALA|AGGREGATE
+     * 注册UDF到table env
+     */
+    public static void registerUDF(String type, String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader) {
+        if ("SCALA".equalsIgnoreCase(type)) {
+            registerScalaUDF(classPath, funcName, tableEnv, classLoader);
+        } else if ("TABLE".equalsIgnoreCase(type)) {
+            registerTableUDF(classPath, funcName, tableEnv, classLoader);
+        } else if ("AGGREGATE".equalsIgnoreCase(type)) {
+            registerAggregateUDF(classPath, funcName, tableEnv, classLoader);
+        } else {
+            throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA, AGGREGATE)");
+        }
+    }
+
+    /**
+     * 注册自定义方法到env上
+     * @param classPath
+     * @param funcName
+     * @param tableEnv
+     */
+    public static void registerScalaUDF(String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader) {
+        try {
+            ScalarFunction udfFunc = Class.forName(classPath, false, classLoader)
+                    .asSubclass(ScalarFunction.class).newInstance();
+            tableEnv.registerFunction(funcName, udfFunc);
+            logger.info("register scala function:{} success.", funcName);
+        } catch (Exception e) {
+            logger.error("", e);
+            throw new RuntimeException("register UDF exception:", e);
+        }
+    }
+
+    /**
+     * 注册自定义TABLEFFUNC方法到env上
+     *
+     * @param classPath
+     * @param funcName
+     * @param tableEnv
+     */
+    public static void registerTableUDF(String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader) {
+        try {
+            checkStreamTableEnv(tableEnv);
+            TableFunction udtf = Class.forName(classPath, false, classLoader)
+                    .asSubclass(TableFunction.class).newInstance();
+
+            ((StreamTableEnvironment) tableEnv).registerFunction(funcName, udtf);
+            logger.info("register table function:{} success.", funcName);
+        } catch (Exception e) {
+            logger.error("", e);
+            throw new RuntimeException("register Table UDF exception:", e);
+        }
+    }
+
+    private static void checkStreamTableEnv(TableEnvironment tableEnv) {
+        if (!(tableEnv instanceof StreamTableEnvironment)) {
+            throw new RuntimeException("no support tableEnvironment class for " + tableEnv.getClass().getName());
+        }
+    }
+
+    /**
+     * 注册自定义Aggregate FUNC方法到env上
+     *
+     * @param classPath
+     * @param funcName
+     * @param tableEnv
+     */
+    public static void registerAggregateUDF(String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader) {
+        try {
+            checkStreamTableEnv(tableEnv);
+
+            AggregateFunction udaf = Class.forName(classPath, false, classLoader)
+                    .asSubclass(AggregateFunction.class).newInstance();
+            ((StreamTableEnvironment) tableEnv).registerFunction(funcName, udaf);
+            logger.info("register Aggregate function:{} success.", funcName);
+        } catch (Exception e) {
+            logger.error("", e);
+            throw new RuntimeException("register Aggregate UDF exception:", e);
+        }
+    }
+
+
+    public static TypeInformation[] transformTypes(Class[] fieldTypes) {
+        TypeInformation[] types = new TypeInformation[fieldTypes.length];
+        for (int i = 0; i < fieldTypes.length; i++) {
+            types[i] = TypeInformation.of(fieldTypes[i]);
+        }
+
+        return types;
+    }
+}