[Feature #1906][git] support python udf develop & auto upload (#1993)

* add py udf * fix log repeat * update sql &console * update sql &console
DataLinkDC · May 30, 2023 · 6c516cb · 6c516cb
1 parent 4473c44
commit 6c516cb
Show file tree

Hide file tree

Showing 27 changed files with 357 additions and 112 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/Dinky.java b/dinky-admin/src/main/java/org/dinky/Dinky.java
@@ -23,8 +23,14 @@
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.freemarker.FreeMarkerAutoConfiguration;
 import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.env.Environment;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
 
+import cn.hutool.system.SystemUtil;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
 /**
  * Dinky 启动器
  *
@@ -33,9 +39,27 @@
 @EnableTransactionManagement
 @SpringBootApplication(exclude = FreeMarkerAutoConfiguration.class)
 @EnableCaching
+@Slf4j
 public class Dinky {
-
+    @SneakyThrows
     public static void main(String[] args) {
-        SpringApplication.run(Dinky.class, args);
+        String ipAddress = SystemUtil.getHostInfo().getAddress();
+        System.setProperty("ipAddr", ipAddress);
+        SpringApplication app = new SpringApplication(Dinky.class);
+        ConfigurableApplicationContext application = app.run(args);
+        Environment env = application.getEnvironment();
+        String port = env.getProperty("server.port");
+        log.info(
+                "\n----------------------------------------------------------\n\t"
+                        + "Application 'Dinky' is running! Access URLs:\n\t"
+                        + "Local: \t\thttp://localhost:{}\n\t"
+                        + "External: \thttp://{}:{}\n\t"
+                        + "Doc: \thttp://{}:{}/doc.html\n"
+                        + "----------------------------------------------------------",
+                port,
+                ipAddress,
+                port,
+                ipAddress,
+                port);
     }
 }
diff --git a/dinky-admin/src/main/java/org/dinky/aop/UdfClassLoaderAspect.java b/dinky-admin/src/main/java/org/dinky/aop/UdfClassLoaderAspect.java
@@ -60,6 +60,7 @@ public Object round(ProceedingJoinPoint proceedingJoinPoint) {
             if (!(e instanceof DinkyException)) {
                 throw new DinkyException(e);
             }
+            e.printStackTrace();
             throw (DinkyException) e;
         } finally {
             if (proceed instanceof JobResult) {

diff --git a/dinky-admin/src/main/java/org/dinky/controller/DownloadController.java b/dinky-admin/src/main/java/org/dinky/controller/DownloadController.java
@@ -23,6 +23,7 @@
 import org.dinky.exception.BusException;
 import org.dinky.function.constant.PathConstant;
 import org.dinky.function.util.ZipWriter;
+import org.dinky.model.FlinkUdfManifest;
 
 import java.io.File;
 import java.io.InputStream;
@@ -40,9 +41,9 @@
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.extra.servlet.ServletUtil;
-import cn.hutool.json.JSONArray;
-import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import lombok.extern.slf4j.Slf4j;
 
 /** @since 0.7.0 */
@@ -61,34 +62,45 @@ public void downloadJavaUDF(@PathVariable Integer taskId, HttpServletResponse re
         if (!depManifestFile.exists()) {
             return;
         }
-        JSONObject jsonObject = new JSONObject(FileUtil.readUtf8String(depManifestFile));
-        JSONArray jars = jsonObject.getJSONArray("jars");
-        List<String> filePath = jars.stream().map(Convert::toStr).collect(Collectors.toList());
+        FlinkUdfManifest flinkUdfManifest =
+                JSONUtil.toBean(FileUtil.readUtf8String(depManifestFile), FlinkUdfManifest.class);
+        List<String> filePath =
+                flinkUdfManifest.getJars().stream()
+                        .map(Convert::toStr)
+                        .collect(Collectors.toList());
+        List<String> pyFilePath =
+                flinkUdfManifest.getPythonFiles().stream()
+                        .map(Convert::toStr)
+                        .collect(Collectors.toList());
         String[] jarNameList =
                 filePath.stream()
                         .map(FileUtil::getName)
                         .map(x -> "jar/" + x)
                         .toArray(String[]::new);
+        String[] pyFileNameList =
+                pyFilePath.stream()
+                        .map(FileUtil::getName)
+                        .map(x -> "py/" + x)
+                        .toArray(String[]::new);
 
         File zipFile = FileUtil.file(udfPackagePath + PathConstant.DEP_ZIP);
         InputStream[] inputStreams =
                 filePath.stream().map(FileUtil::getInputStream).toArray(InputStream[]::new);
+        InputStream[] pyInputStreams =
+                pyFilePath.stream().map(FileUtil::getInputStream).toArray(InputStream[]::new);
         try (ZipWriter zip = new ZipWriter(zipFile, Charset.defaultCharset())) {
-            zip.add(jarNameList, inputStreams);
+            if (ArrayUtil.isNotEmpty(jarNameList)) {
+                zip.add(jarNameList, inputStreams);
+            }
+            if (ArrayUtil.isNotEmpty(pyFileNameList)) {
+                zip.add(pyFileNameList, pyInputStreams);
+            }
             zip.add(depManifestFile.getName(), FileUtil.getInputStream(depManifestFile));
         }
         ServletUtil.write(resp, FileUtil.getInputStream(zipFile));
         FileUtil.del(zipFile);
     }
 
-    @GetMapping("downloadPythonUDF/{taskId}")
-    public void downloadPythonUDF(@PathVariable Integer taskId, HttpServletResponse resp) {
-        ServletUtil.write(
-                resp,
-                FileUtil.file(
-                        PathConstant.getUdfPackagePath(taskId) + PathConstant.UDF_PYTHON_NAME));
-    }
-
     /**
      * 提供docker通过http下载dinky-app.jar
      *

diff --git a/dinky-admin/src/main/java/org/dinky/controller/GitController.java b/dinky-admin/src/main/java/org/dinky/controller/GitController.java
@@ -71,7 +71,7 @@ public class GitController {
      * @return {@link Result} of {@link Void}
      */
     @PutMapping("/saveOrUpdate")
-    public Result<Void> saveOrUpdate(@Validated @RequestBody GitProject gitProject) {
+    public Result<Void> saveOrUpdate(@Validated @RequestBody GitProjectDTO gitProject) {
         gitProjectService.saveOrUpdate(gitProject);
         GitRepository gitRepository =
                 new GitRepository(BeanUtil.copyProperties(gitProject, GitProjectDTO.class));

diff --git a/dinky-admin/src/main/java/org/dinky/dto/GitProjectDTO.java b/dinky-admin/src/main/java/org/dinky/dto/GitProjectDTO.java
@@ -64,4 +64,6 @@ public class GitProjectDTO {
 
     /** */
     @NotNull private Boolean enabled;
+
+    private Integer orderLine;
 }
diff --git a/dinky-admin/src/main/java/org/dinky/init/SystemInit.java b/dinky-admin/src/main/java/org/dinky/init/SystemInit.java
@@ -31,6 +31,7 @@
 import org.dinky.scheduler.config.DolphinSchedulerProperties;
 import org.dinky.scheduler.exception.SchedulerException;
 import org.dinky.scheduler.model.Project;
+import org.dinky.service.GitProjectService;
 import org.dinky.service.JobInstanceService;
 import org.dinky.service.SysConfigService;
 import org.dinky.service.TaskService;
@@ -67,6 +68,7 @@ public class SystemInit implements ApplicationRunner {
     private final TaskService taskService;
     private final TenantService tenantService;
     private final DolphinSchedulerProperties dolphinSchedulerProperties;
+    private final GitProjectService gitProjectService;
     private static Project project;
 
     @Override
@@ -125,6 +127,7 @@ public void registerUDF() {
                 taskService.getAllUDF().stream()
                         .map(UDFUtils::taskToUDF)
                         .collect(Collectors.toList()));
+        UdfCodePool.updateGitPool(gitProjectService.getGitPool());
         TenantContextHolder.set(null);
     }
 }
diff --git a/dinky-admin/src/main/java/org/dinky/model/GitProject.java b/dinky-admin/src/main/java/org/dinky/model/GitProject.java
@@ -92,6 +92,9 @@ public class GitProject extends SuperEntity<GitProject> {
     @TableField(value = "udf_class_map_list")
     private String udfClassMapList;
 
+    @TableField(value = "order_line")
+    private Integer orderLine;
+
     @TableField(exist = false)
     private static final long serialVersionUID = 1L;
 

diff --git a/dinky-admin/src/main/java/org/dinky/service/GitProjectService.java b/dinky-admin/src/main/java/org/dinky/service/GitProjectService.java
@@ -25,6 +25,7 @@
 import org.dinky.model.GitProject;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * @author ZackYoung
@@ -38,6 +39,8 @@ public interface GitProjectService extends ISuperService<GitProject> {
      */
     void saveOrUpdate(GitProjectDTO gitProjectDTO);
 
+    Map<String, String> getGitPool();
+
     /**
      * 更新状态
      *

diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/GitProjectServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/GitProjectServiceImpl.java
@@ -20,9 +20,11 @@
 package org.dinky.service.impl;
 
 import org.dinky.db.service.impl.SuperServiceImpl;
+import org.dinky.dto.GitAnalysisJarDTO;
 import org.dinky.dto.GitProjectDTO;
 import org.dinky.dto.JarClassesDTO;
 import org.dinky.dto.TreeNodeDTO;
+import org.dinky.function.pool.UdfCodePool;
 import org.dinky.mapper.GitProjectMapper;
 import org.dinky.model.GitProject;
 import org.dinky.process.exception.DinkyException;
@@ -34,7 +36,9 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -43,6 +47,8 @@
 
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.lang.Opt;
+import cn.hutool.core.thread.ThreadUtil;
 import cn.hutool.json.JSONUtil;
 
 /**
@@ -66,9 +72,55 @@ public void saveOrUpdate(GitProjectDTO gitProjectDTO) {
         }
 
         GitProject gitProject = BeanUtil.toBean(gitProjectDTO, GitProject.class);
+        if (gitProject.getOrderLine() == null) {
+            Integer maxOrderLine =
+                    Opt.ofNullable(
+                                    baseMapper
+                                            .selectOne(
+                                                    new LambdaQueryWrapper<GitProject>()
+                                                            .orderByAsc(GitProject::getOrderLine)
+                                                            .last(" limit 1"))
+                                            .getOrderLine())
+                            .orElse(999);
+            gitProject.setOrderLine(maxOrderLine + 1);
+        }
         BeanUtil.copyProperties(gitProjectDTO, gitProject);
 
         gitProject.insertOrUpdate();
+
+        ThreadUtil.execAsync(() -> UdfCodePool.updateGitPool(getGitPool()));
+    }
+
+    @Override
+    public Map<String, String> getGitPool() {
+        List<GitProject> list = list();
+        Map<String, String> gitPool = new LinkedHashMap<>();
+        Opt.ofEmptyAble(list)
+                .ifPresent(
+                        l -> {
+                            for (GitProject gitProject : list) {
+                                List<GitAnalysisJarDTO> gitAnalysisJarList =
+                                        JSONUtil.toList(
+                                                gitProject.getUdfClassMapList(),
+                                                GitAnalysisJarDTO.class);
+                                for (GitAnalysisJarDTO analysisJarDTO : gitAnalysisJarList) {
+                                    analysisJarDTO
+                                            .getClassList()
+                                            .forEach(
+                                                    udf -> {
+                                                        gitPool.computeIfAbsent(
+                                                                udf,
+                                                                k -> analysisJarDTO.getJarPath());
+                                                    });
+                                }
+                            }
+                        });
+        return gitPool;
+    }
+
+    @Override
+    public List<GitProject> list() {
+        return list(new LambdaQueryWrapper<GitProject>().orderByAsc(GitProject::getOrderLine));
     }
 
     @Override
@@ -87,7 +139,8 @@ public void updateState(Integer id) {
     @Override
     public List<TreeNodeDTO> getProjectCode(Integer id) {
         GitProject gitProject = getById(id);
-        File projectDir = GitRepository.getProjectDir(gitProject.getName());
+        File projectDir =
+                new File(GitRepository.getProjectDir(gitProject.getName()), gitProject.getBranch());
         return TreeUtil.treeNodeData(projectDir, true);
     }
 

diff --git a/dinky-admin/src/main/java/org/dinky/sse/DoneStepSse.java b/dinky-admin/src/main/java/org/dinky/sse/DoneStepSse.java
@@ -19,13 +19,17 @@
 
 package org.dinky.sse;
 
+import org.dinky.function.pool.UdfCodePool;
+import org.dinky.service.GitProjectService;
+
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import cn.hutool.core.lang.Dict;
+import cn.hutool.extra.spring.SpringUtil;
 
 /**
  * @author ZackYoung
@@ -43,5 +47,10 @@ public DoneStepSse(
     }
 
     @Override
-    public void exec() {}
+    public void exec() {
+        addFileMsgCusLog("Updating UDF pool");
+        GitProjectService gitProjectService = SpringUtil.getBean(GitProjectService.class);
+        UdfCodePool.updateGitPool(gitProjectService.getGitPool());
+        addFileMsgCusLog("The UDF pool has been updated");
+    }
 }
diff --git a/dinky-admin/src/main/resources/mapper/GitProjectMapper.xml b/dinky-admin/src/main/resources/mapper/GitProjectMapper.xml
@@ -40,5 +40,7 @@
                 ${ew.sqlSegment}
             </if>
         </where>
+        order by order_line asc
+
     </select>
 </mapper>
-Original file line number
+Diff line change
@@ Expand Up / @@ -64,4 +64,6 @@ public class GitProjectDTO { @@
         /** */
         @NotNull private Boolean enabled;
+        private Integer orderLine;
     }