Apache DolphinScheduler 2.x API Reference and Java Code Examples for Workflow, Task, and Relation Management

Background: Apache DolphinScheduler 2.0 was released on Dec 17, 2021 after extensive refactoring, providing RESTful APIs documented at http://ip:port/dolphinscheduler/doc.html.

Workflow Definition API

There are 25 workflow definition interfaces; the most commonly used include createProcessDefinition, createEmptyProcessDefinition, update, updateBasicInfo, batchDeleteByCodes, deleteByCode, deleteVersion, release, releaseWorkflowAndSchedule, switchVersion, queryProcessDefinitionByCode, queryListPaging, queryAllByProjectCode. The table below lists each endpoint, HTTP method, and functionality.

接口名称

接口地址

请求方式

功能介绍

createProcessDefinition

/dolphinscheduler/projects/{projectCode}/process-definition

POST

创建具有任务的工作流，对应界面的工作流保存，其中 taskDefinitionJson 和 taskRelationJson 不可为空，以 json 的方式填写必要的信息

createEmptyProcessDefinition

/dolphinscheduler/projects/{projectCode}/process-definition/empty

POST

创建空工作流和定时，其中 scheduleJson 为空时，只创建空工作流而不创建定时

update

/dolphinscheduler/projects/{projectCode}/process-definition/{code

PUT

根据工作流 code 更新接口，taskDefinitionJson 和 taskRelationJson 不可为空，更新时可指定上下线状态，应用于更新后是否直接上线

updateBasicInfo

/dolphinscheduler/projects/{projectCode}/process-definition/{code}/basic-info

PUT

更新工作流的基本信息和定时信息，scheduleJson 为空时，只更新工作流基本信息

batchDeleteByCodes

/dolphinscheduler/projects/{projectCode}/process-definition/batch-delete

POST

根据工作流 code 批量删除工作流，其中 code 以英文逗号分割

deleteByCode

/dolphinscheduler/projects/{projectCode}/process-definition/{code}

DELETE

根据工作流 code 删除工作流

deleteVersion

/dolphinscheduler/projects/{projectCode}/process-definition/{code}/versions/{version}

DELETE

根据工作流 code 和 version 删除，只能删除非主表应用版本的数据

release

/dolphinscheduler/projects/{projectCode}/process-definition/{code}/release

POST

根据工作流 code 上下线工作流

releaseWorkflowAndSchedule

/dolphinscheduler/projects/{projectCode}/process-definition/{code}/release-workflow

POST

根据工作流 code 同时上下线工作流和定时

switchVersion

/dolphinscheduler/projects/{projectCode}/process-definition/{code}/versions/{version}

GET

根据工作流 code 和 version 切换到指定版本

queryProcessDefinitionByCode

/dolphinscheduler/projects/{projectCode}/process-definition/{code}

GET

根据工作流 code 查询工作流信息，包含任务和工作流任务关系

queryListPaging

/dolphinscheduler/projects/{projectCode}/process-definition

GET

分页查询工作流

queryAllByProjectCode

/dolphinscheduler/projects/{projectCode}/process-definition/all

GET

根据项目 code 查询该项目下所有工作流

Task Definition API

Task definition interfaces include save, update, deleteTaskDefinition, deleteVersion, switchVersion, genTaskCodeList, queryTaskDefinitionByCode, and queryTaskDefinitionListPaging. The table below summarizes each endpoint.

接口名称

接口地址

请求方式

功能介绍

save

/dolphinscheduler/projects/{projectCode}/task-definition

POST

创建任务的接口，taskDefinitionJson 必须是 JSON Array 的方式

update

/dolphinscheduler/projects/{projectCode}/task-definition/{code}

PUT

修改任务的接口，taskDefinitionJsonObj 必须是 JSON Object 的方式

deleteTaskDefinition

/dolphinscheduler/projects/{projectCode}/task-definition/{code}

DELETE

删除任务的接口，根据任务 code 进行删除

deleteVersion

/dolphinscheduler/projects/{projectCode}/task-definition/{code}/versions/{version}

DELETE

根据任务 code 和 version 删除，只能删除非主表应用版本的数据

switchVersion

/dolphinscheduler/projects/{projectCode}/task-definition/{code}/versions/{version}

GET

根据任务 code 和 version 切换到指定版本

genTaskCodeList

/dolphinscheduler/projects/{projectCode}/task-definition/gen-task-codes

GET

获取 taskCode，根据 genNum 可获取多个

queryTaskDefinitionByCode

/dolphinscheduler/projects/{projectCode}/task-definition/{code}

GET

根据任务 code 查询任务详情信息

queryTaskDefinitionListPaging

/dolphinscheduler/projects/{projectCode}/task-definition

GET

分页查询任务

Workflow‑Task Relation API

Endpoints for binding and unbinding tasks to workflows include save, deleteRelation, deleteDownstreamRelation, deleteUpstreamRelation, queryDownstreamRelation, and queryUpstreamRelation.

接口名称

接口地址

请求方式

功能介绍

save

/dolphinscheduler/projects/{projectCode}/process-task-relation

POST

工作流和任务绑定接口，支持绑定前置任务、绑定后置任务

deleteRelation

/dolphinscheduler/projects/{projectCode}/process-task-relation/{taskCode}

DELETE

工作流和任务解绑，当任务是条件分支、依赖任务、子工作流时同步删除任务

deleteDownstreamRelation

/dolphinscheduler/projects/{projectCode}/process-task-relation/{taskCode}/downstream

DELETE

删除任务的下游依赖，支持批量删除下游

deleteUpstreamRelation

/dolphinscheduler/projects/{projectCode}/process-task-relation/{taskCode}/upstream

DELETE

删除任务的上游依赖，支持批量删除上游

queryDownstreamRelation

/dolphinscheduler/projects/{projectCode}/process-task-relation/{taskCode}/downstream

GET

查询任务的下游依赖

queryUpstreamRelation

/dolphinscheduler/projects/{projectCode}/process-task-relation/{taskCode}/upstream

GET

查询任务的上游依赖

Calling the APIs from Code

API calls require a token, which can be generated via the UI or obtained through an endpoint.

Maven dependency for Apache HttpClient:

org.apache.httpcomponents
httpclient
4.5.6

Java code examples for sending POST and GET requests with the token header are shown below.

private static String DOLPHIN_BASE_URI = "http://ip:port";
private static String token = "xxx";
private static String sendPost(String uri, List
params) throws Exception {
    CloseableHttpClient httpclient = HttpClients.createDefault();
    CloseableHttpResponse response = null;
    try {
        UrlEncodedFormEntity formEntity = new UrlEncodedFormEntity(params, Consts.UTF_8);
        HttpPost httpPost = new HttpPost(DOLPHIN_BASE_URI + uri);
        httpPost.setEntity(formEntity);
        httpPost.setHeader("token", token);
        response = httpclient.execute(httpPost);
        return EntityUtils.toString(response.getEntity(), Consts.UTF_8);
    } catch (Exception e) {
        throw new Exception(String.format("[dolphin] The %s call failed", uri));
    } finally {
        try {
            if (response != null) {
                response.close();
            }
            httpclient.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
private static String sendGet(String uri, List
params) throws Exception {
    CloseableHttpClient httpclient = HttpClients.createDefault();
    CloseableHttpResponse response = null;
    try {
        HttpGet httpGet = new HttpGet(new URIBuilder(DOLPHIN_BASE_URI + uri).setParameters(params).build());
        httpGet.setHeader("token", token);
        response = httpclient.execute(httpGet);
        return EntityUtils.toString(response.getEntity(), Consts.UTF_8);
    } catch (Exception e) {
        throw new Exception(String.format("[dolphin] The %s call failed", uri));
    } finally {
        if (response != null) {
            response.close();
        }
        httpclient.close();
    }
}

Two Ways to Create a Workflow

(1) Using createProcessDefinition

Key parameters include name, projectCode, taskDefinitionJson, taskRelationJson, tenantCode, description, globalParams, and timeout. Example parameter table and Java code are provided.

参数

参数说明

实例值

name

工作流名称

lee-test-01

projectCode

项目 code，必须填写

4362891840832

taskDefinitionJson

task 所有信息组成 json Array，在这个接口中 task code 必须含有，task version 无需含有

[{"code":4143298469056,"name":"lee-test","description":"","delayTime":0,"taskType":"SHELL","taskParams":{"resourceList":[],"localParams":[],"rawScript":"echo 11333","dependence":{},"conditionResult":{"successNode":[],"failedNode":[]},"waitStartTimeout":{},"switchResult":{}},"flag":"YES","taskPriority":"MEDIUM","workerGroup":"default","failRetryTimes":0,"failRetryInterval":1,"timeoutFlag":"CLOSE","timeoutNotifyStrategy":"WARN","timeout":0,"environmentCode":-1}]

taskRelationJson

dag 上 task 关系描述，postTask 表述当前节点

[{"name":"","preTaskCode":0,"preTaskVersion":0,"postTaskCode":4143298469056,"conditionType":0,"conditionParams":{}}]

tenantCode

租户，对应于租户管理的

操作系统租户

root

globalParams

全局参数

[]

timeout

工作流超时时长

0

public static void main(String[] args) throws Exception {
    long projectCode = 4362891840832L;
    String uri = String.format("/dolphinscheduler/projects/%d/process-definition", projectCode);
    List
params = new ArrayList<>();
    params.add(new BasicNameValuePair("name", "lee-test-04"));
    params.add(new BasicNameValuePair("projectCode", projectCode + ""));
    String taskDefinitionJson = "[{\"code\":4143298469059,\"name\":\"lee-test-4\",\"description\":\"\",\"delayTime\":0,\"taskType\":\"SHELL\",\"taskParams\":{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo 11333\",\"dependence\":{},\"conditionResult\":{\"successNode\":[],\"failedNode\":[]},\"waitStartTimeout\":{},\"switchResult\":{}},\"flag\":\"YES\",\"taskPriority\":\"MEDIUM\",\"workerGroup\":\"default\",\"failRetryTimes\":0,\"failRetryInterval\":1,\"timeoutFlag\":\"CLOSE\",\"timeoutNotifyStrategy\":\"WARN\",\"timeout\":0,\"environmentCode\":-1}]";
    params.add(new BasicNameValuePair("taskDefinitionJson", taskDefinitionJson));
    params.add(new BasicNameValuePair("taskRelationJson", "[{\"name\":\"\",\"preTaskCode\":0,\"preTaskVersion\":0,\"postTaskCode\":4143298469059,\"conditionType\":0,\"conditionParams\":{}}]"));
    params.add(new BasicNameValuePair("tenantCode", "root"));
    params.add(new BasicNameValuePair("description", ""));
    params.add(new BasicNameValuePair("globalParams", "[]"));
    params.add(new BasicNameValuePair("timeout", "0"));
    sendPost(uri, params);
}

(2) Using createEmptyProcessDefinition, task save, and workflow‑task relation save

Parameter tables for createEmptyProcessDefinition, taskDefinition save, and processTaskRelation save are provided, illustrating how to assemble a workflow in multiple steps.

参数

参数说明

示例值

name

工作流名称

lee-test-01

projectCode

项目 code，必须填写

4362891840832

scheduleJson

创建定时，可为空

{"warningType":"NONE","warningGroupId":1,"failureStrategy":"CONTINUE","workerGroup":"prod","environmentCode":-1,"processInstancePriority":"MEDIUM","startTime":"2022-02-07 00:00:00","endTime":"2027-02-07 00:00:00","crontab":"0 11 11 * * ? *","timezoneId":"Asia/Shanghai"}

tenantCode

租户，对应于租户管理的

操作系统租户

root

globalParams

全局参数

[]

timeout

工作流超时时长

0