平台架构 - ML 任务流（Pipeline 编排）

拖拉拽的 pipeline 编排

CubeStudio 使用 argo 作为任务流 workflow，为了能降低用户的学习成本，开发了一套拖拉拽的任务流编排能力。并且在其中加入了很多分布式的任务模板，基本上的分布式框架都可以得到解决。各类分布式框架在分布式任务加速中也有介绍，详见 平台架构-分布式训练与加速.md。

模板变量

有些任务希望在任务实例运行时再产生某些变量的值，目前平台支持以下模板变量：

creator         = pipeline.created_by.username       字符串：pipeline 的创建者
datetime        = datetime                           包
runner          = g.user.username                    字符串：实例的运行者
uuid            = uuid                                包
pipeline_id     = pipeline.id                         字符串
pipeline_name   = pipeline.name                       字符串：pipeline 的名称
cluster_name    = pipeline.project.cluster['NAME']    字符串：当前所属 k8s 集群
execution_date  = eta.strftime('%Y-%m-%d %H:%M:%S')   字符串：实例的周期时间

ℹ️ 核实：模板变量的注入逻辑见 myapp/views/view_pipeline.py:427（pipeline 级渲染）与 myapp/views/view_task.py:602（单 task 级渲染）。代码中 runner 兜底逻辑为：若无登录用户则回退为 pipeline 创建者（g.user.username if g and g.user and g.user.username else pipeline.created_by.username）。除上述变量外，pipeline 级渲染还额外注入了 run_id=pipeline.run_id。execution_date 若未通过调度传入，则默认取当前时间（myapp/views/view_pipeline.py:420-421）。

使用示例：

{{uuid.uuid4().hex[:4]}}
{{datetime.datetime.now().strftime("%Y%m%d-%H%M%S.%f")}}
{{pipeline_id}}
{{pipeline_name}}
{{cluster_name}}
{{runner}}
{{execution_date}}

单个 task 调试

通过 notebook 我们可以进行代码的编辑，通过镜像调试可以构建自己的镜像，然后就可以编排 pipeline 了。pipeline 运行前需要独立运行调试单个 task，所以 CubeStudio 添加了单 task 独立调试的功能。

每个 task 都可以独立 debug 和独立运行，这样就不用每次都运行整体 pipeline 才能进行调试。并且对于分布式任务的 task，CubeStudio 在模板中借用 stern 实现了多个容器 log 聚合查看，这样就不用每个 worker 的日志都单独去看了。

单个任务为分布式任务

往往某个任务需要是分布式任务，所以需要启动分布式任务。云原生分布式任务需要 k8s 对应的 operator。分布式算法模板的实现方法，主要在模板中实现 launcher：创建对应分布式任务的 job yaml，提交到 k8s，对应的 job operator 创建每个 worker 的 pod 和 service，然后运行业务代码。launcher 端启动 stern 组件，监听该分布式任务的所有 pod 的日志并进行输出，这样就可以在当前任务中查看到所有 worker 的日志。

ℹ️ 核实：stern 日志聚合在各分布式框架的 launcher 中使用，例如 job-template/job/pytorch/launcher.py、job-template/job/volcano/launcher.py、job-template/job/deep-learning-framework/launcher-*.py。分布式框架的具体实现见 平台架构-分布式训练与加速.md。

并且每个任务启动后都可以看到相应的资源情况，这样可以边调试边配置资源的申请量。在 gpu 上也可以及时知道是否还有优化的空间、目前任务的主要瓶颈在哪里，这样能更加直观地确定优化的方向。

断点重训 / 重试 / 超时

对于某些情况，一个任务流运行后，可能在某些任务开始出现偶发性错误。CubeStudio 提供了在结束 pipeline 后从失败点继续训练的功能，这样就不用重复训练某些耗时的任务。同时 CubeStudio 也为每个任务提供了重试次数和超时时长的配置，并且提供了任务的实时和离线日志查询两种功能。

pod 运行

pod 调度情况可以最真实地反映每个任务或分布式任务的状态，还有资源使用情况。

定时调度

任务编排调试好以后，还要进行定时调度。CubeStudio 支持了定时调度任务的补录、忽略、重试、依赖、并发限制、过期淘汰等功能，能应对任务堆积、平台故障等可能问题。

并且对同一个 pipeline，区分对待手动运行和定时调度两种实例。

注意（定时调度细节偏使用/运维，可与 03-平台使用/ 互链）：

• 定时调度由 celery 定时任务 make_timerun_config 进行控制。

  ℹ️ 核实：任务名 task.make_timerun_config 注册于 myapp/config.py:612 与 myapp/config.py:696-697（beat 调度），实现见 myapp/tasks/schedules.py。

• pipeline 扩展参数中 cronjob_start_time 用来控制历史任务的补录时间起点（修改 pipeline 调度类型和 crontab 会重置 cronjob_start_time 为当前时间）。
• pipeline 参数中需要使用 execution_date 才能在补录时保留实例时间点状态。
• pipeline 的 depends_on_past 参数控制新的实例运行是否要关注上一次实例是否完成。
• pipeline 的 expired_limit 参数控制过往未调度实例只保留多少个最新实例运行，更久远实例过期。

  ℹ️ 核实：表字段注释为"过期保留个数，此数值有效时，会优先使用，覆盖 max_active_runs 的功能"，见 myapp/migrations/versions-mysql/3471634cbb23_.py:619。

• pipeline 的 max_active_runs 参数控制最多同时并行多少个调度实例运行。
• 最大并发周期为全局变量 PIPELINE_TASK_CRON_RESOLUTION，默认 10（分钟一次）。

  ℹ️ 核实：PIPELINE_TASK_CRON_RESOLUTION = 10，见 myapp/config.py:387，注释为"任务的最小执行间隔 min"。

定时调度记录

查看路径：训练 - 定时调度记录

字段说明：执行时间为这个 pipeline 本次调度该发起的时间点。

状态：comed 为调度配置已经产生，created 为调度已经发起。

操作说明：

1. 平台会根据 pipeline 的配置决定是否发起调度。
2. 状态链接中可以看到本次调度发起的 workflow 的运行情况。
3. 日志链接中可以看到本次调度发起的日志。

开放式插件

平台自带的任务模板并不一定能完全满足所有需求。除了用户可以使用自定义镜像、自定义命令的方式外，平台还将插件的注册功能开放给用户。

这样用户就可以自己定义模板，然后开放给全平台的用户使用。用户只需要自己构建好功能镜像，然后将其他用户使用此镜像需要填写的参数注册进来就可以了。其中也包含模板特定功能，例如固定的挂载、固定的环境变量、k8s 账号等。

任务模板的具体使用与编写规范见 05-任务模板/。