You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zh...@apache.org on 2022/07/28 11:25:35 UTC
[dolphinscheduler] branch dev updated: [Feature] Add new task type chunjun (#10937)
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 513f336015 [Feature] Add new task type chunjun (#10937)
513f336015 is described below
commit 513f3360152cacb3c26a207242ff616e623144d4
Author: xuhhui <xu...@cmss.chinamobile.com>
AuthorDate: Thu Jul 28 19:25:29 2022 +0800
[Feature] Add new task type chunjun (#10937)
---
docs/configs/docsdev.js | 8 +
docs/docs/en/guide/task/chunjun.md | 73 ++++++
docs/docs/zh/guide/task/chunjun.md | 47 ++++
docs/img/tasks/demo/chunjun_task01.png | Bin 0 -> 78991 bytes
docs/img/tasks/demo/chunjun_task02.png | Bin 0 -> 104612 bytes
docs/img/tasks/demo/chunjun_task03.png | Bin 0 -> 14797 bytes
docs/img/tasks/icons/chunjun.png | Bin 0 -> 179798 bytes
.../dolphinscheduler-task-chunjun/pom.xml | 52 +++++
.../plugin/task/chunjun/ChunJunConstants.java | 31 +++
.../plugin/task/chunjun/ChunJunParameters.java | 139 +++++++++++
.../plugin/task/chunjun/ChunJunTask.java | 257 +++++++++++++++++++++
.../plugin/task/chunjun/ChunJunTaskChannel.java | 52 +++++
.../task/chunjun/ChunJunTaskChannelFactory.java | 59 +++++
.../task/chunjun/ChunJunTaskExecutionContext.java | 62 +++++
.../plugin/task/chunjun/ChunJunConstantsTest.java | 46 ++++
.../plugin/task/chunjun/ChunJunParametersTest.java | 78 +++++++
dolphinscheduler-task-plugin/pom.xml | 1 +
.../public/images/task-icons/chunjun.png | Bin 0 -> 1659 bytes
.../public/images/task-icons/chunjun_hover.png | Bin 0 -> 2243 bytes
dolphinscheduler-ui/src/locales/en_US/project.ts | 2 +
dolphinscheduler-ui/src/locales/zh_CN/project.ts | 2 +
.../projects/task/components/node/fields/index.ts | 2 +
.../node/fields/use-chunjun-deploy-mode.ts | 49 ++++
.../task/components/node/fields/use-chunjun.ts | 124 ++++++++++
.../projects/task/components/node/format-data.ts | 8 +
.../projects/task/components/node/tasks/index.ts | 4 +-
.../task/components/node/tasks/use-chunjun.ts | 86 +++++++
.../src/views/projects/task/constants/task-type.ts | 5 +
.../workflow/components/dag/dag.module.scss | 6 +
script/env/dolphinscheduler_env.sh | 3 +-
30 files changed, 1194 insertions(+), 2 deletions(-)
diff --git a/docs/configs/docsdev.js b/docs/configs/docsdev.js
index 813a22424d..c84d7924ce 100644
--- a/docs/configs/docsdev.js
+++ b/docs/configs/docsdev.js
@@ -185,6 +185,10 @@ export default {
title: 'SageMaker',
link: '/en-us/docs/dev/user_doc/guide/task/sagemaker.html',
},
+ {
+ title: 'ChunJun',
+ link: '/en-us/docs/dev/user_doc/guide/task/chunjun.html',
+ },
],
},
{
@@ -626,6 +630,10 @@ export default {
title: 'SageMaker',
link: '/zh-cn/docs/dev/user_doc/guide/task/sagemaker.html',
},
+ {
+ title: 'ChunJun',
+ link: '/zh-cn/docs/dev/user_doc/guide/task/chunjun.html',
+ },
],
},
{
diff --git a/docs/docs/en/guide/task/chunjun.md b/docs/docs/en/guide/task/chunjun.md
new file mode 100644
index 0000000000..b5851f5e88
--- /dev/null
+++ b/docs/docs/en/guide/task/chunjun.md
@@ -0,0 +1,73 @@
+# ChunJun
+
+## Overview
+
+ChunJun task type for executing ChunJun programs. For ChunJun nodes, the worker will execute `${CHUNJUN_HOME}/bin/start-chunjun` to analyze the input json file.
+
+## Create Task
+
+- Click `Project Management -> Project Name -> Workflow Definition`, and click the `Create Workflow` button to enter the DAG editing page.
+- Drag the <img src="../../../../img/tasks/icons/chunjun.png" width="15"/> from the toolbar to the drawing board.
+
+## Task Parameters
+
+| **Parameter** | **Description** |
+| ------- | ---------- |
+| Node name | The node name in a workflow definition is unique. |
+| Run flag | Identifies whether this node schedules normally, if it does not need to execute, select the prohibition execution. |
+| Task priority | When the number of worker threads is insufficient, execute in the order of priority from high to low, and tasks with the same priority will execute in a first-in first-out order. |
+| Description | Describe the function of the node. |
+| Worker group | Assign tasks to the machines of the worker group to execute. If `Default` is selected, randomly select a worker machine for execution. |
+| Environment Name | Configure the environment name in which run the script. |
+| Number of failed retries | The number of times the task failed to resubmit. |
+| Failed retry interval | The time interval (unit minute) for resubmitting the task after a failed task. |
+| Task group name | The task group name. |
+| Priority | The task priority. |
+| Delayed execution time | The time, in minutes, that a task is delayed in execution. |
+| Timeout alarm | Check the timeout alarm and timeout failure. When the task exceeds the "timeout period", an alarm email will be sent and the task execution will fail. |
+| Custom template | Custom the content of the ChunJun node's json profile. |
+| json | json configuration file for ChunJun synchronization. |
+| Custom parameters | It is a user-defined parameter, and will replace the content with `${variable}` in the script.
+| Deploy mode | Execute chunjun task mode, eg local standalone. |
+| Option Parameters | Support such as `-confProp "{\"flink.checkpoint.interval\":60000}"` |
+| Predecessor task | Selecting a predecessor task for the current task will set the selected predecessor task as upstream of the current task. |
+
+## Task Example
+
+This example demonstrates importing data from Hive into MySQL.
+
+### Configuring the ChunJun environment in DolphinScheduler
+
+If you are using the ChunJun task type in a production environment, it is necessary to configure the required environment first. The configuration file is as follows: `/dolphinscheduler/conf/env/dolphinscheduler_env.sh`.
+
+
+
+After the environment has been configured, DolphinScheduler needs to be restarted.
+
+### Configuring ChunJun Task Node
+
+As the data to be read from Hive, a custom json is required, refer to: [Hive Json Template](https://github.com/DTStack/chunjun/blob/master/chunjun-examples/json/hive/binlog_hive.json).
+
+After writing the required json file, you can configure the node content by following the steps in the diagram below.
+
+
+
+### View run results
+
+
+
+### Note
+
+Before execute ${CHUNJUN_HOME}/bin/start-chunjun, need to change the shell ${CHUNJUN_HOME}/bin/start-chunjun, remove '&' in order to run in front.
+
+ such as:
+
+```shell
+nohup $JAVA_RUN -cp $JAR_DIR $CLASS_NAME $@ &
+```
+
+update to following:
+
+```shell
+nohup $JAVA_RUN -cp $JAR_DIR $CLASS_NAME $@
+```
\ No newline at end of file
diff --git a/docs/docs/zh/guide/task/chunjun.md b/docs/docs/zh/guide/task/chunjun.md
new file mode 100644
index 0000000000..cbecc5187e
--- /dev/null
+++ b/docs/docs/zh/guide/task/chunjun.md
@@ -0,0 +1,47 @@
+# ChunJun节点
+
+## 综述
+
+ChunJun 任务类型，用于执行 ChunJun 程序。对于 ChunJun 节点，worker 会通过执行 `${CHUNJUN_HOME}/bin/start-chunjun` 来解析传入的 json 文件。
+
+## 创建任务
+
+- 点击项目管理 -> 项目名称 -> 工作流定义，点击“创建工作流”按钮，进入 DAG 编辑页面；
+- 拖动工具栏的<img src="../../../../img/tasks/icons/chunjun.png" width="15"/> 任务节点到画板中。
+
+## 任务参数
+
+- 节点名称：设置任务节点的名称。一个工作流定义中的节点名称是唯一的。
+- 运行标志：标识这个结点是否能正常调度，如果不需要执行，可以打开禁止执行开关。
+- 描述：描述该节点的功能。
+- 任务优先级：worker 线程数不足时，根据优先级从高到低依次执行，优先级一样时根据先进先出原则执行。
+- Worker 分组：任务分配给 worker 组的机器执行，选择 Default ，会随机选择一台 worker 机执行。
+- 环境名称：配置运行脚本的环境。
+- 任务组名称：任务组的名称。
+- 组内优先级：一个任务组内此任务的优先级。
+- 失败重试次数：任务失败重新提交的次数。
+- 失败重试间隔：任务失败重新提交任务的时间间隔，以分为单位。
+- 延时执行时间：任务延迟执行的时间，以分为单位。
+- 超时警告：勾选超时警告、超时失败，当任务超过“超时时长”后，会发送告警邮件并且任务执行失败。
+- 自定义模板：自定义 ChunJun 节点的 json 配置文件内容，当前支持此种方式。
+- json：ChunJun 同步的 json 配置文件。
+- 自定义参数：用户自定义参数，会替换脚本中以 ${变量} 的内容。
+- 部署方式： 执行ChunJun任务的方式，比如local，standalone等。
+- 选项参数： 支持 `-confProp "{\"flink.checkpoint.interval\":60000}"` 格式。
+- 前置任务：选择当前任务的前置任务，会将被选择的前置任务设置为当前任务的上游。
+
+## 任务样例
+
+该样例演示为从 Hive 数据导入到 MySQL 中。
+
+### 在 DolphinScheduler 中配置 ChunJun 环境
+
+若生产环境中要是使用到 ChunJun 任务类型，则需要先配置好所需的环境。配置文件如下：`/dolphinscheduler/conf/env/dolphinscheduler_env.sh`。
+
+
+
+当环境配置完成之后，需要重启 DolphinScheduler。
+
+### 配置 ChunJun 任务节点
+
+从 Hive 中读取数据，所以需要自定义 json，可参考：[Hive Json Template](https://github.com/DTStack/chunjun/blob/master/chunjun-examples/json/hive/binlog_hive.json)
\ No newline at end of file
diff --git a/docs/img/tasks/demo/chunjun_task01.png b/docs/img/tasks/demo/chunjun_task01.png
new file mode 100644
index 0000000000..7ecb8ed730
Binary files /dev/null and b/docs/img/tasks/demo/chunjun_task01.png differ
diff --git a/docs/img/tasks/demo/chunjun_task02.png b/docs/img/tasks/demo/chunjun_task02.png
new file mode 100644
index 0000000000..0950f75b69
Binary files /dev/null and b/docs/img/tasks/demo/chunjun_task02.png differ
diff --git a/docs/img/tasks/demo/chunjun_task03.png b/docs/img/tasks/demo/chunjun_task03.png
new file mode 100644
index 0000000000..9c2d5592a4
Binary files /dev/null and b/docs/img/tasks/demo/chunjun_task03.png differ
diff --git a/docs/img/tasks/icons/chunjun.png b/docs/img/tasks/icons/chunjun.png
new file mode 100644
index 0000000000..b5833c2636
Binary files /dev/null and b/docs/img/tasks/icons/chunjun.png differ
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/pom.xml
new file mode 100644
index 0000000000..6929353a9b
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/pom.xml
@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>dolphinscheduler-task-plugin</artifactId>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <version>dev-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>dolphinscheduler-task-chunjun</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-datasource-all</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-spi</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-task-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-datasource-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ </dependencies>
+</project>
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstants.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstants.java
new file mode 100644
index 0000000000..c16f0b4e7b
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstants.java
@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+/**
+ * ChunJun constants
+ */
+public class ChunJunConstants {
+
+ public static final String FLINK_CONF_DIR = "${FLINK_HOME}/conf";
+
+ public static final String FLINK_LIB_DIR = "${FLINK_HOME}/lib";
+
+ public static final String HADOOP_CONF_DIR = "${HADOOP_HOME}/etc/hadoop";
+
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParameters.java
new file mode 100644
index 0000000000..7a31b00c65
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParameters.java
@@ -0,0 +1,139 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
+import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
+import org.apache.dolphinscheduler.spi.enums.Flag;
+import org.apache.dolphinscheduler.spi.utils.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+/**
+ * chunjun parameters
+ */
+@Setter
+@Getter
+@ToString
+public class ChunJunParameters extends AbstractParameters {
+
+ /**
+ * custom json config，default 1, support custom json
+ */
+ private int customConfig;
+
+ /**
+ * custom config json
+ */
+ private String json;
+
+ /**
+ * other arguments -confProp "{\"flink.checkpoint.interval\":60000}"
+ */
+ private String others;
+
+ /**
+ * deploy mode local standlone yarn-session yarn-per-job
+ */
+ private String deployMode;
+
+ /**
+ * customConfig value is 0, datasource type，eg mysql
+ */
+ private String dsType;
+
+ /**
+ * customConfig value is 0, datasource id int
+ */
+ private int dataSource;
+
+ /**
+ * customConfig value is 0, datasource targetType，eg MYSQL, POSTGRES
+ */
+ private String dtType;
+
+ /**
+ * customConfig value is 0, data target id
+ */
+ private int dataTarget;
+
+ /**
+ * customConfig value is 0, sql
+ */
+ private String sql;
+
+ /**
+ * customConfig value is 0, target table
+ */
+ private String targetTable;
+
+ /**
+ * pre statements
+ */
+ private List<String> preStatements;
+
+ /**
+ * post statements
+ */
+ private List<String> postStatements;
+
+ /**
+ * customConfig value is 0, job speed byte
+ */
+ private int jobSpeedByte;
+
+ /**
+ * customConfig value is 0, job speed record count
+ */
+ private int jobSpeedRecord;
+
+ @Override
+ public boolean checkParameters() {
+ if (customConfig == Flag.NO.ordinal()) {
+ return dataSource != 0 && dataTarget != 0
+ && StringUtils.isNotEmpty(sql)
+ && StringUtils.isNotEmpty(targetTable);
+ } else {
+ return StringUtils.isNotEmpty(json);
+ }
+ }
+
+ @Override
+ public List<ResourceInfo> getResourceFilesList() {
+ return new ArrayList<>();
+ }
+
+ @Override
+ public ResourceParametersHelper getResources() {
+ ResourceParametersHelper resources = super.getResources();
+
+ if (customConfig == Flag.YES.ordinal()) {
+ return resources;
+ }
+ resources.put(ResourceType.DATASOURCE, dataSource);
+ resources.put(ResourceType.DATASOURCE, dataTarget);
+ return resources;
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTask.java
new file mode 100644
index 0000000000..92a9912eb1
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTask.java
@@ -0,0 +1,257 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.RWXR_XR_X;
+
+import org.apache.dolphinscheduler.plugin.task.api.AbstractTaskExecutor;
+import org.apache.dolphinscheduler.plugin.task.api.ShellCommandExecutor;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskResponse;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParameterUtils;
+import org.apache.dolphinscheduler.spi.enums.Flag;
+import org.apache.dolphinscheduler.spi.utils.JSONUtils;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.SystemUtils;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.FileAttribute;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * chunjun task
+ */
+public class ChunJunTask extends AbstractTaskExecutor {
+ /**
+ * chunjun path
+ */
+ private static final String CHUNJUN_PATH = "${CHUNJUN_HOME}/bin/start-chunjun";
+
+ /**
+ * chunjun dist
+ */
+ private static final String CHUNJUN_DIST_DIR = "${CHUNJUN_HOME}/chunjun-dist";
+
+ /**
+ * chunJun parameters
+ */
+ private ChunJunParameters chunJunParameters;
+
+ /**
+ * shell command executor
+ */
+ private ShellCommandExecutor shellCommandExecutor;
+
+ /**
+ * taskExecutionContext
+ */
+ private TaskExecutionContext taskExecutionContext;
+
+ public ChunJunTask(TaskExecutionContext taskExecutionContext) {
+ super(taskExecutionContext);
+ this.taskExecutionContext = taskExecutionContext;
+
+ this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle,
+ taskExecutionContext, logger);
+ }
+
+ /**
+ * init chunjun config
+ */
+ @Override
+ public void init() {
+ logger.info("chunjun task params {}", taskExecutionContext.getTaskParams());
+ chunJunParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), ChunJunParameters.class);
+
+ if (!chunJunParameters.checkParameters()) {
+ throw new RuntimeException("chunjun task params is not valid");
+ }
+ }
+
+ /**
+ * run chunjun process
+ *
+ * @throws Exception exception
+ */
+ @Override
+ public void handle() throws Exception {
+ try {
+ Map<String, Property> paramsMap = taskExecutionContext.getPrepareParamsMap();
+
+ String jsonFilePath = buildChunJunJsonFile(paramsMap);
+ String shellCommandFilePath = buildShellCommandFile(jsonFilePath, paramsMap);
+ TaskResponse commandExecuteResult = shellCommandExecutor.run(shellCommandFilePath);
+
+ setExitStatusCode(commandExecuteResult.getExitStatusCode());
+ setAppIds(commandExecuteResult.getAppIds());
+ setProcessId(commandExecuteResult.getProcessId());
+ } catch (Exception e) {
+ logger.error("chunjun task failed.", e);
+ setExitStatusCode(EXIT_CODE_FAILURE);
+ throw e;
+ }
+ }
+
+ /**
+ * build chunjun json file
+ *
+ * @param paramsMap
+ * @return
+ * @throws Exception
+ */
+ private String buildChunJunJsonFile(Map<String, Property> paramsMap)
+ throws Exception {
+ // generate json
+ String fileName = String.format("%s/%s_job.json",
+ taskExecutionContext.getExecutePath(),
+ taskExecutionContext.getTaskAppId());
+
+ String json = null;
+
+ Path path = new File(fileName).toPath();
+ if (Files.exists(path)) {
+ return fileName;
+ }
+
+ if (chunJunParameters.getCustomConfig() == Flag.YES.ordinal()) {
+ json = chunJunParameters.getJson().replaceAll("\\r\\n", "\n");
+ }
+
+ // replace placeholder
+ json = ParameterUtils.convertParameterPlaceholders(json, ParamUtils.convert(paramsMap));
+
+ logger.debug("chunjun job json : {}", json);
+
+ // create chunjun json file
+ FileUtils.writeStringToFile(new File(fileName), json, StandardCharsets.UTF_8);
+ return fileName;
+ }
+
+ /**
+ * create command
+ *
+ * @return shell command file name
+ * @throws Exception if error throws Exception
+ */
+ private String buildShellCommandFile(String jobConfigFilePath, Map<String, Property> paramsMap)
+ throws Exception {
+ // generate scripts
+ String fileName = String.format("%s/%s_node.%s",
+ taskExecutionContext.getExecutePath(),
+ taskExecutionContext.getTaskAppId(),
+ SystemUtils.IS_OS_WINDOWS ? "bat" : "sh");
+
+ Path path = new File(fileName).toPath();
+
+ if (Files.exists(path)) {
+ return fileName;
+ }
+
+ // chunjun command
+ List<String> args = new ArrayList<>();
+
+ args.add(CHUNJUN_PATH);
+ args.add("-mode");
+ args.add(getExecMode(chunJunParameters));
+ args.add("-jobType sync");
+ args.add("-job");
+ args.add(jobConfigFilePath);
+ args.add("-chunjunDistDir");
+ args.add(CHUNJUN_DIST_DIR);
+
+ if (!"local".equalsIgnoreCase(getExecMode(chunJunParameters))) {
+ args.add("-flinkConfDir");
+ args.add(ChunJunConstants.FLINK_CONF_DIR);
+
+ args.add("-flinkLibDir");
+ args.add(ChunJunConstants.FLINK_LIB_DIR);
+
+ args.add("-hadoopConfDir");
+ args.add(ChunJunConstants.HADOOP_CONF_DIR);
+ }
+
+ if (chunJunParameters.getOthers() != null) {
+ args.add(chunJunParameters.getOthers());
+ }
+
+ String command = String.join(" ", args);
+
+ // replace placeholder
+ String chunjunCommand = ParameterUtils.convertParameterPlaceholders(command, ParamUtils.convert(paramsMap));
+
+ logger.info("raw script : {}", chunjunCommand);
+
+ // create shell command file
+ Set<PosixFilePermission> perms = PosixFilePermissions.fromString(RWXR_XR_X);
+ FileAttribute<Set<PosixFilePermission>> attr = PosixFilePermissions.asFileAttribute(perms);
+
+ if (SystemUtils.IS_OS_WINDOWS) {
+ Files.createFile(path);
+ } else {
+ Files.createFile(path, attr);
+ }
+
+ Files.write(path, chunjunCommand.getBytes(), StandardOpenOption.APPEND);
+
+ return fileName;
+ }
+
+ public String getExecMode(ChunJunParameters chunJunParameters) {
+ if (chunJunParameters.getDeployMode() == null) {
+ return "local";
+ }
+ return chunJunParameters.getDeployMode();
+ }
+
+ /**
+ * get task parameters
+ *
+ * @return AbstractParameters
+ */
+ @Override
+ public AbstractParameters getParameters() {
+ return chunJunParameters;
+ }
+
+ /**
+ * cancel ChunJun process
+ *
+ * @param cancelApplication cancelApplication
+ * @throws Exception if error throws Exception
+ */
+ @Override
+ public void cancelApplication(boolean cancelApplication) throws Exception {
+ // cancel process
+ shellCommandExecutor.cancelApplication();
+ }
+
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannel.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannel.java
new file mode 100644
index 0000000000..609eeefbcb
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannel.java
@@ -0,0 +1,52 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
+import org.apache.dolphinscheduler.spi.utils.JSONUtils;
+
+/**
+ * chunjun task channel
+ */
+public class ChunJunTaskChannel implements TaskChannel {
+
+ @Override
+ public void cancelApplication(boolean status) {
+
+ }
+
+ @Override
+ public AbstractTask createTask(TaskExecutionContext taskExecutionContext) {
+ return new ChunJunTask(taskExecutionContext);
+ }
+
+ @Override
+ public AbstractParameters parseParameters(ParametersNode parametersNode) {
+ return JSONUtils.parseObject(parametersNode.getTaskParams(), ChunJunParameters.class);
+ }
+
+ @Override
+ public ResourceParametersHelper getResources(String parameters) {
+ return JSONUtils.parseObject(parameters, ChunJunParameters.class).getResources();
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannelFactory.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannelFactory.java
new file mode 100644
index 0000000000..06a0de36f1
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskChannelFactory.java
@@ -0,0 +1,59 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory;
+import org.apache.dolphinscheduler.spi.params.base.PluginParams;
+
+import java.util.List;
+
+import com.google.auto.service.AutoService;
+
+/**
+ * chunjun task channelFactory
+ */
+@AutoService(TaskChannelFactory.class)
+public class ChunJunTaskChannelFactory implements TaskChannelFactory {
+
+ @Override
+ public TaskChannel create() {
+ return new ChunJunTaskChannel();
+ }
+
+ /**
+ * plugin name
+ * Must be UNIQUE .
+ *
+ * @return this task plugin name
+ */
+ @Override
+ public String getName() {
+ return "CHUNJUN";
+ }
+
+ /**
+ * Returns the configurable parameters that this plugin needs to display on the web ui
+ *
+ * @return this plugin params
+ */
+ @Override
+ public List<PluginParams> getParams() {
+ return null;
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskExecutionContext.java
new file mode 100644
index 0000000000..eb2bfb62d2
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/main/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunTaskExecutionContext.java
@@ -0,0 +1,62 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.apache.dolphinscheduler.spi.enums.DbType;
+
+import java.io.Serializable;
+
+import lombok.Data;
+
+/**
+ * chunjun taskExecutionContext
+ */
+@Data
+public class ChunJunTaskExecutionContext implements Serializable {
+
+ /**
+ * dataSourceId
+ */
+ private int dataSourceId;
+
+ /**
+ * sourcetype
+ */
+ private DbType sourcetype;
+
+ /**
+ * sourceConnectionParams
+ */
+ private String sourceConnectionParams;
+
+ /**
+ * dataTargetId
+ */
+ private int dataTargetId;
+
+ /**
+ * targetType
+ */
+ private DbType targetType;
+
+ /**
+ * targetConnectionParams
+ */
+ private String targetConnectionParams;
+
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstantsTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstantsTest.java
new file mode 100644
index 0000000000..bfdfc60f71
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunConstantsTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ChunJunConstantsTest {
+
+ private String flinkConfDir;
+
+ private String flinkLibDir;
+
+ private String hadoopConfDir;
+
+ @Before
+ public void setUp() {
+ flinkConfDir = "${FLINK_HOME}/conf";
+ flinkLibDir = "${FLINK_HOME}/lib";
+ hadoopConfDir = "${HADOOP_HOME}/etc/hadoop";
+ }
+
+ @Test
+ public void testEqualsString() {
+ Assert.assertEquals(ChunJunConstants.FLINK_CONF_DIR, flinkConfDir);
+ Assert.assertEquals(ChunJunConstants.FLINK_LIB_DIR, flinkLibDir);
+ Assert.assertEquals(ChunJunConstants.HADOOP_CONF_DIR, hadoopConfDir);
+ }
+
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParametersTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParametersTest.java
new file mode 100644
index 0000000000..a309e74171
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-chunjun/src/test/java/org/apache/dolphinscheduler/plugin/task/chunjun/ChunJunParametersTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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 org.apache.dolphinscheduler.plugin.task.chunjun;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ChunJunParametersTest {
+
+ private ChunJunParameters chunJunParameters = new ChunJunParameters();
+
+ @Before
+ public void setUp() {
+ chunJunParameters.setCustomConfig(0);
+ chunJunParameters.setDataSource(1);
+ chunJunParameters.setDataTarget(1);
+ chunJunParameters.setDsType("MYSQL");
+ chunJunParameters.setDtType("MYSQL");
+ chunJunParameters.setJobSpeedByte(1);
+ chunJunParameters.setJobSpeedRecord(1);
+ chunJunParameters.setJson("json");
+ }
+
+ @Test
+ public void testToString() {
+
+ String expected = "ChunJunParameters"
+ + "{"
+ + "customConfig=0, "
+ + "json='json', "
+ + "dsType='MYSQL', "
+ + "dataSource=1, "
+ + "dtType='MYSQL', "
+ + "dataTarget=1, "
+ + "sql='null', "
+ + "targetTable='null', "
+ + "preStatements=null, "
+ + "postStatements=null, "
+ + "jobSpeedByte=1, "
+ + "jobSpeedRecord=1, "
+ + "others=xx, "
+ + "deployMode=local"
+ + "}";
+
+ Assert.assertNotEquals(expected, chunJunParameters.toString());
+ }
+
+ @Test
+ public void testCheckParameters() {
+ Assert.assertFalse(chunJunParameters.checkParameters());
+ }
+
+ @Test
+ public void testGetResourceFilesList() {
+ Assert.assertNotNull(chunJunParameters.getResourceFilesList());
+ }
+
+ @Test
+ public void testGetResources() {
+ Assert.assertNotNull(chunJunParameters.getResources());
+ }
+}
diff --git a/dolphinscheduler-task-plugin/pom.xml b/dolphinscheduler-task-plugin/pom.xml
index a9b094b443..3764a730d9 100644
--- a/dolphinscheduler-task-plugin/pom.xml
+++ b/dolphinscheduler-task-plugin/pom.xml
@@ -58,6 +58,7 @@
<module>dolphinscheduler-task-dvc</module>
<module>dolphinscheduler-task-dinky</module>
<module>dolphinscheduler-task-sagemaker</module>
+ <module>dolphinscheduler-task-chunjun</module>
</modules>
<dependencyManagement>
diff --git a/dolphinscheduler-ui/public/images/task-icons/chunjun.png b/dolphinscheduler-ui/public/images/task-icons/chunjun.png
new file mode 100644
index 0000000000..b47f8ccd57
Binary files /dev/null and b/dolphinscheduler-ui/public/images/task-icons/chunjun.png differ
diff --git a/dolphinscheduler-ui/public/images/task-icons/chunjun_hover.png b/dolphinscheduler-ui/public/images/task-icons/chunjun_hover.png
new file mode 100644
index 0000000000..78bce08ebb
Binary files /dev/null and b/dolphinscheduler-ui/public/images/task-icons/chunjun_hover.png differ
diff --git a/dolphinscheduler-ui/src/locales/en_US/project.ts b/dolphinscheduler-ui/src/locales/en_US/project.ts
index 7f468e13d5..895876ea32 100644
--- a/dolphinscheduler-ui/src/locales/en_US/project.ts
+++ b/dolphinscheduler-ui/src/locales/en_US/project.ts
@@ -527,6 +527,8 @@ export default {
datax_job_runtime_memory_xms: 'Low Limit Value',
datax_job_runtime_memory_xmx: 'High Limit Value',
datax_job_runtime_memory_unit: 'G',
+ chunjun_custom_template: 'Custom Template',
+ chunjun_json_template: 'JSON',
current_hour: 'CurrentHour',
last_1_hour: 'Last1Hour',
last_2_hour: 'Last2Hours',
diff --git a/dolphinscheduler-ui/src/locales/zh_CN/project.ts b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
index 6723442962..58bf4bf974 100644
--- a/dolphinscheduler-ui/src/locales/zh_CN/project.ts
+++ b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
@@ -520,6 +520,8 @@ export default {
datax_job_runtime_memory_xms: '最小内存',
datax_job_runtime_memory_xmx: '最大内存',
datax_job_runtime_memory_unit: 'G',
+ chunjun_custom_template: '自定义模板',
+ chunjun_json_template: 'JSON',
current_hour: '当前小时',
last_1_hour: '前1小时',
last_2_hour: '前2小时',
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
index c0f2f25046..6691d67dbe 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
@@ -73,3 +73,5 @@ export { useOpenmldb } from './use-openmldb'
export { useDvc } from './use-dvc'
export { useDinky } from './use-dinky'
export { useSagemaker } from './use-sagemaker'
+export { useChunjun } from './use-chunjun'
+export { useChunjunDeployMode } from './use-chunjun-deploy-mode'
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun-deploy-mode.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun-deploy-mode.ts
new file mode 100644
index 0000000000..9cd5083a9c
--- /dev/null
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun-deploy-mode.ts
@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+import { useI18n } from 'vue-i18n'
+import type { IJsonItem } from '../types'
+
+export function useChunjunDeployMode(span = 24): IJsonItem {
+ const { t } = useI18n()
+
+ return {
+ type: 'radio',
+ field: 'deployMode',
+ name: t('project.node.deploy_mode'),
+ options: DEPLOY_MODES,
+ span
+ }
+}
+
+export const DEPLOY_MODES = [
+ {
+ label: 'local',
+ value: 'local'
+ },
+ {
+ label: 'standlone',
+ value: 'standlone'
+ },
+ {
+ label: 'yarn-session',
+ value: 'yarn-session'
+ },
+ {
+ label: 'yarn-per-job',
+ value: 'yarn-per-job'
+ }
+]
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun.ts
new file mode 100644
index 0000000000..f98933ea92
--- /dev/null
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-chunjun.ts
@@ -0,0 +1,124 @@
+/*
+ * 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.
+ */
+import { ref, onMounted, watch } from 'vue'
+import { useI18n } from 'vue-i18n'
+import type { IJsonItem } from '../types'
+import { useChunjunDeployMode } from './'
+
+export function useChunjun(model: { [field: string]: any }): IJsonItem[] {
+ const { t } = useI18n()
+
+ const jsonEditorSpan = ref(0)
+ const customParameterSpan = ref(0)
+
+ const initConstants = () => {
+ if (model.customConfig) {
+ jsonEditorSpan.value = 24
+ customParameterSpan.value = 24
+ } else {
+ jsonEditorSpan.value = 0
+ customParameterSpan.value = 0
+ }
+ }
+
+ onMounted(() => {
+ initConstants()
+ })
+
+ watch(
+ () => model.customConfig,
+ () => {
+ initConstants()
+ }
+ )
+
+ return [
+ {
+ type: 'switch',
+ field: 'customConfig',
+ value: true,
+ name: t('project.node.chunjun_custom_template'),
+ props: {
+ disabled: true
+ }
+ },
+ {
+ type: 'editor',
+ field: 'json',
+ name: t('project.node.chunjun_json_template'),
+ span: jsonEditorSpan,
+ validate: {
+ trigger: ['input', 'trigger'],
+ required: true,
+ message: t('project.node.sql_empty_tips')
+ }
+ },
+ {
+ type: 'custom-parameters',
+ field: 'localParams',
+ name: t('project.node.custom_parameters'),
+ span: customParameterSpan,
+ children: [
+ {
+ type: 'input',
+ field: 'prop',
+ span: 10,
+ props: {
+ placeholder: t('project.node.prop_tips'),
+ maxLength: 256
+ },
+ validate: {
+ trigger: ['input', 'blur'],
+ required: true,
+ validator(validate: any, value: string) {
+ if (!value) {
+ return new Error(t('project.node.prop_tips'))
+ }
+
+ const sameItems = model.localParams.filter(
+ (item: { prop: string }) => item.prop === value
+ )
+
+ if (sameItems.length > 1) {
+ return new Error(t('project.node.prop_repeat'))
+ }
+ }
+ }
+ },
+ {
+ type: 'input',
+ field: 'value',
+ span: 10,
+ props: {
+ placeholder: t('project.node.value_tips'),
+ maxLength: 256
+ }
+ }
+ ]
+ },
+ useChunjunDeployMode(24),
+ {
+ type: 'input',
+ field: 'others',
+ name: t('project.node.option_parameters'),
+ props: {
+ type: 'textarea',
+ placeholder: t('project.node.option_parameters_tips')
+ }
+ }
+ ]
+}
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
index b6f24d4044..f7ff744b97 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
@@ -402,9 +402,17 @@ export function formatParams(data: INodeData): {
taskParams.sql = data.sql
}
+ if (data.taskType === 'CHUNJUN') {
+ taskParams.customConfig = data.customConfig ? 1 : 0
+ taskParams.json = data.json
+ taskParams.deployMode = data.deployMode
+ taskParams.others = data.others
+ }
+
if (data.taskType === 'PIGEON') {
taskParams.targetJobName = data.targetJobName
}
+
let timeoutNotifyStrategy = ''
if (data.timeoutNotifyStrategy) {
if (data.timeoutNotifyStrategy.length === 1) {
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
index 1d16aeef03..81eaa8c023 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
@@ -41,6 +41,7 @@ import { useOpenmldb } from './use-openmldb'
import { useDvc } from './use-dvc'
import { useDinky } from './use-dinky'
import { userSagemaker } from './use-sagemaker'
+import { useChunjun } from './use-chunjun'
export default {
SHELL: useShell,
@@ -68,5 +69,6 @@ export default {
OPENMLDB: useOpenmldb,
DVC: useDvc,
DINKY: useDinky,
- SAGEMAKER: userSagemaker
+ SAGEMAKER: userSagemaker,
+ CHUNJUN: useChunjun
}
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-chunjun.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-chunjun.ts
new file mode 100644
index 0000000000..362bd944fb
--- /dev/null
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-chunjun.ts
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+import { reactive } from 'vue'
+import * as Fields from '../fields/index'
+import type { IJsonItem, INodeData } from '../types'
+import { ITaskData } from '../types'
+
+export function useChunjun({
+ projectCode,
+ from = 0,
+ readonly,
+ data
+}: {
+ projectCode: number
+ from?: number
+ readonly?: boolean
+ data?: ITaskData
+}) {
+ const model = reactive({
+ name: '',
+ taskType: 'CHUNJUN',
+ flag: 'YES',
+ description: '',
+ deployMode: 'local',
+ timeoutFlag: false,
+ localParams: [],
+ environmentCode: null,
+ failRetryInterval: 1,
+ failRetryTimes: 0,
+ workerGroup: 'default',
+ delayTime: 0,
+ timeout: 30,
+ customConfig: false,
+ preStatements: [],
+ postStatements: [],
+ timeoutNotifyStrategy: ['WARN']
+ } as INodeData)
+
+ let extra: IJsonItem[] = []
+ if (from === 1) {
+ extra = [
+ Fields.useTaskType(model, readonly),
+ Fields.useProcessName({
+ model,
+ projectCode,
+ isCreate: !data?.id,
+ from,
+ processName: data?.processName
+ })
+ ]
+ }
+
+ return {
+ json: [
+ Fields.useName(from),
+ ...extra,
+ Fields.useRunFlag(),
+ Fields.useDescription(),
+ Fields.useTaskPriority(),
+ Fields.useWorkerGroup(),
+ Fields.useEnvironmentName(model, !model.id),
+ ...Fields.useTaskGroup(model, projectCode),
+ ...Fields.useFailed(),
+ Fields.useDelayTime(model),
+ ...Fields.useTimeoutAlarm(model),
+ ...Fields.useChunjun(model),
+ Fields.usePreTasks()
+ ] as IJsonItem[],
+ model
+ }
+}
diff --git a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
index b441c6229c..5f2e4c95f4 100644
--- a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
@@ -41,6 +41,7 @@ export type TaskType =
| 'DVC'
| 'DINKY'
| 'SAGEMAKER'
+ | 'CHUNJUN'
export const TASK_TYPES_MAP = {
SHELL: {
@@ -133,5 +134,9 @@ export const TASK_TYPES_MAP = {
SAGEMAKER: {
alias: 'SageMaker',
helperLinkDisable: true
+ },
+ CHUNJUN: {
+ alias: 'CHUNJUN',
+ helperLinkDisable: true
}
} as { [key in TaskType]: { alias: string; helperLinkDisable?: boolean } }
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss
index c22dec8480..bdfb9e3ac4 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss
+++ b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss
@@ -179,6 +179,9 @@ $bgLight: #ffffff;
&.icon-sagemaker {
background-image: url('/images/task-icons/sagemaker.png');
}
+ &.icon-chunjun {
+ background-image: url('/images/task-icons/chunjun.png');
+ }
}
&:hover {
@@ -261,6 +264,9 @@ $bgLight: #ffffff;
&.icon-sagemaker {
background-image: url('/images/task-icons/sagemaker_hover.png');
}
+ &.icon-chunjun {
+ background-image: url('/images/task-icons/chunjun_hover.png');
+ }
}
}
}
diff --git a/script/env/dolphinscheduler_env.sh b/script/env/dolphinscheduler_env.sh
index 44f0de7792..b326d119a1 100755
--- a/script/env/dolphinscheduler_env.sh
+++ b/script/env/dolphinscheduler_env.sh
@@ -44,5 +44,6 @@ export HIVE_HOME=${HIVE_HOME:-/opt/soft/hive}
export FLINK_HOME=${FLINK_HOME:-/opt/soft/flink}
export DATAX_HOME=${DATAX_HOME:-/opt/soft/datax}
export SEATUNNEL_HOME=${SEATUNNEL_HOME:-/opt/soft/seatunnel}
+export CHUNJUN_HOME=${CHUNJUN_HOME:-/opt/soft/chunjun}
-export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$FLINK_HOME/bin:$DATAX_HOME/bin:$SEATUNNEL_HOME/bin:$PATH
+export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$FLINK_HOME/bin:$DATAX_HOME/bin:$SEATUNNEL_HOME/bin:$CHUNJUN_HOME/bin:$PATH