process_definition_import_tool

Dolphinscheduler 外部作业导入工具(目前支持版本dev:3.3+)

• 用途
• 代码概述
• 特性
• 生成过程

---

用途

将通用的外部定义的作业配置导入到 Dolphinscheduler 指定的项目中, 目前支持版本3.3.0+, 主要适用于作业迁移和开发/测试与生产隔离的情况.

• 从excel离线导入增量配置, 几个要点:

  1. 作业的shell节点必须在项目的整个DAG中保持唯一, 用作业所在流程的完整路径(流程/子流程/子流程...../任务节点)匹配作业
  2. 上线时不能再对生产环境的调度进行开发类配置操作, 符合金融机构环境管理要求

• 由于DS2.0版本后使用了特定code作为实际id的机制, 无法像1.3.x一样直接操作数据库, 当前版本通过调用DS的 API实现项目的导入

  ** dependent任务和sub_workflow任务的参数中,指向其他作业的code在导入时无法自动映射, 在生成dependent任务和sub_workflow任务时, 引用的流程必须已经存在 **

  ** 因为数仓任务涉及业务连续性问题, 不允许跨日执行, 大部分作业节点必须有前一天的任务自依赖, 但迁移后首次执行时不能有依赖节点, 否则会等待不存在的任务实例(空跑仍然会等待依赖节点) **

• 以上问题使用以下解决办法: ** 直接在数据库中生成上一天的伪实例，用于下一天的依赖检测 **

• 同时更新作业定义时, 已存在的作业定义必须保留, 否则自依赖引用会失效, 使用以下流程实现完全增量更新:

1. 先导入服务器中的所有作业，同时导入excel配置中的作业定义(导入时检测是否有重叠的节点,有重叠则更新)
2. 用名称匹配项目下已有的任务code, 不存在时生成新任务code
3. 遍历dag生成流程, 将流程定义更新到服务中

flowchart TB
    %% 定义 Domain
    subgraph domain[Domain]
        direction TB
        
        %% 子分组任务（引用子分组）
        subTask1[cbs]
        subTask2[子分组任务2]
        subTask3[子分组任务3]

    end

    %% 子分组 1 及其任务
    subgraph subgroup1[cbs]
        direction LR
        shell11(Shell 任务1.1)
        shell12(cbs_d_teller)
        dep13(Dependent 任务1.3)
        
        dep13 --> shell12
        shell11 --> shell12
    end

    %% 子分组 2 及其任务
    subgraph subgroup2[子分组2]
        direction LR
        shell21(Shell 任务2.1)
        dep22(Dependent 任务2.2)
        dep22 --> shell21
    end

    %% 子分组 3 及其任务
    subgraph subgroup3[子分组3]
        direction LR
        dep31(Dependent 任务3.1)
        shell32(Shell 任务3.2)
        dep31 --> shell32
    end

    %% 跨分组依赖关系
    dep22 -.-> shell11
    dep31 -.-> shell21


    subTask1 -.-> subgroup1[子分组1]
    subTask2 -.-> subgroup2[子分组2]
    subTask3 -.-> subgroup3[子分组3]

Loading

代码概述

• main_general.py 主程序, 负责导入服务器及excel的作业,生成DAG树.
• ds_config.py DS服务的口令及地址
• ds_api.py http api的封装
• ds_db.py 简单的数据库操作函数
• name_standardizer.py 名称标准化函数, 依赖excel中的配置
• schedule_config.py 用于导入excel配置
• dag_node 实现DAG到流和作业的json的转换, 以及对比差异

特性

** 由于数仓作业种类多且杂且 DS目前不支持项目级的作业操作, 每个作业一个流程会导致任务数巨大且触发配置困难, 上万个有复杂依赖关系的任务只能分层切分至多个流程组, 通过子流程完成整体任务的触发 **

• 增加多项目导入, 支持项目间依赖的定义
• 带定时且定时状态为上线的流程更新后自动上线
• 非自动生成的流程可配置子流程路径
• 增加依赖关系的依赖级别定义
• 对比差异增加中文描述
• 按项目区分分组定义
• 增加工作组和环境配置
• 增加项目保护及自定义路径导入
• 分组的 任务的依赖 会上升为同级别, 但完全由模板控制

生成过程

1. DS中添加 租户, 用户, 项目 和 环境
2. 配置 ds_config.py(配置口令及服务IP)
3. 执行 python/python3 main_general.py [excel(版本1.4)配置文件路径]