提交Ray任务到工作流
更新时间：2026-10-16
开发机支持通过CLI相关指令快速将本地的任务提交到工作流，使用该CLI支持自动打包上传然后构建工作流最后运行工作流，并且同时支持工作流相关的操作。
适用场景
在开发机中完成本地代码调试后，希望一条命令把代码提交为工作流任务并运行，或对已有工作流进行运行、停止、重跑、调度等管理。
前提条件
使用工作流提交前，请确保满足以下条件，否则提交会被平台拒绝：
- 持有一对 BCE AK/SK（长期或 STS 临时凭证均可）；
- 拥有一个有权限的
project（DataBuilder 控制台分配的工作台项目），并对该项目有读写权限； - 所选
project已绑定资源组，且资源组中具备所需的实例或队列资源； - 配置文件中使用的用户须是所选工作空间的成员，并具备对应项目的读写权限。
快速开始
通过一个快速示例，完成从开发机提交 Ray 任务到工作流的完整流程。
步骤一：准备配置文件
在工作文件夹下新建配置文件 workflow_conf.yaml，填写用户凭证与工作流信息：
YAML
1workspace_id: # 工作空间 id
2user_id: # 用户 id
3
4credentials:
5 access_key: # 用户 AK
6 secret_key: # 用户 SK
7
8workflow:
9 name: quickstart # 工作流名称
10 project: /Projects/my_test # 工作台下的项目地址
11 workflow-type: OFFLINE # 工作模式，默认离线
12 global: { bizdate: "20260924", region: bj } # 全局变量参数
13
14 nodes: # 节点参数，示例使用单节点
15 - name: node_test # 节点名称
16 working-dir: . # 可用绝对路径；相对路径相对于 workflow_conf.yaml 所在文件夹
17 entry: python test_code.py # 代码入口步骤二：准备代码
在工作文件夹下新建示例代码 test_code.py：
Python
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3"""
4Workflow 测试代码：验证工作流是否正常运行，仅做简单输出。
5"""
6import time
7import platform
8
9
10def main():
11 start = time.time()
12 print("工作流任务测试开始", flush=True)
13 print(f" Python 版本 : {platform.python_version()}", flush=True)
14 print(f" 开始时间 : {time.strftime('%Y-%m-%d %H:%M:%S')}", flush=True)
15
16 # 模拟一点点工作
17 total = sum(range(1000))
18 print(f" 计算结果 : {total}", flush=True)
19
20 elapsed = time.time() - start
21 print(f" 任务耗时 : {elapsed:.1f}s", flush=True)
22 print("工作流任务测试完成", flush=True)
23
24
25if __name__ == "__main__":
26 main()步骤三：提交任务
在开发机终端执行提交命令，CLI 会自动完成打包上传、创建并运行工作流：
Bash
1databuilder workflow submit --config-file ./workspace/workflow_conf.yaml步骤四：查看工作流结果
进入平台的工作流页面，即可看到工作流已创建成功。
查看工作流配置：可看到设定的相关参数。
查看运行日志：单击工作流运行记录，选择任务节点并单击运行日志，即可查看任务输出和结果。
配置文件详解
配置文件支持更多工作流级别与节点级别的详细配置。
配置定时调度
在 workflow 级别下配置定时调度，配置后默认开启该调度：
YAML
1workflow:
2 schedule:
3 crontab: "0 0 2 * * ? *" # cron 表达式
4 start-time: "2026-01-01 00:00:00" # yyyy-MM-dd HH:mm:ss
5 end-time: "2027-01-01 00:00:00"
6 type: crontab配置多节点
在 nodes 级别下配置多节点，deps 为 DAG 边配置：B deps A 表示 B 依赖 A（A 在前）。
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 - name: test1
6 working-dir: ./code
7 entry: python test_code1.py
8 - name: test2
9 working-dir: ./code1
10 entry: python test_code2.py
11 deps: [test1]示例为两条流程：test 为独立流程，test1 -> test2 为另一条流程。
配置工作流资源
在 workflow 级别下配置资源类型与规格：
YAML
1workflow:
2 resource:
3 replica-num: 2
4 vcpu: 1
5 vgpu: 0
6 gpu-type: CPU
7 image:
8 engine: Ray
9 mirror-version: v1.0配置节点资源
节点支持单独配置资源：
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 resource:
6 replica-num: 4
7 vgpu: 1
8 gpu-type: A100
9 queue-type: SERVERLESS # 队列类型 DEDICATED / SERVERLESS
10 type: RAY_TASK
11 use-daemon: true # 是否使用常驻集群
12 resource-type: GPU配置节点变量
节点支持单独配置变量（变量名称必须大写）：
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 env: { LOG_LEVEL: INFO, TIME: now }自定义节点镜像
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 custom-image:
6 image: # 镜像地址
7 image-username: xxx # 私有化仓库必须提供镜像仓库账号
8 image-password: xxx # 私有化仓库必须提供镜像仓库密码配置重试策略
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 fail-retry-times: 2 # 失败重试次数（默认 0）
6 fail-retry-interval: 5 # 重试间隔，单位分钟（默认 1）配置告警策略
YAML
1nodes:
2 - name: test
3 working-dir: .
4 entry: python test_code.py
5 alerts:
6 - incident-action: xxx # 告警策略 action id，在 https://console.bce.baidu.com/bcm/#/bcm/alarm/action/list 获取
7 scenes: [START] # START / SUCCESS / FAILURE / TIMEOUT
8 timeout: 10 # 用于 TIMEOUT 场景时的超时常用命令
工作流相关命令支持对工作流进行多种操作。常用命令速查如下：
表1 工作流常用命令
| 命令 | 作用 |
|---|---|
workflow submit |
打包上传 → 创建工作流 → 运行工作流，并输出运行实例 id |
workflow create |
仅创建工作流，不运行 |
workflow build |
仅生成节点 JSON，不上传、不创建工作流 |
workflow run |
运行指定工作流，输出运行实例 id |
workflow detail |
查看运行实例详情与状态 |
workflow stop |
停止整个工作流或单个运行实例 |
workflow pause |
暂停工作流实例到当前执行节点 |
workflow resume |
从暂停节点恢复运行 |
workflow rerun |
重跑某个运行实例 |
workflow schedule |
开启 / 关闭工作流调度 |
提交多节点（快速提交）
提交后自动完成打包上传 → 创建工作流 → 运行工作流，并输出运行实例 id：
Bash
1databuilder workflow submit --config-file ./workflow_conf.yaml也支持在命令行直接配置节点信息、工作流名称与项目路径：
Bash
1databuilder workflow submit --config-file ./workflow_conf.yaml \
2 --name '工作流名称' \
3 --project '工作台项目路径' \
4 --global-param bizdate=20260101 \
5 --node '{"name":"extract","working-dir":"./ws","entry":"python extract.py"}' \
6 --node '{"name":"transform","working-dir":"./ws","entry":"python transform.py","deps":["extract"]}'运行工作流
运行后输出对应运行实例的 id：
Bash
1databuilder workflow run --config-file ./workflow_conf.yaml --job-id=jid-xxx查看运行实例详情
查看运行实例的详情与状态：
Bash
1databuilder workflow detail --config-file ./workflow_conf.yaml --job-instance-id=jid-instanceid-xxx停止整个工作流
停止该工作流产生的全部运行实例：
Bash
1databuilder workflow stop --config-file ./workflow_conf.yaml --job-id=jid-xxx停止工作流实例
单独停止某一个运行实例：
Bash
1databuilder workflow stop --config-file ./workflow_conf.yaml --job-instance-id=jid-instanceid-xxx暂停工作流实例
暂停工作流实例到当前执行的节点：
Bash
1databuilder workflow pause --config-file ./workflow_conf.yaml --job-instance-id=jid-instanceid-xxx恢复暂停的工作流实例
从暂停的节点继续运行：
Bash
1databuilder workflow resume --config-file ./workflow_conf.yaml --job-instance-id=jid-instanceid-xxx重跑工作流
支持重跑某个运行实例，--strategy FROM_FIRST（从头跑，默认）｜FROM_FAILURE（从失败点跑）：
Bash
1databuilder workflow rerun --config-file ./workflow_conf.yaml --job-instance-id=jid-instanceid-xxx --strategy FROM_FIRST创建工作流
只创建工作流，不运行：
Bash
1databuilder workflow create --config-file ./workflow_conf.yaml创建工作流 JSON
仅创建节点 JSON，不上传也不创建工作流：
Bash
1databuilder workflow build --config-file ./workflow_conf.yaml控制工作流调度
开启或关闭工作流调度：
Bash
1databuilder workflow schedule --config-file ./workflow_conf.yaml --enable/disable --job-id=xxx评价此篇文章