DolphinScheduler API服务：RESTful接口完整指南

概述

Apache DolphinScheduler作为现代化的数据编排平台，提供了强大而全面的RESTful API服务，支持开发者通过编程方式管理和调度工作流。本文将深入解析DolphinScheduler的API架构、核心接口和使用方法，帮助开发者快速掌握API集成技巧。

API架构设计

整体架构

DolphinScheduler采用分层架构设计，API服务位于系统的最上层，提供统一的RESTful接口：

graph TB
    A[客户端应用] --> B[RESTful API]
    B --> C[业务逻辑层]
    C --> D[数据访问层]
    D --> E[数据库/存储]
    C --> F[任务调度引擎]
    F --> G[Worker节点]

核心组件

组件	功能描述	对应Controller
项目管理	项目创建、查询、更新、删除	ProjectController
工作流管理	工作流定义、实例管理	WorkflowDefinitionController
任务管理	任务定义、实例监控	TaskDefinitionController
资源管理	文件、目录资源操作	ResourcesController
调度管理	定时调度配置	SchedulerController
监控统计	系统状态监控	MonitorController

认证与授权

Token认证机制

DolphinScheduler支持基于Token的认证方式，所有API请求都需要在Header中包含有效的认证信息：

GET /api/v2/projects HTTP/1.1
Authorization: Bearer {access_token}
Content-Type: application/json

Token管理接口

// 创建Token
POST /v2/access-tokens
{
  "userId": 1,
  "expireTime": "2024-12-31 23:59:59",
  "token": "custom_token_string"
}

// 查询Token列表  
GET /v2/access-tokens?pageNo=1&pageSize=10

// 删除Token
DELETE /v2/access-tokens/{id}

核心API接口详解

项目管理接口

项目管理是DolphinScheduler的基础，所有工作流和任务都在特定项目下运行：

// 创建项目
POST /v2/projects
{
  "projectName": "数据分析平台",
  "description": "企业级数据分析工作流项目"
}

// 查询项目列表
GET /v2/projects?pageNo=1&pageSize=20&searchVal=分析

// 更新项目信息  
PUT /v2/projects/{projectCode}
{
  "projectName": "更新后的项目名称",
  "description": "更新描述信息"
}

// 删除项目
DELETE /v2/projects/{projectCode}

工作流管理接口

工作流是DolphinScheduler的核心概念，支持复杂的任务依赖关系：

// 创建工作流定义
POST /projects/{projectCode}/workflow-definition
{
  "name": "每日数据ETL流程",
  "description": "每日定时执行的数据抽取转换加载流程",
  "globalParams": [
    {
      "prop": "bizDate",
      "value": "${system.biz.date}"
    }
  ],
  "tasks": [
    {
      "name": "数据抽取",
      "taskType": "SQL",
      "params": {
        "type": "MYSQL",
        "datasource": 1,
        "sql": "SELECT * FROM source_table WHERE date = '${bizDate}'"
      }
    }
  ]
}

// 触发工作流执行
POST /projects/{projectCode}/workflow-definition/{code}/execution
{
  "failureStrategy": "CONTINUE",
  "warningType": "NONE",
  "execType": "START_PROCESS",
  "startNodeList": "",
  "taskDependType": "TASK_POST",
  "runMode": "RUN_MODE_SERIAL",
  "processInstancePriority": "MEDIUM",
  "workerGroup": "default",
  "environmentCode": -1
}

// 查询工作流实例
GET /v2/workflow-instances?pageNo=1&pageSize=10&projectCode={projectCode}

任务管理接口

任务是最小的执行单元，支持多种任务类型：

// 创建任务定义
POST /projects/{projectCode}/task-definition
{
  "name": "Python数据处理任务",
  "taskType": "PYTHON",
  "description": "使用Python进行数据清洗",
  "params": {
    "rawScript": "import pandas as pd\nimport numpy as np\n# 数据处理逻辑",
    "resourceList": [],
    "localParams": [],
    "dependence": {},
    "conditionResult": {},
    "waitStartTimeout": {}
  }
}

// 查询任务实例状态
GET /v2/projects/{projectCode}/task-instances?pageNo=1&pageSize=20&processInstanceId={instanceId}

// 停止任务实例
POST /task-instances/{id}/stop

高级功能API

数据源管理

// 创建数据源
POST /datasources
{
  "name": "生产MySQL数据库",
  "type": "MYSQL",
  "connectionParams": {
    "address": "jdbc:mysql://localhost:3306/prod_db",
    "database": "prod_db",
    "user": "admin",
    "password": "encrypted_password"
  }
}

// 测试数据源连接
POST /datasources/{id}/connect-test

资源文件管理

// 上传资源文件
POST /resources/upload
Content-Type: multipart/form-data

// 创建目录
POST /resources/directory
{
  "name": "scripts",
  "description": "Python脚本目录",
  "pid": -1
}

// 查询资源列表
GET /resources?type=FILE&pageNo=1&pageSize=50

调度配置

// 创建调度配置
POST /v2/schedules
{
  "processDefinitionCode": 123456,
  "startTime": "2024-01-01 00:00:00",
  "endTime": "2024-12-31 23:59:59",
  "crontab": "0 0 2 * * ?",
  "timezoneId": "Asia/Shanghai",
  "failureStrategy": "CONTINUE",
  "warningType": "NONE",
  "warningGroupId": 1
}

错误处理与状态码

统一响应格式

所有API接口都返回统一的JSON响应格式：

{
  "code": 10000,
  "msg": "success",
  "data": {
    "totalList": [...],
    "currentPage": 1,
    "totalPage": 5,
    "total": 100
  }
}

常见状态码

状态码	含义	描述
10000	成功	操作执行成功
10001	内部错误	服务器内部错误
10002	参数错误	请求参数验证失败
10003	未授权	用户认证失败
10004	权限不足	用户没有操作权限
10005	资源不存在	请求的资源不存在

最佳实践

1. 批量操作优化

// 批量创建任务
List<TaskDefinition> tasks = new ArrayList<>();
for (int i = 0; i < 100; i++) {
    TaskDefinition task = new TaskDefinition();
    task.setName("task_" + i);
    task.setTaskType("SHELL");
    task.setParams(new HashMap<>());
    tasks.add(task);
}

// 使用线程池并行处理
ExecutorService executor = Executors.newFixedThreadPool(10);
tasks.forEach(task -> 
    executor.submit(() -> createTask(projectCode, task))
);

2. 异常重试机制

public <T> T executeWithRetry(Callable<T> callable, int maxRetries) {
    int retries = 0;
    while (retries < maxRetries) {
        try {
            return callable.call();
        } catch (Exception e) {
            retries++;
            if (retries >= maxRetries) {
                throw new RuntimeException("操作失败，重试次数耗尽", e);
            }
            // 指数退避策略
            long waitTime = (long) (Math.pow(2, retries) * 1000);
            try {
                Thread.sleep(waitTime);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("操作被中断", ie);
            }
        }
    }
    throw new RuntimeException("无法完成操作");
}

3. 性能监控

// API调用性能监控
@Aspect
@Component
public class ApiPerformanceAspect {
    
    @Around("execution(* org.apache.dolphinscheduler.api.controller..*.*(..))")
    public Object monitorPerformance(ProceedingJoinPoint joinPoint) throws Throwable {
        long startTime = System.currentTimeMillis();
        try {
            return joinPoint.proceed();
        } finally {
            long duration = System.currentTimeMillis() - startTime;
            String methodName = joinPoint.getSignature().getName();
            // 记录到监控系统
            monitorService.recordApiCall(methodName, duration);
        }
    }
}

安全注意事项

1. 敏感信息保护

// 使用环境变量或配置中心管理敏感信息
@Value("${datasource.password}")
private String encryptedPassword;

// 避免在日志中记录敏感数据
@Slf4j
public class SecureLogger {
    public void logSensitiveOperation(String operation, Map<String, Object> params) {
        Map<String, Object> sanitizedParams = params.entrySet().stream()
            .collect(Collectors.toMap(
                Map.Entry::getKey,
                e -> isSensitive(e.getKey()) ? "***" : e.getValue()
            ));
        log.info("操作: {}, 参数: {}", operation, sanitizedParams);
    }
    
    private boolean isSensitive(String key) {
        return key.contains("password") || key.contains("token") || key.contains("secret");
    }
}

2. 请求限流

// 使用Guava RateLimiter进行限流
private final RateLimiter rateLimiter = RateLimiter.create(100.0); // 100 QPS

@PostMapping("/api/operation")
public Result performOperation(@RequestBody OperationRequest request) {
    if (!rateLimiter.tryAcquire()) {
        return Result.error(429, "请求过于频繁，请稍后重试");
    }
    // 正常处理逻辑
}

总结

DolphinScheduler的RESTful API提供了完整的工作流管理能力，通过合理的架构设计和丰富的接口功能，能够满足各种复杂的调度场景需求。在实际使用中，建议：

1. 合理使用分页：大数据量查询时务必使用分页参数
2. 实现重试机制：网络不稳定时自动重试失败的操作
3. 监控API性能：建立完善的监控体系跟踪API调用情况
4. 安全第一：妥善管理认证信息，实施适当的访问控制

通过本文的指南，开发者可以快速掌握DolphinScheduler API的使用方法，构建稳定可靠的数据调度平台集成方案。