2026/9/15 19:21:25
FEATURED · 精选文章 
SkyPilot Job Groups并行异构任务编排与 RL 后训练实战指南【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot导读Job Groups 是 SkyPilot 中把多个相互关联的任务作为一个整体并行调度与管理的核心机制。与顺序执行的 managed job pipeline 不同Job Group 一次性并发启动所有任务允许每个任务声明异构资源GPU 训练、CPU 数据服务并通过 Kubernetes 集群内的 hostname 自动完成服务发现。读完本文你将掌握如何用多文档 YAML 或 Python SDK 定义 Job Group、配置 primary/auxiliary 任务生命周期、控制组内网络以及在 RL 后训练、并行训练-评估等真实场景中落地这一机制。Job Groups 是什么从顺序管道到并行单元Job Groups 允许你将多个相关任务并行运行并作为一个统一的管理单元来操作。它与 SkyPilot 其他任务编排方式形成互补编排方式执行模型典型场景Managed Jobs单任务单任务独立运行独立训练/推理任务Managed Job Pipelines任务顺序执行管道有先后依赖的多阶段流程Job Groups任务同时启动、并行运行RL 后训练、并行训练-评估、多组件服务架构在管理维度上Job Group 是一个整体sky jobs queue中每个任务显示在组 ID 之下sky jobs cancel可以一次取消整组日志按任务名或任务 ID 分开查看。核心特性包括并行执行多个任务同时启动各自独立运行异构资源每个任务可声明不同资源如训练用 GPU、数据服务用 CPU自动服务发现任务通过组内 hostname 相互发现并通信Kubernetes 上独立恢复每个任务独立处理抢占恢复互不影响组内再启动任务组内任务可以继续sky jobs launch新的 managed job这些任务挂在该组名下并随组一起取消。选择建议来自原文档 job-groups.rst当你的负载包含异构任务、需要并行运行且相互通信时使用 Job Groups。单任务内的同构多节点训练请使用 distributed jobs顺序执行请使用 managed job pipelines。在源码层面Job Group 是 DAG 的一种特殊执行模式。sky/jobs/server/core.py中通过dag.is_job_group()判断并将execution写为parallel见 server/core.py数据库迁移sky/schemas/db/spot_jobs/013_add_job_group_columns.py明确记录了execution (TEXT)paralleljob group或 serial。一个典型的 SkyPilot Job Group五个异构任务并行运行。样本按 rollout → reward → buffer → trainer 流动新策略权重每步回传 rollout server形成完整的 RL 反馈闭环。创建 Job Group多文档 YAML 与启动命令Job Group 通过多文档 YAMLmulti-document YAML定义第一个文档是描述组属性的header后续每个文档是一个独立任务定义# job-group.yaml --- # Header: Job Group 配置 name: my-job-group execution: parallel # 必填表明这是一个 Job Group --- # Task 1: Trainer name: trainer resources: accelerators: A100:1 run: | python train.py --- # Task 2: Evaluator name: evaluator resources: accelerators: A100:1 run: | python evaluate.py启动命令$ sky jobs launch job-group.yamlheader 之后的每个任务文档遵循标准 SkyPilot 任务 YAML 格式。组内每个任务必须有唯一的名字——名字用于服务发现和日志查看。Header 字段详解header 文档支持以下字段完整继承自原文档 job-groups.rst字段默认值说明name必填Job Group 的名称execution必填必须为parallel才表示是 Job Groupprimary_tasks无所有任务都是 primaryprimary 任务名列表。不在列表中的任务为 auxiliary如 data server、replay buffer 等常驻服务等待终止信号。当所有 primary 任务完成时auxiliary 任务被终止termination_delay无primary 任务完成后、终止 auxiliary 任务前的延迟让它们有机会完成收尾工作如 flush 数据。可以是字符串如30s、5m或按任务区分的字典如{default: 30s, replay-buffer: 1m}inter_connection不设置默认按true处理任务是否需要通过 hostname 相互访问。true将所有任务放在单个 Kubernetes 集群并配置 hostname 连通性任一条件不满足则硬失败false刻意跳过所有网络配置任务仍倾向同址部署但可能落到不同集群不设置假定为true——所有任务同址到单个 Kubernetes 集群并配置网络除非任务请求了非 Kubernetes 基础设施或固定了无法同址的基础设施此时降级为false并给出警告。详见下文组内网络以termination_delay为例延迟给了辅助服务例如内存中的 replay buffer、待 flush 的数据管道一段善后窗口——它们在这个窗口内完成数据落盘或清理之后才收到终止信号。服务发现hostname 与环境变量SkyPilot 自动配置网络使 Job Group 内任务可以通过 hostname 相互访问。Hostname 格式每个任务的 head 节点可通过以下 hostname 访问{task_name}-0.{job_group_name}多节点任务中worker 节点使用{task_name}-{node_index}.{job_group_name}例如名为rlhf-experiment的 Job Group 中有一个 2 节点的trainer任务trainer-0.rlhf-experiment—— head 节点rank 0trainer-1.rlhf-experiment—— worker 节点rank 1这个 hostname 格式在源码中有精确定义sky/jobs/job_group_networking.py中的get_job_group_hostname()返回{job_name}-{node_idx}.{job_group_name}格式并注明 K8s 与 SSH 云环境使用同一格式见 job_group_networking.py。环境变量SkyPilot 向所有任务注入以下环境变量变量说明SKYPILOT_JOBGROUP_NAMEJob Group 的名称任务内使用示例在 evaluator 中访问 trainer# 通过 hostname 访问 trainer 任务 curl http://trainer-0.${SKYPILOT_JOBGROUP_NAME}:8000/status该环境变量在 sky/jobs/constants.py 中定义为SKYPILOT_JOBGROUP_NAME_ENV_VAR是所有任务中可用的标准注入变量。组内网络inter_connection 的三种模式组内服务发现目前仅在Kubernetes上受支持。默认情况下组内网络是启用的SkyPilot 将 Job Group 的所有任务放在单个 Kubernetes 集群上任务在运行前会等待对端 hostname 变为可解析。如果网络无法初始化任务会以明确的错误失败而不是在无连通性的情况下运行。三种模式的行为对比配置行为不设置默认视为true所有任务同址到单个 K8s 集群并配置网络若任务要求非 K8s 基础设施或固定了无法同址的设施则降级为false并发出警告跳过网络配置inter_connection: true更严格无法建立组内网络的放置方案非 K8s 基础设施、无公共选项的固定设施直接报错拒绝inter_connection: false刻意跳过所有组内网络配置任务立即启动、不等待对端显式设置inter_connection: false适用于任务之间不需要通过 hostname 互访的场景例如通过外部端点或共享对象存储协调的组件name: my-job-group execution: parallel inter_connection: false --- # ... task documents ...使用inter_connection: false时不建立组内网络任务立即启动无需等待对端当单个集群无法容纳整组任务时例如所需 GPU 类型分布在不同集群任务可能被放置到不同的 Kubernetes 集群或放置到非 Kubernetes 基础设施也可以通过每个任务的infra: k8s/context显式固定到不同集群。注意跨集群的 hostname 服务发现目前尚不支持——被放置在不同集群的任务无法通过组内 hostname 相互访问。从源码看sky/jobs/job_group_networking.py实现了完整的组内 DNS 机制K8s 侧通过generate_k8s_dns_mappings()生成k8s_dns - simple_hostname映射并由后台skypilot-jobgroup-dns-updater-{job_group_name}进程持续更新见 job_group_networking.pySSH 云侧则向/etc/hosts注入internal_ip hostname条目见 job_group_networking.py。网络就绪状态通过 marker file 标记任务启动时等待该标记出现。查看日志Job Group 支持按任务查看日志# 按任务名查看某个任务的日志 $ sky jobs logs job_id trainer # 按任务 ID 查看某个任务的日志 $ sky jobs logs job_id 0 # 查看所有任务的日志默认行为 $ sky jobs logs job_id查看多任务作业日志时SkyPilot 会显示提示Hint: This job has 3 tasks. Use sky jobs logs 42 TASK to view logs for a specific task (TASK can be task ID or name).实战示例一并行训练-评估共享存储训练与评估并行运行通过 Kubernetes PVC 卷共享 checkpoint--- name: train-eval execution: parallel --- name: trainer resources: accelerators: A100:1 volumes: /checkpoints: my-checkpoint-volume run: | python train.py --checkpoint-dir /checkpoints --- name: evaluator resources: accelerators: A100:1 volumes: /checkpoints: my-checkpoint-volume run: | python evaluate.py --checkpoint-dir /checkpointstrainer 把 checkpoint 写入共享卷evaluator 实时监控并评估新 checkpoint——训练与评估互不阻塞。仓库中对应的完整示例位于 examples/job-group-sdk/job_group.yamlserver-client 通信示例以及 llm/train-eval-jobgroup 目录下的并行训练-评估示例。实战示例二RL 后训练架构下面演示一个包含 5 个任务的分布式 RL 后训练架构对应文首的架构图。trainer 与 rollout-server 共享一个ReadWriteManyKubernetes 卷trainer 每隔 N 步把新更新的策略权重推回 rollout-server从而闭合 RL 反馈循环--- name: rlhf-training execution: parallel --- name: />Primary 与 Auxiliary 任务生命周期管理在分布式负载中通常有一个主任务如 trainer和若干持续运行、直到主任务发出完成信号才退出的支撑服务如 data server、replay buffer。这些支撑服务就是auxiliary 任务——它们没有天然的终止点需要被告知何时关闭。用primary_tasks指定哪些任务驱动作业的生命周期。未被列出的 auxiliary 任务会在所有 primary 任务完成时自动终止--- name: train-with-services execution: parallel primary_tasks: [trainer] # 只有 trainer 是 primary termination_delay: 30s # trainer 完成后给服务 30s 收尾 --- name: trainer resources: accelerators: A100:1 run: | python train.py # Primary 任务它结束时作业完成 --- name:>--- name: train-and-eval execution: parallel primary_tasks: [trainer, eval-watcher] --- name: trainer resources: accelerators: H100:8 run: | python train.py --checkpoint-dir /checkpoints --- name: eval-watcher resources: cpus: 2 run: | # SkyPilot client 已安装在每个节点上Kubernetes 下位于 $HOME # Slurm 下位于 $SKY_RUNTIME_DIR任务的凭据将其指向 API server。 source ${SKY_RUNTIME_DIR:-$HOME}/skypilot-runtime/bin/activate # 每个新 checkpoint 一个评估作业。每个都是本组的动态任务 # 显示在 sky jobs queue 组名下随组一起取消。 for ckpt in $(python watch_checkpoints.py /checkpoints); do sky jobs launch -y -d -n eval-$ckpt eval.yaml --env CKPT$ckpt done # 退出前等待评估完成组在两个 primary 都完成后才结束 # 届时仍在运行的评估会被取消。 python wait_for_evals.pywatcher 是 primary 任务因此组会存活到它所启动的最后一个评估结束。如果它是 auxiliary 任务则会在 trainer 完成后经过组的termination_delay被终止其评估也随之取消。队列中动态任务显示在组下、编号为任务 2 和 3$ sky jobs queue ID TASK NAME ... STATUS 42 - train-and-eval ... RUNNING ↳ 0 trainer [P] ... RUNNING ↳ 1 eval-watcher [P] ... RUNNING ↳ 2 eval-step-1000 ... SUCCEEDED ↳ 3 eval-step-2000 ... RUNNING $ sky jobs cancel 42 --task 3 # 单独取消第二个评估组的归属选择。从 Job Group 内部启动的作业默认挂到该组。--job-group job id or name可从任意位置显式挂到运行中的 Job Group--no-job-group即使在组内也以顶层作业启动。SDK 中sky.jobs.launch接受job_group参数sky.jobs.AUTO_JOB_GROUP默认、job id 或唯一运行中的 job 名称或None顶层作业。注意挂接 Job Group 需要远程 SkyPilot API server 以 consolidation mode 运行 managed jobs并且启动任务需要具有 API server 访问权限。在其他环境下--job-group会被拒绝组内启动的作业作为顶层作业运行。动态成员不参与组的服务发现无法通过组 hostname 访问、也不获得 hostname独立于组调度且没有自己的termination_delay。仓库中有一个可直接运行的完整示例 examples/job-group-sdk/job_group_eval_watcher.yamltrainer 每 30 秒产出一个checkpointeval-watcher 为每个 checkpoint 启动一个评估作业sky jobs launch -y -d -n eval-step-$step ...并通过解析sky jobs queue的输出等待所有评估完成后再退出。文件中还处理了队列调用失败重试、仅统计本组名下eval-step-*作业等细节。使用 Python SDK 创建 Job Group除了 YAMLJob Groups 完全可以用 SkyPilot SDK 在 Python 中创建和启动。以下是最小示例来自原文档与仓库中 examples/job-group-sdk/job_group_sdk.py 一致import sky server sky.Task(nameserver, runpython3 -m http.server 8080) server.set_resources(sky.Resources(cpus2, infrakubernetes)) client sky.Task(nameclient, runcurl http://server-0.${SKYPILOT_JOBGROUP_NAME}:8080/) client.set_resources(sky.Resources(cpus2, infrakubernetes)) with sky.Dag() as dag: dag.add(server) dag.add(client) dag.name my-group dag.set_execution(sky.DagExecution.PARALLEL) sky.stream_and_get(sky.jobs.launch(dag))要点每个sky.Task对应 Job Group 的一个任务sky.Dag把任务组合起来dag.name即 Job Group 名称dag.set_execution(sky.DagExecution.PARALLEL)等价于 YAML header 中的execution: parallelsky.jobs.launch(dag)启动整个组sky.stream_and_get实时输出日志。仓库中的job_group_sdk.py演示了一个更完整的版本server 启动 HTTP 服务并运行 60 秒client 用server-0.${SKYPILOT_JOBGROUP_NAME}:8080做最多 10 次带重试的连接测试。此外 examples/job-group-sdk/job_group_primary_aux_sdk.py 展示了 SDK 方式配置 primary/auxiliary 任务。当前限制同址部署Co-locationJob Group 的所有任务运行在同一个基础设施上同一个 Kubernetes 集群或云可用区。网络服务发现任务间的 hostname 通信目前仅支持 Kubernetes。在其他云上任务可以并行运行但无法使用 hostname 格式相互通信。执行模式Job Groups 要求 header 中有execution: parallel。需要顺序执行时请使用 managed job pipelines省略execution字段或设为serial见 managed-jobs。小结Job Groups 是 SkyPilot 处理异构、并行、需互通信的多组件负载的核心原语。回顾要点多文档 YAML或Python SDKDag PARALLEL两种方式定义组execution: parallel是组的关键标志primary_taskstermination_delay控制组生命周期Kubernetes 上默认启用 hostname 服务发现{task}-{idx}.{group}SKYPILOT_JOBGROUP_NAMEinter_connection显式控制网络行为动态任务机制允许组内launch job支撑 eval watcher 这类按 checkpoint 起作业的架构牢记限制同址部署、hostname 服务发现仅限 Kubernetes。进一步阅读单任务/顺序管道见 managed-jobs单任务内多节点分布式训练见 dist-jobs更多 YAML 与 SDK 示例分别位于 examples/job-group-sdk 与 llm/rl-post-training-jobgroup。【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
SkyPilot Job Groups：并行异构任务编排与 RL 后训练实战指南
SkyPilot Job Groups并行异构任务编排与 RL 后训练实战指南【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot导读Job Groups 是 SkyPilot 中把多个相互关联的任务作为一个整体并行调度与管理的核心机制。与顺序执行的 managed job pipeline 不同Job Group 一次性并发启动所有任务允许每个任务声明异构资源GPU 训练、CPU 数据服务并通过 Kubernetes 集群内的 hostname 自动完成服务发现。读完本文你将掌握如何用多文档 YAML 或 Python SDK 定义 Job Group、配置 primary/auxiliary 任务生命周期、控制组内网络以及在 RL 后训练、并行训练-评估等真实场景中落地这一机制。Job Groups 是什么从顺序管道到并行单元Job Groups 允许你将多个相关任务并行运行并作为一个统一的管理单元来操作。它与 SkyPilot 其他任务编排方式形成互补编排方式执行模型典型场景Managed Jobs单任务单任务独立运行独立训练/推理任务Managed Job Pipelines任务顺序执行管道有先后依赖的多阶段流程Job Groups任务同时启动、并行运行RL 后训练、并行训练-评估、多组件服务架构在管理维度上Job Group 是一个整体sky jobs queue中每个任务显示在组 ID 之下sky jobs cancel可以一次取消整组日志按任务名或任务 ID 分开查看。核心特性包括并行执行多个任务同时启动各自独立运行异构资源每个任务可声明不同资源如训练用 GPU、数据服务用 CPU自动服务发现任务通过组内 hostname 相互发现并通信Kubernetes 上独立恢复每个任务独立处理抢占恢复互不影响组内再启动任务组内任务可以继续sky jobs launch新的 managed job这些任务挂在该组名下并随组一起取消。选择建议来自原文档 job-groups.rst当你的负载包含异构任务、需要并行运行且相互通信时使用 Job Groups。单任务内的同构多节点训练请使用 distributed jobs顺序执行请使用 managed job pipelines。在源码层面Job Group 是 DAG 的一种特殊执行模式。sky/jobs/server/core.py中通过dag.is_job_group()判断并将execution写为parallel见 server/core.py数据库迁移sky/schemas/db/spot_jobs/013_add_job_group_columns.py明确记录了execution (TEXT)paralleljob group或 serial。一个典型的 SkyPilot Job Group五个异构任务并行运行。样本按 rollout → reward → buffer → trainer 流动新策略权重每步回传 rollout server形成完整的 RL 反馈闭环。创建 Job Group多文档 YAML 与启动命令Job Group 通过多文档 YAMLmulti-document YAML定义第一个文档是描述组属性的header后续每个文档是一个独立任务定义# job-group.yaml --- # Header: Job Group 配置 name: my-job-group execution: parallel # 必填表明这是一个 Job Group --- # Task 1: Trainer name: trainer resources: accelerators: A100:1 run: | python train.py --- # Task 2: Evaluator name: evaluator resources: accelerators: A100:1 run: | python evaluate.py启动命令$ sky jobs launch job-group.yamlheader 之后的每个任务文档遵循标准 SkyPilot 任务 YAML 格式。组内每个任务必须有唯一的名字——名字用于服务发现和日志查看。Header 字段详解header 文档支持以下字段完整继承自原文档 job-groups.rst字段默认值说明name必填Job Group 的名称execution必填必须为parallel才表示是 Job Groupprimary_tasks无所有任务都是 primaryprimary 任务名列表。不在列表中的任务为 auxiliary如 data server、replay buffer 等常驻服务等待终止信号。当所有 primary 任务完成时auxiliary 任务被终止termination_delay无primary 任务完成后、终止 auxiliary 任务前的延迟让它们有机会完成收尾工作如 flush 数据。可以是字符串如30s、5m或按任务区分的字典如{default: 30s, replay-buffer: 1m}inter_connection不设置默认按true处理任务是否需要通过 hostname 相互访问。true将所有任务放在单个 Kubernetes 集群并配置 hostname 连通性任一条件不满足则硬失败false刻意跳过所有网络配置任务仍倾向同址部署但可能落到不同集群不设置假定为true——所有任务同址到单个 Kubernetes 集群并配置网络除非任务请求了非 Kubernetes 基础设施或固定了无法同址的基础设施此时降级为false并给出警告。详见下文组内网络以termination_delay为例延迟给了辅助服务例如内存中的 replay buffer、待 flush 的数据管道一段善后窗口——它们在这个窗口内完成数据落盘或清理之后才收到终止信号。服务发现hostname 与环境变量SkyPilot 自动配置网络使 Job Group 内任务可以通过 hostname 相互访问。Hostname 格式每个任务的 head 节点可通过以下 hostname 访问{task_name}-0.{job_group_name}多节点任务中worker 节点使用{task_name}-{node_index}.{job_group_name}例如名为rlhf-experiment的 Job Group 中有一个 2 节点的trainer任务trainer-0.rlhf-experiment—— head 节点rank 0trainer-1.rlhf-experiment—— worker 节点rank 1这个 hostname 格式在源码中有精确定义sky/jobs/job_group_networking.py中的get_job_group_hostname()返回{job_name}-{node_idx}.{job_group_name}格式并注明 K8s 与 SSH 云环境使用同一格式见 job_group_networking.py。环境变量SkyPilot 向所有任务注入以下环境变量变量说明SKYPILOT_JOBGROUP_NAMEJob Group 的名称任务内使用示例在 evaluator 中访问 trainer# 通过 hostname 访问 trainer 任务 curl http://trainer-0.${SKYPILOT_JOBGROUP_NAME}:8000/status该环境变量在 sky/jobs/constants.py 中定义为SKYPILOT_JOBGROUP_NAME_ENV_VAR是所有任务中可用的标准注入变量。组内网络inter_connection 的三种模式组内服务发现目前仅在Kubernetes上受支持。默认情况下组内网络是启用的SkyPilot 将 Job Group 的所有任务放在单个 Kubernetes 集群上任务在运行前会等待对端 hostname 变为可解析。如果网络无法初始化任务会以明确的错误失败而不是在无连通性的情况下运行。三种模式的行为对比配置行为不设置默认视为true所有任务同址到单个 K8s 集群并配置网络若任务要求非 K8s 基础设施或固定了无法同址的设施则降级为false并发出警告跳过网络配置inter_connection: true更严格无法建立组内网络的放置方案非 K8s 基础设施、无公共选项的固定设施直接报错拒绝inter_connection: false刻意跳过所有组内网络配置任务立即启动、不等待对端显式设置inter_connection: false适用于任务之间不需要通过 hostname 互访的场景例如通过外部端点或共享对象存储协调的组件name: my-job-group execution: parallel inter_connection: false --- # ... task documents ...使用inter_connection: false时不建立组内网络任务立即启动无需等待对端当单个集群无法容纳整组任务时例如所需 GPU 类型分布在不同集群任务可能被放置到不同的 Kubernetes 集群或放置到非 Kubernetes 基础设施也可以通过每个任务的infra: k8s/context显式固定到不同集群。注意跨集群的 hostname 服务发现目前尚不支持——被放置在不同集群的任务无法通过组内 hostname 相互访问。从源码看sky/jobs/job_group_networking.py实现了完整的组内 DNS 机制K8s 侧通过generate_k8s_dns_mappings()生成k8s_dns - simple_hostname映射并由后台skypilot-jobgroup-dns-updater-{job_group_name}进程持续更新见 job_group_networking.pySSH 云侧则向/etc/hosts注入internal_ip hostname条目见 job_group_networking.py。网络就绪状态通过 marker file 标记任务启动时等待该标记出现。查看日志Job Group 支持按任务查看日志# 按任务名查看某个任务的日志 $ sky jobs logs job_id trainer # 按任务 ID 查看某个任务的日志 $ sky jobs logs job_id 0 # 查看所有任务的日志默认行为 $ sky jobs logs job_id查看多任务作业日志时SkyPilot 会显示提示Hint: This job has 3 tasks. Use sky jobs logs 42 TASK to view logs for a specific task (TASK can be task ID or name).实战示例一并行训练-评估共享存储训练与评估并行运行通过 Kubernetes PVC 卷共享 checkpoint--- name: train-eval execution: parallel --- name: trainer resources: accelerators: A100:1 volumes: /checkpoints: my-checkpoint-volume run: | python train.py --checkpoint-dir /checkpoints --- name: evaluator resources: accelerators: A100:1 volumes: /checkpoints: my-checkpoint-volume run: | python evaluate.py --checkpoint-dir /checkpointstrainer 把 checkpoint 写入共享卷evaluator 实时监控并评估新 checkpoint——训练与评估互不阻塞。仓库中对应的完整示例位于 examples/job-group-sdk/job_group.yamlserver-client 通信示例以及 llm/train-eval-jobgroup 目录下的并行训练-评估示例。实战示例二RL 后训练架构下面演示一个包含 5 个任务的分布式 RL 后训练架构对应文首的架构图。trainer 与 rollout-server 共享一个ReadWriteManyKubernetes 卷trainer 每隔 N 步把新更新的策略权重推回 rollout-server从而闭合 RL 反馈循环--- name: rlhf-training execution: parallel --- name: />Primary 与 Auxiliary 任务生命周期管理在分布式负载中通常有一个主任务如 trainer和若干持续运行、直到主任务发出完成信号才退出的支撑服务如 data server、replay buffer。这些支撑服务就是auxiliary 任务——它们没有天然的终止点需要被告知何时关闭。用primary_tasks指定哪些任务驱动作业的生命周期。未被列出的 auxiliary 任务会在所有 primary 任务完成时自动终止--- name: train-with-services execution: parallel primary_tasks: [trainer] # 只有 trainer 是 primary termination_delay: 30s # trainer 完成后给服务 30s 收尾 --- name: trainer resources: accelerators: A100:1 run: | python train.py # Primary 任务它结束时作业完成 --- name:>--- name: train-and-eval execution: parallel primary_tasks: [trainer, eval-watcher] --- name: trainer resources: accelerators: H100:8 run: | python train.py --checkpoint-dir /checkpoints --- name: eval-watcher resources: cpus: 2 run: | # SkyPilot client 已安装在每个节点上Kubernetes 下位于 $HOME # Slurm 下位于 $SKY_RUNTIME_DIR任务的凭据将其指向 API server。 source ${SKY_RUNTIME_DIR:-$HOME}/skypilot-runtime/bin/activate # 每个新 checkpoint 一个评估作业。每个都是本组的动态任务 # 显示在 sky jobs queue 组名下随组一起取消。 for ckpt in $(python watch_checkpoints.py /checkpoints); do sky jobs launch -y -d -n eval-$ckpt eval.yaml --env CKPT$ckpt done # 退出前等待评估完成组在两个 primary 都完成后才结束 # 届时仍在运行的评估会被取消。 python wait_for_evals.pywatcher 是 primary 任务因此组会存活到它所启动的最后一个评估结束。如果它是 auxiliary 任务则会在 trainer 完成后经过组的termination_delay被终止其评估也随之取消。队列中动态任务显示在组下、编号为任务 2 和 3$ sky jobs queue ID TASK NAME ... STATUS 42 - train-and-eval ... RUNNING ↳ 0 trainer [P] ... RUNNING ↳ 1 eval-watcher [P] ... RUNNING ↳ 2 eval-step-1000 ... SUCCEEDED ↳ 3 eval-step-2000 ... RUNNING $ sky jobs cancel 42 --task 3 # 单独取消第二个评估组的归属选择。从 Job Group 内部启动的作业默认挂到该组。--job-group job id or name可从任意位置显式挂到运行中的 Job Group--no-job-group即使在组内也以顶层作业启动。SDK 中sky.jobs.launch接受job_group参数sky.jobs.AUTO_JOB_GROUP默认、job id 或唯一运行中的 job 名称或None顶层作业。注意挂接 Job Group 需要远程 SkyPilot API server 以 consolidation mode 运行 managed jobs并且启动任务需要具有 API server 访问权限。在其他环境下--job-group会被拒绝组内启动的作业作为顶层作业运行。动态成员不参与组的服务发现无法通过组 hostname 访问、也不获得 hostname独立于组调度且没有自己的termination_delay。仓库中有一个可直接运行的完整示例 examples/job-group-sdk/job_group_eval_watcher.yamltrainer 每 30 秒产出一个checkpointeval-watcher 为每个 checkpoint 启动一个评估作业sky jobs launch -y -d -n eval-step-$step ...并通过解析sky jobs queue的输出等待所有评估完成后再退出。文件中还处理了队列调用失败重试、仅统计本组名下eval-step-*作业等细节。使用 Python SDK 创建 Job Group除了 YAMLJob Groups 完全可以用 SkyPilot SDK 在 Python 中创建和启动。以下是最小示例来自原文档与仓库中 examples/job-group-sdk/job_group_sdk.py 一致import sky server sky.Task(nameserver, runpython3 -m http.server 8080) server.set_resources(sky.Resources(cpus2, infrakubernetes)) client sky.Task(nameclient, runcurl http://server-0.${SKYPILOT_JOBGROUP_NAME}:8080/) client.set_resources(sky.Resources(cpus2, infrakubernetes)) with sky.Dag() as dag: dag.add(server) dag.add(client) dag.name my-group dag.set_execution(sky.DagExecution.PARALLEL) sky.stream_and_get(sky.jobs.launch(dag))要点每个sky.Task对应 Job Group 的一个任务sky.Dag把任务组合起来dag.name即 Job Group 名称dag.set_execution(sky.DagExecution.PARALLEL)等价于 YAML header 中的execution: parallelsky.jobs.launch(dag)启动整个组sky.stream_and_get实时输出日志。仓库中的job_group_sdk.py演示了一个更完整的版本server 启动 HTTP 服务并运行 60 秒client 用server-0.${SKYPILOT_JOBGROUP_NAME}:8080做最多 10 次带重试的连接测试。此外 examples/job-group-sdk/job_group_primary_aux_sdk.py 展示了 SDK 方式配置 primary/auxiliary 任务。当前限制同址部署Co-locationJob Group 的所有任务运行在同一个基础设施上同一个 Kubernetes 集群或云可用区。网络服务发现任务间的 hostname 通信目前仅支持 Kubernetes。在其他云上任务可以并行运行但无法使用 hostname 格式相互通信。执行模式Job Groups 要求 header 中有execution: parallel。需要顺序执行时请使用 managed job pipelines省略execution字段或设为serial见 managed-jobs。小结Job Groups 是 SkyPilot 处理异构、并行、需互通信的多组件负载的核心原语。回顾要点多文档 YAML或Python SDKDag PARALLEL两种方式定义组execution: parallel是组的关键标志primary_taskstermination_delay控制组生命周期Kubernetes 上默认启用 hostname 服务发现{task}-{idx}.{group}SKYPILOT_JOBGROUP_NAMEinter_connection显式控制网络行为动态任务机制允许组内launch job支撑 eval watcher 这类按 checkpoint 起作业的架构牢记限制同址部署、hostname 服务发现仅限 Kubernetes。进一步阅读单任务/顺序管道见 managed-jobs单任务内多节点分布式训练见 dist-jobs更多 YAML 与 SDK 示例分别位于 examples/job-group-sdk 与 llm/rl-post-training-jobgroup。【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
