[feat][task plugin] Add Jupyter task plugin (#9872)

Co-authored-by: Amy0104 <97265214+Amy0104@users.noreply.github.com>

docs/configs/docsdev.js

@@ -170,6 +170,10 @@ export default {
                                 title: 'Apache Zeppelin',
                                 link: '/en-us/docs/dev/user_doc/guide/task/zeppelin.html',
                             },
+                            {
+                                title: 'Jupyter',
+                                link: '/en-us/docs/dev/user_doc/guide/task/jupyter.html',
+                            },
                         ],
                     },
                     {
@@ -517,6 +521,10 @@ export default {
                                 title: 'Apache Zeppelin',
                                 link: '/zh-cn/docs/dev/user_doc/guide/task/zeppelin.html',
                             },
+                            {
+                                title: 'Jupyter',
+                                link: '/zh-cn/docs/dev/user_doc/guide/task/jupyter.html',
+                            },
                         ],
                     },
                     {

docs/docs/en/guide/task/jupyter.md

@@ -0,0 +1,45 @@
+# Jupyter
+
+## Overview
+
+Use `Jupyter Task` to create a jupyter-type task and execute jupyter notes. When the worker executes `Jupyter Task`,
+it will use `papermill` to evaluate jupyter notes. Click [here](https://papermill.readthedocs.io/en/latest/) for details about `papermill`.
+
+## Conda Configuration
+ 
+- Config `conda.path` in `common.properties` to the path of your `conda.sh`, which should be the same `conda` you use to manage the python environment of your `papermill` and `jupyter`.
+Click [here](https://docs.conda.io/en/latest/) for more information about `conda`.
+- `conda.path` is set to `/opt/anaconda3/etc/profile.d/conda.sh` by default. If you have no idea where your `conda` is, simply run `conda info | grep -i 'base environment'`.
+
+## Create Task
+
+- Click Project Management-Project Name-Workflow Definition, and click the "Create Workflow" button to enter the DAG editing page.
+- Drag <img src="/img/tasks/icons/jupyter.png" width="15"/> from the toolbar to the canvas.
+
+## Task Parameter
+
+- Node name: The node name in a workflow definition is unique.
+- Run flag: Identifies whether this node can be scheduled normally, if it does not need to be executed, you can turn on the prohibition switch.
+- Descriptive information: Describe the function of the node.
+- 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.
+- Worker grouping: Assign tasks to the machines of the worker group to execute. If `Default` is selected, randomly select a worker machine for execution.
+- Number of failed retry attempts: The failure task resubmitting times. It supports drop-down and hand-filling.
+- Failed retry interval: The time interval for resubmitting the task after a failed task. It supports drop-down and hand-filling.
+- Timeout alarm: Check the timeout alarm and timeout failure. When the task exceeds the "timeout period", an alarm email will send and the task execution will fail.
+- Conda Env Name: Name of conda environment.
+- Input Note Path: Path of input jupyter note template.
+- Out Note Path: Path of output note.
+- Jupyter Parameters: Parameters in json format used for jupyter note parameterization.
+- Kernel: Jupyter notebook kernel.
+- Engine: Engine to evaluate jupyter notes.
+- Jupyter Execution Timeout: Timeout set for each jupyter notebook cell.
+- Jupyter Start Timeout: Timeout set for jupyter notebook kernel.
+- Others: Other command options for papermill.
+
+## Task Example
+
+### Jupyter Task Example
+
+This example illustrates how to create a jupyter task node.
+
+![demo-jupyter-simple](/img/tasks/demo/jupyter.png)

docs/docs/zh/guide/task/jupyter.md

@@ -0,0 +1,48 @@
+# Jupyter
+
+## Overview
+
+`Jupyter`任务类型，用于创建并执行`Jupyter`类型任务。worker 执行该任务的时候，会通过`papermill`执行`jupyter note`。
+点击[这里](https://papermill.readthedocs.io/en/latest/) 获取更多关于`papermill`的信息。
+
+## Conda Configuration
+ 
+- 在`common.properties`配置`conda.path`，将其指向您的`conda.sh`。这里的`conda`应该是您用来管理您的 `papermill`和`jupyter`所在python环境的相同`conda`。
+点击 [这里](https://docs.conda.io/en/latest/) 获取更多关于`conda`的信息.
+- `conda.path`默认设置为`/opt/anaconda3/etc/profile.d/conda.sh`。 如果您不清楚您的`conda`环境在哪里，只需要在命令行执行`conda info | grep -i 'base environment'`即可获得。
+
+
+## Create Task
+
+- 点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入DAG编辑页面。
+- 工具栏中拖动 <img src="/img/tasks/icons/jupyter.png" width="15"/> 到画板中，即可完成创建。
+
+## Task Parameter
+
+- 任务名称：设置任务的名称。一个工作流定义中的节点名称是唯一的。
+- 运行标志：标识这个节点是否能正常调度,如果不需要执行，可以打开禁止执行开关。
+- 描述：描述该节点的功能。
+- 任务优先级：worker线程数不足时，根据优先级从高到低依次执行，优先级一样时根据先进先出原则执行。
+- Worker分组：任务分配给worker组的机器机执行，选择Default，会随机选择一台worker机执行。
+- 失败重试次数：任务失败重新提交的次数，支持下拉和手填。
+- 失败重试间隔：任务失败重新提交任务的时间间隔，支持下拉和手填。
+- 超时告警：勾选超时告警、超时失败，当任务超过"超时时长"后，会发送告警邮件并且任务执行失败.
+- 前置任务：选择当前任务的前置任务，会将被选择的前置任务设置为当前任务的上游。
+- Conda Env Name: Conda环境名称。
+- Input Note Path: 输入的jupyter note模板路径。
+- Out Note Path: 输出的jupyter note路径。
+- Jupyter Parameters: 用于对接jupyter note参数化的JSON格式参数。
+- Kernel: Jupyter notebook 内核。
+- Engine: 用于执行Jupyter note的引擎名称。
+- Jupyter Execution Timeout: 对于每个jupyter notebook cell设定的超时时间。
+- Jupyter Start Timeout: 对于jupyter notebook kernel设定的启动超时时间。
+- Others: 传入papermill命令的其他参数。
+
+## Task Example
+
+### Jupyter Task Example
+
+这个示例展示了如何创建Jupyter任务节点：
+
+![demo-jupyter-simple](/img/tasks/demo/jupyter.png)
+

dolphinscheduler-common/src/main/resources/common.properties

@@ -93,3 +93,6 @@ alert.rpc.port=50052
 
 # Url endpoint for zeppelin RESTful API
 zeppelin.rest.url=http://localhost:8080
+
+# set path of conda.sh
+conda.path=/opt/anaconda3/etc/profile.d/conda.sh

dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml

@@ -142,6 +142,12 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-task-jupyter</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.dolphinscheduler</groupId>
             <artifactId>dolphinscheduler-task-blocking</artifactId>

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskConstants.java

@@ -403,5 +403,9 @@ public class TaskConstants {
      */
     public static final String ZEPPELIN_REST_URL= "zeppelin.rest.url";
 
+    /**
+     * conda config used by jupyter task plugin
+     */
+    public static final String CONDA_PATH = "conda.path";
 
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/pom.xml

@@ -0,0 +1,42 @@
+<?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-jupyter</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <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>
+    </dependencies>
+</project>

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/main/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterConstants.java

@@ -0,0 +1,80 @@
+/*
+ * 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.jupyter;
+
+public class JupyterConstants {
+
+    private JupyterConstants() {
+        throw new IllegalStateException("Utility class");
+    }
+
+    /**
+     * conda init
+     */
+    public static final String CONDA_INIT = "source";
+
+    /**
+     * conda activate
+     */
+    public static final String CONDA_ACTIVATE = "conda activate";
+
+    /**
+     * jointer to combine two command
+     */
+    public static final String JOINTER = "&&";
+
+    /**
+     * papermill
+     */
+    public static final String PAPERMILL = "papermill";
+
+    /**
+     * Parameters to pass to the parameters cell.
+     */
+    public static final String PARAMETERS = "--parameters";
+
+    /**
+     * Name of kernel to run.
+     */
+    public static final String KERNEL = "--kernel";
+
+    /**
+     * The execution engine name to use in evaluating the notebook.
+     */
+    public static final String ENGINE = "--engine";
+
+    /**
+     * Time in seconds to wait for each cell before failing execution (default: forever)
+     */
+    public static final String EXECUTION_TIMEOUT = "--execution-timeout";
+
+    /**
+     * Time in seconds to wait for kernel to start.
+     */
+    public static final String START_TIMEOUT = "--start-timeout";
+
+    /**
+     * Insert the paths of input/output notebooks as PAPERMILL_INPUT_PATH/PAPERMILL_OUTPUT_PATH as notebook parameters.
+     */
+    public static final String INJECT_PATHS = "--inject-paths";
+
+    /**
+     * Flag for turning on the progress bar.
+     */
+    public static final String PROGRESS_BAR = "--progress-bar";
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/main/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterParameters.java

@@ -0,0 +1,164 @@
+/*
+ * 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.jupyter;
+
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+
+/**
+ * jupyter parameters
+ */
+public class JupyterParameters extends AbstractParameters {
+
+    /**
+     * conda env name
+     */
+    private String condaEnvName;
+
+    /**
+     * input note path
+     */
+    private String inputNotePath;
+
+    /**
+     * output note path
+     */
+    private String outputNotePath;
+
+    /**
+     * parameters to pass into jupyter note cells
+     */
+    private String parameters;
+
+    /**
+     * jupyter kernel
+     */
+    private String kernel;
+
+    /**
+     * the execution engine name to use in evaluating the notebook
+     */
+    private String engine;
+
+    /**
+     * time in seconds to wait for each cell before failing execution (default: forever)
+     */
+    private String executionTimeout;
+
+    /**
+     * time in seconds to wait for kernel to start
+     */
+    private String startTimeout;
+
+    /**
+     * other arguments
+     */
+    private String others;
+
+
+    public String getCondaEnvName() {
+        return condaEnvName;
+    }
+
+    public void setCondaEnvName(String condaEnvName) {
+        this.condaEnvName = condaEnvName;
+    }
+
+    public String getInputNotePath() {
+        return inputNotePath;
+    }
+
+    public void setInputNotePath(String inputNotePath) {
+        this.inputNotePath = inputNotePath;
+    }
+
+    public String getOutputNotePath() {
+        return outputNotePath;
+    }
+
+    public void setOutputNotePath(String outputNotePath) {
+        this.outputNotePath = outputNotePath;
+    }
+
+    public String getParameters() {
+        return parameters;
+    }
+
+    public void setParameters(String parameters) {
+        this.parameters = parameters;
+    }
+
+    public String getKernel() {
+        return kernel;
+    }
+
+    public void setKernel(String kernel) {
+        this.kernel = kernel;
+    }
+
+    public String getEngine() {
+        return engine;
+    }
+
+    public void setEngine(String engine) {
+        this.engine = engine;
+    }
+
+    public String getExecutionTimeout() {
+        return executionTimeout;
+    }
+
+    public void setExecutionTimeout(String executionTimeout) {
+        this.executionTimeout = executionTimeout;
+    }
+
+    public String getStartTimeout() {
+        return startTimeout;
+    }
+
+    public void setStartTimeout(String startTimeout) {
+        this.startTimeout = startTimeout;
+    }
+
+    public String getOthers() {
+        return others;
+    }
+
+    public void setOthers(String others) {
+        this.others = others;
+    }
+
+    @Override
+    public boolean checkParameters() {
+        return condaEnvName != null && inputNotePath != null && outputNotePath != null;
+    }
+
+    @Override
+    public String toString() {
+        return "JupyterParameters{" +
+                "condaEnvName='" + condaEnvName + '\'' +
+                ", inputNotePath='" + inputNotePath + '\'' +
+                ", outputNotePath='" + outputNotePath + '\'' +
+                ", parameters='" + parameters + '\'' +
+                ", kernel='" + kernel + '\'' +
+                ", engine='" + engine + '\'' +
+                ", executionTimeout=" + executionTimeout +
+                ", startTimeout=" + startTimeout +
+                ", others='" + others + '\'' +
+                '}';
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/main/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterTask.java

@@ -0,0 +1,219 @@
+/*
+ * 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.jupyter;
+
+
+import org.apache.dolphinscheduler.plugin.task.api.AbstractTaskExecutor;
+import org.apache.dolphinscheduler.plugin.task.api.ShellCommandExecutor;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+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.plugin.task.api.utils.MapUtils;
+import org.apache.dolphinscheduler.spi.utils.JSONUtils;
+import org.apache.dolphinscheduler.spi.utils.PropertyUtils;
+import org.apache.dolphinscheduler.spi.utils.StringUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class JupyterTask extends AbstractTaskExecutor {
+
+    /**
+     * jupyter parameters
+     */
+    private JupyterParameters jupyterParameters;
+
+    /**
+     * taskExecutionContext
+     */
+    private TaskExecutionContext taskExecutionContext;
+
+    private ShellCommandExecutor shellCommandExecutor;
+
+    public JupyterTask(TaskExecutionContext taskExecutionContext) {
+        super(taskExecutionContext);
+        this.taskExecutionContext = taskExecutionContext;
+        this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle,
+                taskExecutionContext,
+                logger);
+    }
+
+    @Override
+    public void init() {
+        logger.info("jupyter task params {}", taskExecutionContext.getTaskParams());
+
+        jupyterParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), JupyterParameters.class);
+
+        if (null == jupyterParameters) {
+            logger.error("jupyter params is null");
+            return;
+        }
+
+        if (!jupyterParameters.checkParameters()) {
+            throw new RuntimeException("jupyter task params is not valid");
+        }
+    }
+
+    @Override
+    public void handle() throws Exception {
+        try {
+            // SHELL task exit code
+            TaskResponse response = shellCommandExecutor.run(buildCommand());
+            setExitStatusCode(response.getExitStatusCode());
+            setAppIds(response.getAppIds());
+            setProcessId(response.getProcessId());
+        } catch (Exception e) {
+            logger.error("jupyter task execution failure", e);
+            exitStatusCode = -1;
+            throw e;
+        }
+    }
+
+    /**
+     * create command
+     *
+     * @return command
+     */
+    protected String buildCommand() throws IOException {
+        /**
+         * papermill [OPTIONS] NOTEBOOK_PATH [OUTPUT_PATH]
+         */
+        List<String> args = new ArrayList<>();
+        final String condaPath = PropertyUtils.getString(TaskConstants.CONDA_PATH);
+        args.add(JupyterConstants.CONDA_INIT);
+        args.add(condaPath);
+        args.add(JupyterConstants.JOINTER);
+        args.add(JupyterConstants.CONDA_ACTIVATE);
+        args.add(jupyterParameters.getCondaEnvName());
+        args.add(JupyterConstants.JOINTER);
+        args.add(JupyterConstants.PAPERMILL);
+        args.add(jupyterParameters.getInputNotePath());
+        args.add(jupyterParameters.getOutputNotePath());
+
+        // populate jupyter parameterization
+        args.addAll(populateJupyterParameterization());
+
+        // populate jupyter options
+        args.addAll(populateJupyterOptions());
+
+        // replace placeholder, and combining local and global parameters
+        Map<String, Property> paramsMap = ParamUtils.convert(taskExecutionContext, getParameters());
+        if (MapUtils.isEmpty(paramsMap)) {
+            paramsMap = new HashMap<>();
+        }
+        if (MapUtils.isNotEmpty(taskExecutionContext.getParamsMap())) {
+            paramsMap.putAll(taskExecutionContext.getParamsMap());
+        }
+
+        String command = ParameterUtils.convertParameterPlaceholders(String.join(" ", args), ParamUtils.convert(paramsMap));
+
+        logger.info("jupyter task command: {}", command);
+
+        return command;
+    }
+
+
+    /**
+     * build jupyter parameterization
+     *
+     * @return argument list
+     */
+    private List<String> populateJupyterParameterization() throws IOException {
+        List<String> args = new ArrayList<>();
+        String parameters = jupyterParameters.getParameters();
+        if (StringUtils.isNotEmpty(parameters)) {
+            ObjectMapper mapper = new ObjectMapper();
+            try {
+                // convert JSON string to Map
+                Map<String, String> jupyterParamsMap = mapper.readValue(parameters, Map.class);
+                for (String key : jupyterParamsMap.keySet()) {
+                    args.add(JupyterConstants.PARAMETERS);
+                    args.add(key);
+                    args.add(jupyterParamsMap.get(key));
+                }
+
+            } catch (IOException e) {
+                logger.error("fail to parse jupyter parameterization", e);
+                throw e;
+            }
+        }
+        return args;
+    }
+
+    /**
+     * build jupyter options
+     *
+     * @return argument list
+     */
+    private List<String> populateJupyterOptions() {
+        List<String> args = new ArrayList<>();
+        String kernel = jupyterParameters.getKernel();
+        if (StringUtils.isNotEmpty(kernel)) {
+            args.add(JupyterConstants.KERNEL);
+            args.add(kernel);
+        }
+
+        String engine = jupyterParameters.getEngine();
+        if (StringUtils.isNotEmpty(engine)) {
+            args.add(JupyterConstants.ENGINE);
+            args.add(engine);
+        }
+
+        String executionTimeout = jupyterParameters.getExecutionTimeout();
+        if (StringUtils.isNotEmpty(executionTimeout)) {
+            args.add(JupyterConstants.EXECUTION_TIMEOUT);
+            args.add(executionTimeout);
+        }
+
+        String startTimeout = jupyterParameters.getStartTimeout();
+        if (StringUtils.isNotEmpty(startTimeout)) {
+            args.add(JupyterConstants.START_TIMEOUT);
+            args.add(startTimeout);
+        }
+
+        String others = jupyterParameters.getOthers();
+        if (StringUtils.isNotEmpty(others)) {
+            args.add(others);
+        }
+
+        args.add(JupyterConstants.INJECT_PATHS);
+        args.add(JupyterConstants.PROGRESS_BAR);
+        return args;
+    }
+
+    @Override
+    public void cancelApplication(boolean cancelApplication) throws Exception {
+        // cancel process
+        shellCommandExecutor.cancelApplication();
+    }
+
+    @Override
+    public AbstractParameters getParameters() {
+        return jupyterParameters;
+    }
+
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/main/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterTaskChannel.java

@@ -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.
+ */
+
+package org.apache.dolphinscheduler.plugin.task.jupyter;
+
+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;
+
+public class JupyterTaskChannel implements TaskChannel {
+
+    @Override
+    public void cancelApplication(boolean status) {
+
+    }
+
+    @Override
+    public AbstractTask createTask(TaskExecutionContext taskRequest) {
+        return new JupyterTask(taskRequest);
+    }
+
+    @Override
+    public AbstractParameters parseParameters(ParametersNode parametersNode) {
+        return JSONUtils.parseObject(parametersNode.getTaskParams(), JupyterParameters.class);
+    }
+
+    @Override
+    public ResourceParametersHelper getResources(String parameters) {
+        return null;
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/main/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterTaskChannelFactory.java

@@ -0,0 +1,44 @@
+/*
+ * 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.jupyter;
+
+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;
+
+@AutoService(TaskChannelFactory.class)
+public class JupyterTaskChannelFactory implements TaskChannelFactory {
+    @Override
+    public String getName() {
+        return "JUPYTER";
+    }
+
+    @Override
+    public List<PluginParams> getParams() {
+        return null;
+    }
+
+    @Override
+    public TaskChannel create() {
+        return new JupyterTaskChannel();
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-jupyter/src/test/java/org/apache/dolphinscheduler/plugin/task/jupyter/JupyterTaskTest.java

@@ -0,0 +1,88 @@
+/*
+ * 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.jupyter;
+
+
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.spi.utils.JSONUtils;
+
+import org.apache.dolphinscheduler.spi.utils.PropertyUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.core.classloader.annotations.SuppressStaticInitializationFor;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.powermock.api.mockito.PowerMockito.spy;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({
+    JSONUtils.class,
+    PropertyUtils.class,
+})
+@PowerMockIgnore({"javax.*"})
+@SuppressStaticInitializationFor("org.apache.dolphinscheduler.spi.utils.PropertyUtils")
+public class JupyterTaskTest {
+
+    @Test
+    public void testBuildJupyterCommand() throws Exception {
+        String parameters = buildJupyterCommand();
+        TaskExecutionContext taskExecutionContext = PowerMockito.mock(TaskExecutionContext.class);
+        when(taskExecutionContext.getTaskParams()).thenReturn(parameters);
+        PowerMockito.mockStatic(PropertyUtils.class);
+        when(PropertyUtils.getString(any())).thenReturn("/opt/anaconda3/etc/profile.d/conda.sh");
+        JupyterTask jupyterTask = spy(new JupyterTask(taskExecutionContext));
+        jupyterTask.init();
+        Assert.assertEquals(jupyterTask.buildCommand(),
+            "source /opt/anaconda3/etc/profile.d/conda.sh && " +
+                "conda activate jupyter-lab && " +
+                "papermill " +
+                "/test/input_note.ipynb " +
+                "/test/output_note.ipynb " +
+                "--parameters city Shanghai " +
+                "--parameters factor 0.01 " +
+                "--kernel python3 " +
+                "--engine default_engine " +
+                "--execution-timeout 10 " +
+                "--start-timeout 3 " +
+                "--version " +
+                "--inject-paths " +
+                "--progress-bar");
+    }
+
+    private String buildJupyterCommand() {
+        JupyterParameters jupyterParameters = new JupyterParameters();
+        jupyterParameters.setCondaEnvName("jupyter-lab");
+        jupyterParameters.setInputNotePath("/test/input_note.ipynb");
+        jupyterParameters.setOutputNotePath("/test/output_note.ipynb");
+        jupyterParameters.setParameters("{\"city\": \"Shanghai\", \"factor\": \"0.01\"}");
+        jupyterParameters.setKernel("python3");
+        jupyterParameters.setEngine("default_engine");
+        jupyterParameters.setExecutionTimeout("10");
+        jupyterParameters.setStartTimeout("3");
+        jupyterParameters.setOthers("--version");
+        return JSONUtils.toJsonString(jupyterParameters);
+    }
+
+}

dolphinscheduler-task-plugin/pom.xml

@@ -51,5 +51,6 @@
         <module>dolphinscheduler-task-emr</module>
         <module>dolphinscheduler-task-blocking</module>
         <module>dolphinscheduler-task-zeppelin</module>
+        <module>dolphinscheduler-task-jupyter</module>
     </modules>
 </project>

dolphinscheduler-ui/src/locales/modules/en_US.ts

@@ -933,6 +933,24 @@ const project = {
     zeppelin_paragraph_id: 'zeppelinParagraphId',
     zeppelin_paragraph_id_tips:
       'Please enter the paragraph id of your zeppelin paragraph',
+    jupyter_conda_env_name: 'condaEnvName',
+    jupyter_conda_env_name_tips: 'Please enter the conda environment name of papermill',
+    jupyter_input_note_path: 'inputNotePath',
+    jupyter_input_note_path_tips: 'Please enter the input jupyter note path',
+    jupyter_output_note_path: 'outputNotePath',
+    jupyter_output_note_path_tips: 'Please enter the output jupyter note path',
+    jupyter_parameters: 'parameters',
+    jupyter_parameters_tips: 'Please enter the parameters for jupyter parameterization',
+    jupyter_kernel: 'kernel',
+    jupyter_kernel_tips: 'Please enter the jupyter kernel name',
+    jupyter_engine: 'engine',
+    jupyter_engine_tips: 'Please enter the engine name',
+    jupyter_execution_timeout: 'executionTimeout',
+    jupyter_execution_timeout_tips: 'Please enter the execution timeout for each jupyter note cell',
+    jupyter_start_timeout: 'startTimeout',
+    jupyter_start_timeout_tips: 'Please enter the start timeout for jupyter kernel',
+    jupyter_others: 'others',
+    jupyter_others_tips: 'Please enter the other options you need for papermill',
     send_email: 'Send Email',
     log_display: 'Log display',
     rows_of_result: 'rows of result',

dolphinscheduler-ui/src/locales/modules/zh_CN.ts

@@ -921,6 +921,24 @@ const project = {
     zeppelin_note_id_tips: '请输入zeppelin note id',
     zeppelin_paragraph_id: 'zeppelin_paragraph_id',
     zeppelin_paragraph_id_tips: '请输入zeppelin paragraph id',
+    jupyter_conda_env_name: 'condaEnvName',
+    jupyter_conda_env_name_tips: '请输入papermill所在的conda环境名',
+    jupyter_input_note_path: 'inputNotePath',
+    jupyter_input_note_path_tips: '请输入jupyter note的输入路径',
+    jupyter_output_note_path: 'outputNotePath',
+    jupyter_output_note_path_tips: '请输入jupyter note的输出路径',
+    jupyter_parameters: 'parameters',
+    jupyter_parameters_tips: '请输入jupyter parameterization参数',
+    jupyter_kernel: 'kernel',
+    jupyter_kernel_tips: '请输入jupyter kernel名',
+    jupyter_engine: 'engine',
+    jupyter_engine_tips: '请输入引擎名称',
+    jupyter_execution_timeout: 'executionTimeout',
+    jupyter_execution_timeout_tips: '请输入jupyter note cell的执行最长时间',
+    jupyter_start_timeout: 'startTimeout',
+    jupyter_start_timeout_tips: '请输入jupyter kernel的启动最长时间',
+    jupyter_others: 'others',
+    jupyter_others_tips: '请输入papermill的其他参数',
     send_email: '发送邮件',
     log_display: '日志显示',
     rows_of_result: '行查询结果',

dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts

@@ -62,3 +62,4 @@ export { useConditions } from './use-conditions'
 export { useDependent } from './use-dependent'
 export { useEmr } from './use-emr'
 export { useZeppelin } from './use-zeppelin'
+export { useJupyter } from './use-jupyter'

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-jupyter.ts

@@ -0,0 +1,180 @@
+/*
+ * 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 { useCustomParams } from '.'
+import type { IJsonItem } from '../types'
+
+export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
+  const { t } = useI18n()
+
+  return [
+    {
+      type: 'input',
+      field: 'condaEnvName',
+      name: t('project.node.jupyter_conda_env_name'),
+      props: {
+        placeholder: t('project.node.jupyter_conda_env_name_tips')
+      },
+      validate: {
+        trigger: ['input', 'blur'],
+        required: true,
+        validator(validate: any, value: string) {
+          if (!value) {
+            return new Error(t('project.node.jupyter_conda_env_name_tips'))
+          }
+        }
+      }
+    },
+    {
+      type: 'input',
+      field: 'inputNotePath',
+      name: t('project.node.jupyter_input_note_path'),
+      props: {
+        placeholder: t('project.node.jupyter_input_note_path_tips')
+      },
+      validate: {
+        trigger: ['input', 'blur'],
+        required: true,
+        validator(validate: any, value: string) {
+          if (!value) {
+            return new Error(t('project.node.jupyter_input_note_path_tips'))
+          }
+        }
+      }
+    },
+    {
+      type: 'input',
+      field: 'outputNotePath',
+      name: t('project.node.jupyter_output_note_path'),
+      props: {
+        placeholder: t('project.node.jupyter_output_note_path_tips')
+      },
+      validate: {
+        trigger: ['input', 'blur'],
+        required: true,
+        validator(validate: any, value: string) {
+          if (!value) {
+            return new Error(t('project.node.jupyter_output_note_path_tips'))
+          }
+        }
+      }
+    },
+    {
+      type: 'input',
+      field: 'parameters',
+      name: t('project.node.jupyter_parameters'),
+      props: {
+        placeholder: t('project.node.jupyter_parameters_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_parameters_tips'))
+//           }
+//         }
+//       }
+    },
+    {
+      type: 'input',
+      field: 'kernel',
+      name: t('project.node.jupyter_kernel'),
+      props: {
+        placeholder: t('project.node.jupyter_kernel_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_kernel_tips'))
+//           }
+//         }
+//       }
+    },
+    {
+      type: 'input',
+      field: 'engine',
+      name: t('project.node.jupyter_engine'),
+      props: {
+        placeholder: t('project.node.jupyter_engine_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_engine_tips'))
+//           }
+//         }
+//       }
+    },
+    {
+      type: 'input',
+      field: 'executionTimeout',
+      name: t('project.node.jupyter_execution_timeout'),
+      props: {
+        placeholder: t('project.node.jupyter_execution_timeout_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_execution_timeout_tips'))
+//           }
+//         }
+//       }
+    },
+    {
+      type: 'input',
+      field: 'startTimeout',
+      name: t('project.node.jupyter_start_timeout'),
+      props: {
+        placeholder: t('project.node.zeppelin_note_id_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_start_timeout_tips'))
+//           }
+//         }
+//       }
+    },
+    {
+      type: 'input',
+      field: 'others',
+      name: t('project.node.jupyter_others'),
+      props: {
+        placeholder: t('project.node.jupyter_others_tips')
+      }
+//       validate: {
+//         trigger: ['input', 'blur'],
+//         required: false,
+//         validator(validate: any, value: string) {
+//           if (!value) {
+//             return new Error(t('project.node.jupyter_others_tips'))
+//           }
+//         }
+//       }
+    },
+    ...useCustomParams({ model, field: 'localParams', isSimple: false })
+  ]
+}

dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts

@@ -310,6 +310,18 @@ export function formatParams(data: INodeData): {
     taskParams.paragraphId = data.zeppelinParagraphId
   }
 
+  if (data.taskType === 'JUPYTER') {
+    taskParams.condaEnvName = data.condaEnvName
+    taskParams.inputNotePath = data.inputNotePath
+    taskParams.outputNotePath = data.outputNotePath
+    taskParams.parameters = data.parameters
+    taskParams.kernel = data.kernel
+    taskParams.engine = data.engine
+    taskParams.executionTimeout = data.executionTimeout
+    taskParams.startTimeout = data.startTimeout
+    taskParams.others = data.others
+  }
+
   if (data.taskType === 'PIGEON') {
     taskParams.targetJobName = data.targetJobName
   }
@@ -521,22 +533,6 @@ export function formatModel(data: ITaskData) {
     params.others = data.taskParams.sparkParameters.others
   }
 
-  if (data.taskParams?.jobFlowDefineJson) {
-    params.jobFlowDefineJson = data.taskParams.jobFlowDefineJson
-  }
-
-  if (data.taskParams?.zeppelinNoteId) {
-    params.zeppelinNoteId = data.taskParams.zeppelinNoteId
-  }
-
-  if (data.taskParams?.zeppelinParagraphId) {
-    params.zeppelinParagraphId = data.taskParams.zeppelinParagraphId
-  }
-
-  if (data.taskParams?.processDefinitionCode) {
-    params.processDefinitionCode = data.taskParams.processDefinitionCode
-  }
-
   if (data.taskParams?.conditionResult?.successNode?.length) {
     params.successBranch = data.taskParams.conditionResult.successNode[0]
   }

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts

@@ -34,6 +34,7 @@ import { useDependent } from './use-dependent'
 import { useDataQuality } from './use-data-quality'
 import { useEmr } from './use-emr'
 import { useZeppelin } from './use-zeppelin'
+import { useJupyter } from './use-jupyter'
 
 export default {
   SHELL: useShell,
@@ -54,5 +55,6 @@ export default {
   DEPENDENT: useDependent,
   DATA_QUALITY: useDataQuality,
   EMR: useEmr,
-  ZEPPELIN: useZeppelin
+  ZEPPELIN: useZeppelin,
+  JUPYTER: useJupyter
 }

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-jupyter.ts

@@ -0,0 +1,80 @@
+/*
+ * 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, ITaskData } from '../types'
+
+export function useJupyter({
+  projectCode,
+  from = 0,
+  readonly,
+  data
+}: {
+  projectCode: number
+  from?: number
+  readonly?: boolean
+  data?: ITaskData
+}) {
+  const model = reactive({
+    name: '',
+    taskType: 'JUPYTER',
+    flag: 'YES',
+    description: '',
+    timeoutFlag: false,
+    localParams: [],
+    environmentCode: null,
+    failRetryInterval: 1,
+    failRetryTimes: 0,
+    workerGroup: 'default',
+    delayTime: 0,
+    timeout: 30
+  } 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.useJupyter(model),
+      Fields.usePreTasks()
+    ] as IJsonItem[],
+    model
+  }
+}

dolphinscheduler-ui/src/views/projects/task/components/node/types.ts

@@ -290,6 +290,14 @@ interface ITaskParams {
   zeppelinParagraphId?: string
   noteId?: string
   paragraphId?: string
+  condaEnvName?: string
+  inputNotePath?: string
+  outputNotePath?: string
+  parameters?: string
+  kernel?: string
+  engine?: string
+  executionTimeout?: string
+  startTimeout?: string
   processDefinitionCode?: number
   conditionResult?: {
     successNode?: number[]

dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts

@@ -34,6 +34,7 @@ export type TaskType =
   | 'SEATUNNEL'
   | 'EMR'
   | 'ZEPPELIN'
+  | 'JUPYTER'
 
 export const TASK_TYPES_MAP = {
   SHELL: {
@@ -98,5 +99,9 @@ export const TASK_TYPES_MAP = {
   ZEPPELIN: {
     alias: 'ZEPPELIN',
     helperLinkDisable: true
+  },
+  JUPYTER: {
+    alias: 'JUPYTER',
+    helperLinkDisable: true
   }
 } as { [key in TaskType]: { alias: string; helperLinkDisable?: boolean } }

dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss

@@ -158,6 +158,9 @@ $bgLight: #ffffff;
     &.icon-zeppelin {
       background-image: url('/images/task-icons/zeppelin.png');
     }
+    &.icon-jupyter {
+      background-image: url('/images/task-icons/jupyter.png');
+    }
   }
 
   &:hover {
@@ -219,6 +222,9 @@ $bgLight: #ffffff;
       &.icon-zeppelin {
         background-image: url('/images/task-icons/zeppelin_hover.png');
       }
+      &.icon-jupyter {
+        background-image: url('/images/task-icons/jupyter_hover.png');
+      }
     }
   }
 }