数据 ETL 调度引擎对接（二次开发）

1. 背景与原理

数据 ETL 模块在平台上提供 任务流编排能力：拖拉拽算子、配置每个节点、连出上下游关系，编排结果以 dag_json 形式存进数据库（etl_pipeline 表）。

但 cube-studio 本身只负责编排，并不直接向远程调度系统创建/删除任务。要真正把编排出来的 DAG 提交到 Airflow / Azkaban / DolphinScheduler 等调度系统执行，需要二次开发实现一个 调度引擎（workflow） 适配，完成：

• 向远程调度系统创建任务、删除任务；
• 把远程任务的信息记录回本地数据库。

2. 调度引擎的派发约定（关键）

每个 ETL 任务流（ETL_Pipeline）有一个 workflow 字段（界面标签为"计算引擎"，模型注释为"调度引擎"，默认 data_analysis，见 myapp/models/model_etl_pipeline.py:29、myapp/views/view_etl_pipeline.py:242-250）。

平台根据 workflow 的值，按固定命名约定动态加载 对应的实现类（myapp/views/view_etl_pipeline.py:479-480 等多处）：

import importlib
# 模块名 = view_etl_pipeline_<workflow>
params = importlib.import_module('myapp.views.view_etl_pipeline_' + pipeline.workflow)
# 类名   = <WORKFLOW 大写>_ETL_PIPELINE，构造参数是 pipeline 模型对象
etl_pipeline = getattr(params, pipeline.workflow.upper() + '_ETL_PIPELINE')(pipeline)

所以要新增一个调度引擎 xxx，约定是：

1. 新建文件 myapp/views/view_etl_pipeline_xxx.py；
2. 在其中定义类 XXX_ETL_PIPELINE，构造函数接收 pipeline（ETL_Pipeline 模型对象）；
3. 把 ['xxx', 'xxx'] 加进 workflow 字段的 choices（myapp/views/view_etl_pipeline.py:242-250），用户新建任务流时即可选择该引擎。

说明：当前代码库 不存在 view_etl_pipeline_airflow.py / view_etl_pipeline_azkaban.py / view_etl_pipeline_dolphinscheduler.py（已核实），只有两个内置引擎实现：

• myapp/views/view_etl_pipeline_etl.py（类 ETL_ETL_PIPELINE，是一个 Airflow 风格的对接模板/桩，默认 host='http://airflow.oa.com'，submit_pipeline / delete_pipeline 内部多为 todo，需二开补全）；
• myapp/views/view_etl_pipeline_data_analysis.py（类 DATA_ANALYSIS_ETL_PIPELINE，cube-studio 内置 的数据分析任务流 runner，已完整实现，跑在 K8s 上）。

当前 workflow 的 choices 也只有 etl 和 data_analysis 两项（myapp/views/view_etl_pipeline.py:248）。要对接 Airflow/Azkaban/DolphinScheduler，请按上面的命名约定新建文件并扩展 choices，把 view_etl_pipeline_etl.py 当作模板照着实现。

3. 一个调度引擎要实现哪些东西

实现类（如 ETL_ETL_PIPELINE）需要提供以下属性/方法，平台基类 view_etl_pipeline.py 会按下面的方式调用它们：

注意：pipeline_config_ui 被当作属性访问（无括号），view_etl_pipeline_etl.py 里是类属性 dict，view_etl_pipeline_data_analysis.py 里则用方法/属性形式提供，两种都可以。

3.1 all_template 结构

all_template 是一个 dict，关键字段（参考 myapp/views/view_etl_pipeline_etl.py:18 起）：

• templte_list：算子模板列表，按分组（group）组织；平台会遍历它给每个模板补 template_id / changed_on / created_on / username（view_etl_pipeline.py:528-535）。
  • ⚠️ 字段名是 templte_list（源码中的拼写，少一个 a），照抄即可，别"纠正"成 template_list。
• templte_common_ui_config：公共配置参数，每个任务节点都带（任务名、调度周期 crontab、队列、监控告警、超时等）。
• task_metadata_ui_config：任务元数据配置（如中文名 label）。
• message / status：固定返回字段。

3.2 submit_pipeline 返回值

submit_pipeline 返回二元组 (dag_json, redirect_url)，平台按 dag_json, redirect_url = etl_pipeline.submit_pipeline() 解包（view_etl_pipeline.py:570）。redirect_url 一般指向远程调度系统中该 DAG 的页面。模板里的示例实现（view_etl_pipeline_etl.py:977-980）：

def submit_pipeline(self):
    # todo: 检查任务是否存在；创建/修改/删除远程任务；发起远程调度
    return "", self.host + "/code?dag_id=%s" % self.pipeline.name

3.3 跳转/运行按钮结构

pipeline_jump_button / pipeline_run_button 返回按钮列表，每个按钮形如（view_etl_pipeline_etl.py:947-973）：

{
    "name": "调度实例",
    "action_url": self.host + "/tree?dag_id=%s" % self.pipeline.name,  # 跳转目标
    "icon_svg": "<svg ...>...</svg>"                                    # 按钮图标
}

其中"提交"按钮的 action_url 通常指向平台自身的提交接口 "/etl_pipeline_modelview/api/submit_etl_pipeline/<pipeline.id>"，由它再调用你的 submit_pipeline()。

4. 实现步骤小结

1. 复制 myapp/views/view_etl_pipeline_etl.py 为 view_etl_pipeline_<引擎名>.py，重命名类为 <引擎名大写>_ETL_PIPELINE；
2. 把 __init__ 里的 host 改成你的调度系统地址；
3. 按上表实现 all_template、pipeline_config_ui、pipeline_jump_button、pipeline_run_button、submit_pipeline、delete_pipeline；
4. 在 view_etl_pipeline.py:242-250 的 workflow choices 里加上该引擎；
5. 新建 ETL 任务流时选择该计算引擎，即会使用你的实现类。

相关源码