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

一、背景为什么要跨越大版本旧环境情况当前 DolphinScheduler 版本3.1.3当前 seatunnel 版本2.1.3部署规模1台Master 2台Worker工作流定义3700日均调度任务量20000使用年限已经稳定运行3年升级驱动因素功能需求因业务需求增大当前版本的架构设计缺陷、元数据库处理限制以及服务器资源不足社区支持官方推荐使用最新稳定版本获得更好的技术支持性能优化3.4.1版本在调度性能和稳定性方面有显著提升为什么不用官方升级方案版本跨度太大3.1.3 → 3.2.0 → 3.3.0 → 3.4.1需要多次中间升级生产环境限制无法接受多次停机窗口业务连续性要求高架构变更风险资源中心重构、元数据库表结构变更等存在较高迁移风险工作量考量任务数量多手动重建工作流工作量巨大需要自动化方案二、整体迁移思路绕开官方升级路径采用「重建API」方案2.1 核心思想不追求「逐步升级」而是「换新环境数据迁移」旧环境3.1.3继续稳定运行不影响业务新环境3.4.1全新部署保证架构干净编写脚本代码通过旧版本API读取工作流定义和任务配置通过新版本的API批量创建工作流切换业务任务到新环境2.2 方案优势与风险对比维度官方升级方案本方案重建API停机时间多次升级累计停机可能数小时甚至是数天切换时停旧任务启动新任务基本是无缝衔接】回滚难度困难数据库已变更简单旧环境完好随时切回数据一致性需验证所有表结构迁移只迁移核心业务数据工作流定义历史执行记录不迁移版本适配需处理中间版本的所有兼容性问题直接适配3.4.1只改造必要参数工作量需多次验证主要集中在脚本开发适用场景小版本升级大版本跨越、任务量大三、详细实施步骤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这样在开发还原任务脚本的时候能减少很多改动。核对事项表租户表t_ds_tenant项目表t_ds_project用户表t_ds_user环境表t_ds_environmentt_ds_environment_worker_group_relation工作组t_ds_worker_group数据源表。t_ds_datasource3.3 数据迁移脚本开发3.3.1 前期准备测试先对自己的任务分类模板化的任务非模板化的任务选取代表性的任务在新版本上创建运行保证任务的正常运行同步后的表的数据正常。3.3.2 代码开发-读取原任务定义... //获取整个 循环处理 String processDefinitionUrl OLD_URL /dolphinscheduler/projects/ oldProjectCode /process-definition/query-process-definition-list; MapString, String map new HashMap(); map.put(projectCode, oldProjectCode); String pdRes httpClientUtilOld.doGetRequest(processDefinitionUrl, map); ArrayListJSONObject dataList parseResDataToList(pdRes); for (JSONObject job : dataList) { String oldWFCode job.get(code).toString(); MapString, 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(); ListString 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; HashMapString, 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); ... } MapString, 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 流程备份旧版本调度表t_ds_schedules_20260416_10选择试点项目选择一个任务量适中、业务影响小的项目作为试点迁移任务定义迁移200个任务定义包括定时配置上线任务不上线定时先上线任务定义但不启用定时调度手动提交验证分批手动提交运行注意和原集群的任务是否冲突因为大部分是天/时/15分钟定时一般不冲突核查处理失败任务分析失败原因修复问题后重新运行上线定时配置所有任务验证无误后上线这些任务的定时下线原集群定时确认新环境稳定运行后下线原集群对应任务的定时分项目、分批次迁移3.4.2 迁移实施按照流程先找个任务少并且有代表性的项目199个任务迁移后试运行一段时间测试稳定性运行一周没有问题。后续迁移完成50个项目总共约3700个任务耗时10天左右表格记录序号项目名项目code任务数处理进度备注1项目113*******1990416完成2项目2..............................50...............3.4.3 运行情况目前已经运行了近1个月时间。以前调度延迟的问题几乎是10秒到1分钟甚至更严重现在几乎没有调度延迟调度缺失实例的问题目前也未发生。目前一切运行良好暂未发现问题后续持续观察。四、风险控制与应急预案4.1 主要风险点数据丢失风险迁移过程中可能遗漏部分配置兼容性问题旧版本特有配置在新版本不支持业务中断风险切换期间可能出现调度延迟4.2 应急预案4.2.1 回滚方案立即停止新环境调度恢复旧环境调度服务分析问题原因修复后重新尝试4.2.2 数据备份完整备份旧环境数据库备份新环境初始配置