DolphinScheduler 3.1.3 跨越升级 3.4.1：基于 API 的自动化迁移方案

一、背景：为什么要跨越大版本？

旧环境情况

当前 DolphinScheduler 版本：3.1.3
当前 seatunnel 版本：2.1.3

部署规模：1台Master + 2台Worker，工作流定义3700+，日均调度任务量20000+

使用年限：已经稳定运行3年

升级驱动因素

1. 功能需求：因业务需求增大，当前版本的架构设计缺陷、元数据库处理限制以及服务器资源不足；
2. 社区支持：官方推荐使用最新稳定版本，获得更好的技术支持
3. 性能优化：3.4.1版本在调度性能和稳定性方面有显著提升

为什么不用官方升级方案

1. 版本跨度太大：3.1.3 → 3.2.0 → 3.3.0 → 3.4.1，需要多次中间升级
2. 生产环境限制：无法接受多次停机窗口，业务连续性要求高
3. 架构变更风险：资源中心重构、元数据库表结构变更等存在较高迁移风险
4. 工作量考量：任务数量多，手动重建工作流工作量巨大，需要自动化方案

二、整体迁移思路：绕开官方升级路径，采用「重建+API」方案

2.1 核心思想

不追求「逐步升级」，而是「换新环境+数据迁移」：

• 旧环境（3.1.3）继续稳定运行，不影响业务
• 新环境（3.4.1）全新部署，保证架构干净
• 编写脚本代码通过旧版本API读取工作流定义和任务配置
• 通过新版本的API批量创建工作流
• 切换业务任务到新环境

2.2 方案优势与风险对比

三、详细实施步骤

3.1 环境准备阶段

3.1.1 新环境部署

• 搭建全新的3.4.1版本环境
• 配置数据库、Registry等依赖，弃用zk注册中心，改用Jdbc
• 配置必要的组件：datax、seatunnel等
• 了解新版本的特性，如个人关注点在：
  • 集成seatunnel 2.1.3，启动方式从引擎：从spark改成start-seatunnel-spark.sh；
  • 默认参数配。如租户、工作组、环境等，在项目管理设置里面配置偏好；
  • 参数传递变化。新版必须在下游节点添加“IN”类型的参数，才可以接收上游变量值
• 验证新环境基本功能正常，另外选取代表性任务手动创建执行验证

3.1.2 API访问配置

• 在新环境中配置API访问权限：令牌管理创建新的令牌
• 获取管理员token用于API调用
• 验证API接口连通性

3.2 元数据库基础数据配置

• 表数据复制：按照旧版本元数据表的基本配置，复刻基础数据到新元数据库表（包括id），这样在开发还原任务脚本的时候能减少很多改动。

3.3 数据迁移脚本开发

3.3.1 前期准备测试

1. 先对自己的任务分类：模板化的任务，非模板化的任务；
2. 选取代表性的任务，在新版本上创建运行，保证任务的正常运行，同步后的表的数据正常。

3.3.2 代码开发-读取原任务定义

... //获取整个， 循环处理 String processDefinitionUrl = OLD_URL + "/dolphinscheduler/projects/" + oldProjectCode + "/process-definition/query-process-definition-list"; Map<String, String> map = new HashMap<>(); map.put("projectCode", oldProjectCode); String pdRes = httpClientUtilOld.doGetRequest(processDefinitionUrl, map); ArrayList<JSONObject> dataList = parseResDataToList(pdRes); for (JSONObject job : dataList) { String oldWFCode = job.get("code").toString(); Map<String, String> mapPara = new HashMap<>(); String oldurl = OLD_URL + "/dolphinscheduler/projects/" + oldProjectCode + "/process-definition/" + oldWFCode; mapPara.put("code", oldWFCode); mapPara.put("projectCode", oldProjectCode); String res = httpClientUtilOld.doGetRequest(oldurl, mapPara); JSONObject jsonObject = JSON.parseObject(res); JSONObject data = (JSONObject) jsonObject.get("data"); JSONObject processDefinition = data.getJSONObject("processDefinition"); JSONArray processTaskRelationList = data.getJSONArray("processTaskRelationList"); JSONArray taskDefinitionList = data.getJSONArray("taskDefinitionList"); //todo 获取 n个编码，替换旧任务编码 //填充 信息创建任务 createWF(processDefinition, processTaskRelationList, taskDefinitionList, NEW_IP, newProjectCode);

3.3.3 API创建新工作流

//看原工作流的任务有几个 获取任务编码 int taskCnt = taskDefinitionList.size(); List<String> taskCodeList = taskDefinitionList.stream() .map(obj -> (JSONObject) obj) .map(obj -> obj.getString("code")) .collect(Collectors.toList()); try { // todo 获取 任务编码 String taskCodeUrl = NEW_URL + "/dolphinscheduler/projects/" + newProjectCode + "/task-definition/gen-task-codes"; HashMap<String, String> taskCodeMap = new HashMap<>(); //需要创建n个任务 taskCodeMap.put("genNum", String.valueOf(taskCnt)); String codeData = httpClientUtilNew.doGetRequest(taskCodeUrl, taskCodeMap); Object codes = JSON.parseObject(codeData).get("data"); JSONArray taskCodeArr = JSON.parseArray(codes.toString()); //根据实际任务 添加任务下游接收参数 for (int i = 0; i < taskDefinitionList.size(); i++) { JSONObject logTask = (JSONObject) taskDefinitionList.get(i); if (“判断逻辑”)) { JSONObject taskParams = logTask.getJSONObject("taskParams"); JSONArray localParams = taskParams.getJSONArray("localParams"); // 构造 hiveAmount JSONObject hiveParam = new JSONObject(); hiveParam.put("prop", "hiveAmount"); hiveParam.put("direct", "IN"); hiveParam.put("type", "VARCHAR"); hiveParam.put("value", ""); localParams.add(hiveParam); logTask.put("taskParamList", localParams); JSONObject paramMap = new JSONObject(); for (Object obj : localParams) { JSONObject param = (JSONObject) obj; paramMap.put(param.getString("prop"), param.getString("value")); } logTask.put("taskParamMap", paramMap); .... // 替换必要参数 code 和seatunnel执行引擎参数 for (int i = 0; i < taskCodeList.size(); i++) { String oldCode = taskCodeList.get(i); String newCode = taskCodeArr.getString(i); //替换编码 // 替换sea引擎："SPARK" "start-seatunnel-spark.sh" taskDefinitionListJsonStr = taskDefinitionListJsonStr .replace("\"code\":" + oldCode + ",", "\"code\":" + newCode + ",") .replace("\"engine\":\"SPARK\",", "\"startupScript\":\"start-seatunnel-spark.sh\","); taskRelationListJsonStr = taskRelationListJsonStr .replace("TaskCode\":" + oldCode + ",", "TaskCode\":" + newCode + ","); locationsJsonStr = locationsJsonStr .replace(oldCode, newCode); ... } Map<String, String> map = new HashMap<>(); map.put("taskDefinitionJson", taskDefinitionListJsonStr); map.put("taskRelationJson", taskRelationListJsonStr); map.put("locations", locationsJsonStr); map.put("name", processDefinition.getString("name")); map.put("tenantCode", "omm");//processDefinition.getString("tenantCode") map.put("executionType", processDefinition.getString("executionType")); map.put("description", processDefinition.getString("description") == null ? "" : processDefinition.getString("description")); map.put("globalParams", processDefinition.getString("globalParams")); map.put("timeout", processDefinition.getString("timeout")); String processDefinitionUrl = NEW_URL + "/dolphinscheduler/projects/" + newProjectCode + "/workflow-definition"; String processDefinitionRes = httpClientUtilNew.doPostRequest(processDefinitionUrl, map);

3.4 迁移执行阶段

3.4.1 流程

1. 备份旧版本调度表：t_ds_schedules_20260416_10
2. 选择试点项目：选择一个任务量适中、业务影响小的项目作为试点
3. 迁移任务定义：迁移200个任务定义（包括定时配置）
4. 上线任务不上线定时：先上线任务定义，但不启用定时调度
5. 手动提交验证：分批手动提交运行，注意和原集群的任务是否冲突（因为大部分是天/时/15分钟定时，一般不冲突）
6. 核查处理失败任务：分析失败原因，修复问题后重新运行
7. 上线定时配置：所有任务验证无误后，上线这些任务的定时
8. 下线原集群定时：确认新环境稳定运行后，下线原集群对应任务的定时
9. 分项目、分批次迁移

3.4.2 迁移实施

按照流程，先找个任务少并且有代表性的项目（199个任务），迁移后试运行一段时间，测试稳定性运行一周没有问题。后续迁移完成50个项目，总共约3700个任务，耗时10天左右

• 表格记录

3.4.3 运行情况

目前已经运行了近1个月时间。以前调度延迟的问题：几乎是10秒到1分钟甚至更严重，现在几乎没有调度延迟；调度缺失实例的问题，目前也未发生。

目前一切运行良好，暂未发现问题，后续持续观察。

四、风险控制与应急预案

4.1 主要风险点

数据丢失风险：迁移过程中可能遗漏部分配置

兼容性问题：旧版本特有配置在新版本不支持

业务中断风险：切换期间可能出现调度延迟

4.2 应急预案

4.2.1 回滚方案

• 立即停止新环境调度
• 恢复旧环境调度服务
• 分析问题原因，修复后重新尝试

4.2.2 数据备份

• 完整备份旧环境数据库
• 备份新环境初始配置