[Fix][udf,core] fix udf can not use 3 languages at the same time (#1162)

* 修复udf不能三种语言代码同时使用

* 添加 python udf yarn-application使用

dlink-app/dlink-app-base/src/main/java/com/dlink/app/flinksql/Submiter.java

@@ -31,6 +31,7 @@ import com.dlink.trans.Operations;
 import com.dlink.utils.SqlUtil;
 
 import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.python.PythonOptions;
 
 import java.io.IOException;
 import java.sql.SQLException;
@@ -52,6 +53,7 @@ import org.slf4j.LoggerFactory;
  * @since 2021/10/27
  **/
 public class Submiter {
+
     private static final Logger logger = LoggerFactory.getLogger(Submiter.class);
 
     private static String getQuerySQL(Integer id) throws SQLException {
@@ -75,7 +77,8 @@ public class Submiter {
         try {
             statement = DBUtil.getOneByID(getQuerySQL(id), config);
         } catch (IOException | SQLException e) {
-            logger.error("{} --> 获取 FlinkSQL 配置异常，ID 为 {}, 连接信息为：{} ,异常信息为：{} ", LocalDateTime.now(), id, config.toString(), e.getMessage(), e);
+            logger.error("{} --> 获取 FlinkSQL 配置异常，ID 为 {}, 连接信息为：{} ,异常信息为：{} ", LocalDateTime.now(), id,
+                    config.toString(), e.getMessage(), e);
         }
         return statement;
     }
@@ -85,7 +88,8 @@ public class Submiter {
         try {
             task = DBUtil.getMapByID(getTaskInfo(id), config);
         } catch (IOException | SQLException e) {
-            logger.error("{} --> 获取 FlinkSQL 配置异常，ID 为 {}, 连接信息为：{} ,异常信息为：{} ", LocalDateTime.now(), id, config.toString(), e.getMessage(), e);
+            logger.error("{} --> 获取 FlinkSQL 配置异常，ID 为 {}, 连接信息为：{} ,异常信息为：{} ", LocalDateTime.now(), id,
+                    config.toString(), e.getMessage(), e);
         }
         return task;
     }
@@ -117,6 +121,7 @@ public class Submiter {
             executorSetting.getConfig().put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
                     executorSetting.getConfig().get(CheckpointingOptions.SAVEPOINT_DIRECTORY.key()) + "/" + uuid);
         }
+        executorSetting.getConfig().put(PythonOptions.PYTHON_FILES.key(), "./python_udf.zip");
         logger.info("作业配置如下： {}", executorSetting);
         Executor executor = Executor.buildAppStreamExecutor(executorSetting);
         List<StatementParam> ddl = new ArrayList<>();
@@ -184,6 +189,6 @@ public class Submiter {
                 logger.error("执行失败, {}", e.getMessage(), e);
             }
         }
-        logger.info("{}任务提交成功",LocalDateTime.now());
+        logger.info("{}任务提交成功", LocalDateTime.now());
     }
 }

dlink-core/src/main/java/com/dlink/job/JobManager.java

@@ -81,6 +81,8 @@ import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
+import cn.hutool.core.util.ArrayUtil;
+
 /**
  * JobManager
  *
@@ -164,7 +166,7 @@ public class JobManager {
         manager.executor.initPyUDF(config.getPyFiles());
 
         if (config.getGatewayConfig() != null) {
-            config.getGatewayConfig().setJarPaths(config.getJarFiles());
+            config.getGatewayConfig().setJarPaths(ArrayUtil.append(config.getJarFiles(),config.getPyFiles()));
         }
         return manager;
     }

dlink-core/src/main/java/com/dlink/udf/UDF.java

@@ -34,7 +34,20 @@ import lombok.Setter;
 @Setter
 @Builder
 public class UDF {
+    /**
+     * 函数名
+     */
+    String name;
+    /**
+     * 类名
+     */
     String className;
+    /**
+     * udf 代码语言
+     */
     FunctionLanguage functionLanguage;
+    /**
+     * udf源代码
+     */
     String code;
 }

dlink-core/src/main/java/com/dlink/utils/UDFUtil.java

@@ -40,7 +40,6 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -53,7 +52,6 @@ import cn.hutool.core.convert.Convert;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.lang.Dict;
 import cn.hutool.core.lang.Opt;
-import cn.hutool.core.lang.PatternPool;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.ReUtil;
 import cn.hutool.core.util.StrUtil;
@@ -77,6 +75,7 @@ public class UDFUtil {
      */
     protected static final Map<String, Integer> UDF_MD5_MAP = new HashMap<>();
     private static final String FUNCTION_REGEX = "function (.*?)'(.*?)'";
+    private static final String FUNCTION_SQL_REGEX = "create\\s+.*function\\s+(.*)\\s+as\\s+'(.*)'(\\s+language (.*))?;";
     private static final String LANGUAGE_REGEX = "language (.*);";
     public static final String PYTHON_UDF_ATTR = "(\\S)\\s+=\\s+ud(?:f|tf|af|taf)";
     public static final String PYTHON_UDF_DEF = "@ud(?:f|tf|af|taf).*\\n+def\\s+(.*)\\(.*\\):";
@@ -117,15 +116,18 @@ public class UDFUtil {
     public static List<UDF> getUDF(String statement) {
         ProcessEntity process = ProcessContextHolder.getProcess();
         process.info("Parse UDF class name:");
-        Pattern pattern = Pattern.compile(FUNCTION_REGEX, Pattern.CASE_INSENSITIVE);
-        Matcher matcher = pattern.matcher(statement);
-        List<UDF> udfList = new ArrayList<>();
-        while (matcher.find()) {
-            UDF udf = UDF.builder().className(matcher.group(2))
-                .functionLanguage(FunctionLanguage.valueOf(Opt.ofNullable(ReUtil.getGroup1(PatternPool.get(LANGUAGE_REGEX, Pattern.CASE_INSENSITIVE), statement)).orElse("JAVA").toUpperCase()))
+        Pattern pattern = Pattern.compile(FUNCTION_SQL_REGEX, Pattern.CASE_INSENSITIVE);
+        List<String> udfSqlList = ReUtil.findAllGroup0(pattern, statement);
+        List<UDF> udfList = udfSqlList.stream().map(sql -> {
+            List<String> groups = ReUtil.getAllGroups(pattern, sql);
+            String udfName = groups.get(1);
+            String className = groups.get(2);
+            return UDF.builder()
+                .name(udfName)
+                .className(className)
+                .functionLanguage(FunctionLanguage.valueOf(Opt.ofNullable(groups.get(4)).orElse("JAVA").toUpperCase()))
                 .build();
-            udfList.add(udf);
-        }
+        }).collect(Collectors.toList());
         List<String> classNameList = udfList.stream().map(UDF::getClassName).collect(Collectors.toList());
         process.info(StringUtils.join(",", classNameList));
         process.info(StrUtil.format("A total of {} UDF have been Parsed.", classNameList.size()));