alibabacloud-dataworks-datastudio-develop 是什么？

DataWorks 数据开发全流程。创建、配置、校验、发布数据开发节点和工作流。覆盖 Shell、SQL、Python、数据同步（DI）、Flink、EMR 等 130+ 种节点类型。支持周期工作流和手动工作流的编排，包含依赖管理、调度配置、发布上线全流程。支持 OpenAPI Mode（aliyun CLI / Python SDK）和 Git Mode 两种提交方式。当用户提到 DataWorks、数据开发节点、工作流、FlowSpec、调度任务、数据同步、ETL 管道、`.spec.json` 文件时使用本 Skill。即使用户没有明确说"DataWorks"，只要涉及阿里云数据开发、调度节点配置、FlowSpec 格式、数据集成任务编排，都应使用本 Skill。

alibabacloud-dataworks-datastudio-develop 怎么安装？

打开本页上方的「安装命令」标签页，复制命令到终端运行即可完成安装。

alibabacloud-dataworks-datastudio-develop 适合哪些场景？

主要用于「编程开发」等场景，适合需要 DataWorks 数据开发全流程。创建、配置、校验、发布数据开发节点和工作流。覆盖 Shell、SQL、Python、数据同步（DI）、Flink、EMR 等 130+ 种节点类型。支持周期工作流和手动工作流的编排，包含依赖管理、调度配置、发布上线全流程。支持 OpenAPI Mode（aliyun CLI / Python SDK）和 Git Mode 两种提交方式。当用户提到 DataWorks、数据开发节点、工作流、FlowSpec、调度任务、数据同步、ETL 管道、`.spec.json` 文件时使用本 Skill。即使用户没有明确说"DataWorks"，只要涉及阿里云数据开发、调度节点配置、FlowSpec 格式、数据集成任务编排，都应使用本 Skill。的任务。

alibabacloud-dataworks-datastudio-develop 收费吗？

本站收录的 Skill 安装包可免费获取使用；部分 Skill 依赖的第三方服务（如外部 API）可能需单独申请或订阅，以官方说明为准。

alibabacloud-dataworks-datastudio-develop：DataWorks 数据开发全流程。创建、配置、校验、发

DataWorks 数据开发

❗ 5 秒速览 —— 先读这里

凭证：先运行 aliyun configure list。CLI 几乎总是已预配置（STS token）。不要去搜索凭证文件——直接检查 CLI 配置即可。
先安装插件：运行 aliyun plugin install --names dataworks-public。所有命令均使用插件模式（kebab-case）：aliyun dataworks-public create-node ...
API 调用顺序：create-workflow-definition → create-node（逐节点）→ create-pipeline-run（部署）。更新时：update-node → create-pipeline-run。绝不使用 deploy-file、submit-file、create-file、create-business。
FlowSpec：直接复制下方「快速开始」中的 JSON，不要自行猜测格式。version 为 "2.0.0"，kind 为 "CycleWorkflow" 或 "Node"。常见错误值：apiVersion、type、Workflow、metadata —— 全部错误。
更新后必须发布：update-node → create-pipeline-run(type=Online) → 轮询 get-pipeline-run → exec-pipeline-run-stage。deploy-file 无效。

⚡ 强制要求：任何 API 调用前必读

以下绝对规则不是可选项——违反任何一条都意味着任务必然失败：

前置检查：需要 Aliyun CLI >= 3.3.3

运行 aliyun version 确认 >= 3.3.3。若未安装或版本过低，
运行 curl -fsSL --connect-timeout 10 --max-time 120 https://aliyuncli.alicdn.com/setup.sh | bash 更新，
或参阅 references/cli-installation-guide.md 获取安装说明。

Aliyun CLI 配置（首次使用前执行）：

安装插件（必需）

aliyun plugin install --names dataworks-public

更新插件（定期运行）

aliyun plugin update --names dataworks-public

AI-Mode：可用命令为 enable/disable/set-user-agent

本 Skill 关闭 AI-Mode（需要精确的参数控制）

aliyun configure ai-mode disable

不要运行：aliyun configure ai-mode enable

设置 user-agent 用于追踪

aliyun configure ai-mode set-user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop


逐命令 UA 标志（仅业务命令）：每次 `aliyun dataworks-public` 调用都必须追加 `--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop`。

0. **第一件事：检查 CLI 凭证。** 在执行任何 `aliyun` 命令之前，先运行 `aliyun configure list`。CLI 通常已预配置有效的 STS token 凭证——**不要**在检查之前就去搜索凭证文件（例如 `testconfig.json`）。如果 `aliyun configure list` 显示有效 profile，直接使用。如果存在多个 profile，运行 `aliyun configure switch --profile <name>` 选择正确的那个。优先级：优先选择名称包含 `dataworks` 的 profile（不区分大小写）；否则使用 `default`。**不要跳过此步骤。切换前不要运行任何 `aliyun dataworks-public` 命令。** 绝不读取 / 回显 / 打印 AK/SK 值。
1. **首次使用前安装插件。** 运行 `aliyun plugin install --names dataworks-public`。若已安装，运行 `aliyun plugin update --names dataworks-public` 确保为最新版本。插件提供 kebab-case 命令（`create-node`、`create-workflow-definition` 等），这是必需的调用形式。
2. **只使用插件模式（kebab-case）。** 每次 DataWorks API 调用都必须形如：`aliyun dataworks-public create-node --project-id ... --spec '...'`。绝不要使用 PascalCase RPC（`CreateNode`、`CreateWorkflowDefinition`）——始终使用插件模式。
3. **创建时只使用这些命令：** `create-workflow-definition` → `create-node`（逐节点，带 `--container-id`）→ `create-pipeline-run`（部署）。
4. **更新时只使用这些命令：** `update-node`（增量，`kind:Node`）→ `create-pipeline-run`（部署）。绝不要用 `import-workflow-definition`、`deploy-file` 或 `submit-file` 做更新或发布。
4a. **部署 / 发布时只使用这些命令：** `create-pipeline-run --type Online --object-ids <ID>` → `get-pipeline-run --id <PipelineRunId>`（轮询）→ `exec-pipeline-run-stage --id <PipelineRunId> --code <StageCode>`（推进）。**绝不使用** `deploy-file`、`submit-file`、`list-deployment-packages` 或 `get-deployment-package` —— 这些都是旧版 API，会失败。⚠️ `--object-ids` 是**空格分隔的裸 ID**（例如 `--object-ids 7567482277219412494`），**不是** JSON 数组字符串。写成 `'["ID"]'` 会导致 `未找到发布对象: [["ID"]]`，因为 CLI 会把方括号文本当作字面 ID 传入。

5. **如果 `create-workflow-definition` 或 `create-node` 返回错误，修改 Spec —— 不要回退到旧版 API。** 错误 58014884415 / `0x5083000000000005`（"Spec JSON parse failed"）意味着你的 FlowSpec JSON 格式有误（例如用了 `"kind":"Workflow"` 而非 `"kind":"CycleWorkflow"`，或用了 `"apiVersion"` 而非 `"version"`，或用了扁平的 `{"type":"SHELL","content":"..."}` 结构而非 `{"version":"2.0.0","kind":"Node","spec":{"nodes":[...]}}` 结构）。**停止猜测，从下方「快速开始」复制确切的 Spec，然后只修改你需要的值。**
6. **直接运行 CLI 命令 —— 不要创建包装脚本。** 绝不要创建 `.sh` 脚本来批量调用 API。在 shell 中直接运行每条 `aliyun` 命令。包装脚本会增加复杂度并掩盖错误。
7. **仅在本地保存文件不等于完成。** 只有当 API 返回成功响应（例如 `create-workflow-definition`/`create-node` 返回 `{"Id": "..."}`）时任务才算完成。只把 JSON 文件写到磁盘而不调用 API，意味着工作流 / 节点并未创建。没有真实 API 响应就绝不声称成功。
8. **绝不模拟、mock 或伪造 API 响应。** 如果凭证缺失、CLI 配置错误，或 API 调用返回错误——向用户报告确切错误信息并**停止**。绝不生成虚假 JSON 响应、编写模拟文档、回显硬编码输出，或以任何形式声称成功。模拟的成功比明确的失败更糟。
9. **凭证失败 = 硬停止。** 如果 `aliyun configure list` 显示凭证为空或无效，或任何 CLI 调用返回 `InvalidAccessKeyId`、`access_key_id must be assigned` 或类似鉴权错误——**立即停止**。告知用户在本会话之外配置有效凭证。不要尝试变通方案（手工写 config.json、使用占位凭证、无鉴权继续）。在凭证验证可用之前，不得尝试任何后续 API 调用。
10. **只使用本文档列出的 API。** 你调用的每个 API 都必须出现在下方 API 快速参考表中。如果需要表中未列出的操作，再查一遍表——该操作很可能以另一个名称存在。**绝不臆造 API 名称**（例如 `CreateDeployment`、`ApproveDeployment`、`DeployNode` 都不存在）。如果找不到正确的 API，询问用户。

**如果你发现自己正在输入以下任何旧版命令，立即停止并重读下方「快速开始」：**
`create-file`、`create-business`、`create-folder`、`--file-type`、`/bizroot`、`/workflowroot`、`deploy-file`、`submit-file`、`list-files`、`get-file`、`list-deployment-packages`、`get-deployment-package`、`create-deployment`、`approve-deployment`、`deploy-node`、`create-flow`、`create-file-depends`、`create-schedule`

### ⚠️ FlowSpec 反模式

Agent 常臆造错误的 FlowSpec 字段。正确格式见下方「快速开始」。

| ❌ 错误 | ✅ 正确 | 说明 |
|----------|-----------|-------|
| `"apiVersion": "v1"` 或 `"apiVersion": "dataworks.aliyun.com/v1"` | `"version": "2.0.0"` | FlowSpec 用 `version`，不是 `apiVersion` |
| `"kind": "Flow"` 或 `"kind": "Workflow"` | `"kind": "CycleWorkflow"`（工作流）或 `"kind": "Node"`（节点） | 只有 `Node`、`CycleWorkflow`、`ManualWorkflow` 有效。单独的 `"Workflow"` **无效** |
| `"metadata": {"name": "..."}` | `"spec": {"workflows": [{"name": "..."}]}` | FlowSpec 没有 `metadata` 字段；name 放在 `spec.workflows[0]` 或 `spec.nodes[0]` 内 |
| `"type": "SHELL"`（节点级） | `"script": {"runtime": {"command": "DIDE_SHELL"}}` | 节点类型放在 `script.runtime.command` |
| `"schedule": {"cron": "..."}` | `"trigger": {"cron": "...", "type": "Scheduler"}` | 调度用 `trigger`，不是 `schedule` |
| `"script": {"content": "..."}` 缺少 `path` | `"script": {"path": "node_name", ...}` | `script.path` 始终必填 |

### 🚀 快速开始：端到端创建工作流

完整可运行示例 —— 创建一个含 2 个依赖节点的周期工作流：

第 1 步：创建工作流容器

aliyun dataworks-public create-workflow-definition \

--project-id 585549 \

--spec '{"version":"2.0.0","kind":"CycleWorkflow","spec":{"workflows":[{"name":"my_etl_workflow","script":{"path":"my_etl_workflow","runtime":{"command":"WORKFLOW"}}}]}}' \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

→ 返回 {"Id": "WORKFLOW_ID", ...}

第 2 步：在工作流内创建上游节点（Shell）

重要：创建前先确认输出名 "my_project.check_data" 未被其他节点占用（list-nodes）

aliyun dataworks-public create-node \

--project-id 585549 \

--scene DATAWORKS_PROJECT \

--container-id WORKFLOW_ID \

--spec '{"version":"2.0.0","kind":"Node","spec":{"nodes":[{"name":"check_data","id":"check_data","script":{"path":"check_data","runtime":{"command":"DIDE_SHELL"},"content":"#!/bin/bash\necho done"},"outputs":{"nodeOutputs":[{"data":"my_project.check_data","artifactType":"NodeOutput"}]}}]}}' \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

→ 返回 {"Id": "NODE_A_ID", ...}

第 3 步：创建依赖上游的下游节点（SQL）

依赖说明："nodeId" 是当前节点的名称（自引用），"output" 是上游节点的输出

aliyun dataworks-public create-node \

--project-id 585549 \

--scene DATAWORKS_PROJECT \

--container-id WORKFLOW_ID \

--spec '{"version":"2.0.0","kind":"Node","spec":{"nodes":[{"name":"transform_data","id":"transform_data","script":{"path":"transform_data","runtime":{"command":"ODPS_SQL"},"content":"SELECT 1;"},"outputs":{"nodeOutputs":[{"data":"my_project.transform_data","artifactType":"NodeOutput"}]}}],"dependencies":[{"nodeId":"transform_data","depends":[{"type":"Normal","output":"my_project.check_data"}]}]}}' \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

第 4 步：设置工作流调度（每天 00:30）

aliyun dataworks-public update-workflow-definition \

--project-id 585549 \

--id WORKFLOW_ID \

--spec '{"version":"2.0.0","kind":"CycleWorkflow","spec":{"workflows":[{"name":"my_etl_workflow","script":{"path":"my_etl_workflow","runtime":{"command":"WORKFLOW"}},"trigger":{"cron":"00 30 00 * * ?","timezone":"Asia/Shanghai","type":"Scheduler"}}]}}' \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

第 5 步：将工作流部署上线（必需 —— 未部署前工作流不会生效）

aliyun dataworks-public create-pipeline-run \

--project-id 585549 \

--type Online --object-ids WORKFLOW_ID \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

→ 返回 {"Id": "PIPELINE_RUN_ID", ...}

然后轮询 get-pipeline-run，并用 exec-pipeline-run-stage 推进阶段

（完整轮询流程见下方「发布与部署」章节）


> **关键模式**：`create-workflow-definition` → `create-node`（带 `--container-id` + outputs.nodeOutputs）→ `update-workflow-definition`（加 trigger）→ **`create-pipeline-run`（部署）**。工作流内每个节点都必须有 `outputs.nodeOutputs`。**工作流在通过 `create-pipeline-run` 部署之前不会生效。**
>
> **依赖接线摘要**：在 `spec.dependencies` 中，`nodeId` 是**当前节点自身的名称**（自引用，不是上游节点），`depends[].output` 是**上游节点的输出**（`projectIdentifier.upstream_node_name`）。上游节点的 `outputs.nodeOutputs[].data` 值与下游节点的 `depends[].output` 值必须**逐字符完全一致**，否则依赖会静默失败。

核心工作流

环境发现（创建前必需）

第 0 步 —— 检查 CLI 凭证（必须是第一个动作）：

运行 aliyun configure list。CLI 几乎总是已预配置 STS token 凭证——不要在运行此命令前声称"我没有凭证"或搜索凭证文件。如果输出显示 Valid profile，说明你有可用凭证——立即继续。如果存在多个 profile，运行 aliyun configure switch --profile <name>（优先 dataworks 命名的 profile，否则 default）。在此步骤完成前，不得运行任何 aliyun dataworks-public 命令。

如果凭证为空或无效，就此停止。 不要继续任何 API 调用。向用户报告错误，并指示其在本会话之外配置有效凭证（通过 aliyun configure 或环境变量）。不要尝试手工写配置文件或使用占位值等变通方案。

在创建节点或工作流之前，先了解项目的现有环境。建议使用 subagent 执行查询，只向主 Agent 返回摘要，避免原始数据占用过多上下文。

Subagent 任务：

调用 list-workflow-definitions 获取工作流列表
调用 list-nodes 获取现有节点列表
调用 list-data-sources 和 list-compute-resources 获取所有可用数据源和计算引擎绑定（EMR、Hologres、StarRocks 等）。list-compute-resources 补充 list-data-sources，后者可能不返回计算引擎类型的资源
返回摘要（不要返回原始数据）：

工作流清单：名称 + 包含节点数 + 类型（周期 / 手动）
与当前任务相关的现有节点：名称 + 类型 + 所属工作流
可用数据源 + 计算资源（名称、类型）—— 合并两个列表
建议的目标工作流（若可从任务描述推断）

基于摘要，主 Agent 决定：目标工作流（现有或新建，用户决定）、节点命名（遵循现有约定）、依赖关系（从 SQL 引用和现有节点推断）。

创建前冲突检查（必需，适用于所有对象类型）：

名称重复检查：创建任何对象前，用对应的 list 命令检查同名对象是否已存在：

工作流 → list-workflow-definitions
节点 → list-nodes（节点名在项目内全局唯一）
资源 → list-resources
函数 → list-functions
组件 → list-components

已存在对象的处理：告知用户并询问如何处理（使用现有 / 重命名 / 更新现有）。禁止直接删除现有对象
输出名冲突检查（关键）：节点的 outputs.nodeOutputs[].data（格式 ${projectIdentifier}.NodeName）必须在项目内全局唯一，即使跨不同工作流。用 list-nodes --name NodeName 并检查响应中的 Outputs.NodeOutputs[].Data 验证。如果输出名与现有节点冲突，必须在创建前解决冲突——否则部署会失败并报 "can not exported multiple nodes into the same output"（见 troubleshooting.md #11b）

确定性程度决定交互方式：

确定的信息 → 直接使用，不询问用户
有把握的推断 → 继续，并在输出中说明推理
不确定的信息 → 必须询问用户

创建节点

统一工作流：无论 OpenAPI Mode 还是 Git Mode，都生成相同的本地文件结构。

⚠️ 必须先创建本地文件，再调用 API
无论使用 OpenAPI Mode 还是 Git Mode，都必须先在本地创建完整的节点文件夹（spec.json + 代码文件 + dataworks.properties），经过 build.py 合并和 validate.py 校验后，再调用 create-node/create-workflow-definition 命令。禁止跳过本地文件创建直接调用 API 构造 Spec。

#### 第 1 步：创建节点目录和三个文件

一个文件夹 = 一个节点，包含三个文件：

my_node/
├── my_node.spec.json          # FlowSpec 节点定义
├── my_node.sql                # 代码文件（扩展名取决于 contentFormat）
└── dataworks.properties       # 运行时配置（实际值）

spec.json —— 从 references/nodetypes/{category}/{TYPE}.md 复制最小 Spec，修改 name 和 path，用 ${spec.xxx} 占位符引用 properties 中的值。如果用户指定了 trigger、dependencies、rerunTimes 等，也加入 spec。

代码文件 —— 根据节点类型文档中的 contentFormat 确定格式（sql/shell/python/json/empty）；根据 extension 字段确定扩展名。

dataworks.properties —— 填入实际值：

projectIdentifier=<实际项目标识符>
spec.datasource.name=<实际数据源名称>
spec.runtimeResource.resourceGroup=<实际资源组标识符>

不确定的值不要填——省略后服务端会自动使用项目默认值。

参考示例：assets/templates/

#### 第 2 步：提交

默认使用 OpenAPI（除非用户明确说"提交到 Git"）：

用 build.py 将三个文件合并为 API 输入：

   python $SKILL/scripts/build.py ./my_node > /tmp/spec.json

build.py 做三件事（无第三方依赖；若出错，参考源码手动执行）：

读取 dataworks.properties → 替换 spec.json 中的 ${spec.xxx} 和 ${projectIdentifier} 占位符
读取代码文件（包括 DI .json 代码文件）→ 替换占位符 → 嵌入 script.content
输出合并后的完整 JSON

提交前校验 spec：

   python $SKILL/scripts/validate.py ./my_node

提交前验证（强制） —— 调用 create-node 之前，必须通过 API 验证以下信息确实存在且正确：

环境验证（首次提交前执行一次）：

[ ] runtimeResource.resourceGroup —— 调用 list-resource-groups 确认资源组存在，使用返回的资源组 ID（如 Serverless_res_group_...），不要使用人类可读名称（如 cx_res_4）。如不确定，省略让服务端使用项目默认值
[ ] datasource —— 计算引擎节点（ODPS_SQL、HOLOGRES_SQL 等）需要数据源。调用 list-data-sources 或 list-compute-resources 确认数据源名称和类型匹配。如不确定，省略让服务端使用项目默认值

Spec 内容审查（每个节点提交前）：

[ ] script.runtime.command 与预期节点类型匹配（查 references/nodetypes/{category}/{TYPE}.md）
[ ] script.content —— 代码节点确认合并后的 spec 含非空代码。对 DI 节点尤其要注意，script.content 必须是有效的 DIJob JSON 字符串，顶层为扁平键 type、version、steps、order、setting、extend —— 它不是旧版 DataX 形状 {"job":{"content":[{"reader":{"plugin":...}}]}}。如果你生成的内容有顶层 "job" 包装或 content[].reader.plugin，说明你在使用训练记忆中的错误格式；在调用 CreateNode 前重写为符合 references/nodetypes/data_integration/DI.md 和 DATAX.md 的格式
[ ] trigger —— 工作流节点：省略以继承工作流调度；仅在用户明确指定逐节点调度时设置。独立节点：若用户指定了调度则设置
[ ] outputs.nodeOutputs —— 工作流节点必需。格式：{"data":"${projectIdentifier}.NodeName","artifactType":"NodeOutput"}。验证输出名在项目内全局唯一（list-nodes --name）
[ ] dependencies —— nodeId 必须是当前节点自身的名称（自引用）。depends[].output 必须完全匹配上游节点的 outputs.nodeOutputs[].data。每个工作流节点都必须有 dependencies：根节点（无上游）必须依赖 ${projectIdentifier}_root（下划线，不是点）；下游节点依赖上游输出。没有任何 dependencies 条目的工作流节点会成为孤儿节点
[ ] 无臆造字段 —— 对照上方 FlowSpec 反模式表；删除 references/flowspec-guide.md 中未记录的任何字段

调用 API 提交（参阅 references/api/CreateNode.md）：

   aliyun dataworks-public create-node \
     --project-id $PROJECT_ID \
     --scene DATAWORKS_PROJECT \
     --spec "$(cat /tmp/spec.json)" \
     --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

> 注意：需要 dataworks-public 插件（见上方 Aliyun CLI 配置）。如果命令未找到，先安装插件。绝不使用旧版命令（create-file/create-folder）。

> 沙箱兜底：如果 $(cat ...) 被阻止，使用 Python subprocess.run(['aliyun', 'dataworks-public', 'create-node', '--project-id', str(PID), '--scene', 'DATAWORKS_PROJECT', '--spec', spec_str, '--user-agent', 'AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop'])。

要放入工作流内，加上 --container-id $WorkflowId

Git Mode（用户明确要求时）：git add ./my_node && git commit，DataWorks 会自动同步并替换占位符

最小必填字段（实践验证，130+ 类型通用）：

name —— 节点名称
id —— 必须设为与 name 相等。确保 spec.dependencies[*].nodeId 能匹配。没有显式 id，API 可能静默丢弃依赖
script.path —— 脚本路径，必须以节点名结尾；服务端自动前置工作流前缀
script.runtime.command —— 节点类型（例如 ODPS_SQL、DIDE_SHELL）

可复制的最小节点 Spec（Shell 节点示例）：

{"version":"2.0.0","kind":"Node","spec":{"nodes":[{
  "name":"my_shell_node","id":"my_shell_node",
  "script":{"path":"my_shell_node","runtime":{"command":"DIDE_SHELL"},"content":"#!/bin/bash\necho hello"}
}]}}

其他字段非必填；服务端会自动填充项目默认值：

datasource、runtimeResource —— 如不确定，不要传；服务端自动绑定项目默认值
trigger —— 不传则继承工作流调度。仅在用户指定时传
dependencies、rerunTimes 等 —— 仅在用户指定时传
outputs.nodeOutputs —— 独立节点可选；工作流内节点必需（{"data":"${projectIdentifier}.NodeName","artifactType":"NodeOutput"}），否则下游依赖会静默失败。⚠️ 输出名（${projectIdentifier}.NodeName）必须在项目内全局唯一——如果另一个节点（即使在不同工作流）已使用相同输出名，部署会失败并报 "can not exported multiple nodes into the same output"。创建前务必用 list-nodes 检查

创建工作流

⚠️ 工作流创建也必须先创建本地文件。 先为工作流内每个节点创建本地文件夹（spec.json + 代码文件 + dataworks.properties），验证通过后再依次调用 API。首次提交前，必须通过 list-resource-groups、list-data-sources 等命令确认资源组、数据源等环境信息存在且正确。目录结构详见 workflow-guide.md。

创建工作流定义（最小 spec）：

   {"version":"2.0.0","kind":"CycleWorkflow","spec":{"workflows":[{
     "name":"workflow_name","script":{"path":"workflow_name","runtime":{"command":"WORKFLOW"}}
   }]}}

调用 create-workflow-definition → 返回 WorkflowId

按依赖顺序创建节点（每个节点传 --container-id WorkflowId）

每个节点创建前：确认 ${projectIdentifier}.NodeName 未被项目中任何现有节点用作输出（用 list-nodes --name 并检查 Outputs.NodeOutputs[].Data）。输出名重复会导致部署失败
每个节点的 spec 必须包含 outputs.nodeOutputs：{"data":"${projectIdentifier}.NodeName","artifactType":"NodeOutput"}
下游节点在 spec.dependencies 中声明依赖：nodeId = 当前节点自身名称（自引用），depends[].output = 上游节点输出（见 workflow-guide.md）

验证依赖（所有节点创建后强制） —— 对每个下游节点调用 list-node-dependencies --id <NodeID>。如果 TotalCount 为 0 但节点本应有上游依赖，说明 create-node 静默丢弃了它们。立即用 update-node 修复，使用 spec.dependencies（见下方「更新依赖」）。在所有依赖确认前不要继续部署
设置调度 —— 用 update-workflow-definition 加 trigger（若用户指定了调度）
部署上线（必需） —— create-pipeline-run --type Online --object-ids <WorkflowId> → 轮询 get-pipeline-run --id <PipelineRunId> → 用 exec-pipeline-run-stage --id <PipelineRunId> --code <StageCode> 推进阶段。工作流在部署前不会生效。 不要跳过此步骤，也不要让用户手动执行。

详细指南和可复制的完整节点 Spec 示例（含 outputs 和 dependencies）：references/workflow-guide.md

更新现有节点

必须使用增量更新 —— 只传节点 id + 要修改的字段：

{"version":"2.0.0","kind":"Node","spec":{"nodes":[{
  "id":"NodeID",
  "script":{"content":"new code"}
}]}}

⚠️ 关键：update-node 始终使用 "kind":"Node"，即使节点属于某个工作流。不要用 "kind":"CycleWorkflow" —— 那仅用于工作流级操作（update-workflow-definition）。

不要传未变更的字段如 datasource 或 runtimeResource（服务端可能已修正这些值；回传可能导致错误）。

⚠️ 更新依赖：要通过 update-node 修复或更改节点依赖，使用 spec.dependencies。示例：
```json
{"version":"2.0.0","kind":"Node","spec":{"nodes":[{"id":"NodeID"}],"dependencies":[{"nodeId":"current_node_name","depends":[{"type":"Normal","output":"project.upstream_node"}]}]}}
```

#### 更新 + 重新发布工作流

修改现有节点并部署变更的完整端到端流程：

查找节点 —— list-nodes(--name xxx) → 获取 Node ID
更新节点 —— update-node 传增量 spec（kind:Node，只有 id + 变更字段）
发布 —— create-pipeline-run --type Online --object-ids <PublishObjectId> → 轮询 get-pipeline-run --id <PipelineRunId> → 用 exec-pipeline-run-stage --id <PipelineRunId> --code <StageCode> 推进阶段。⚠️ <PublishObjectId> 选择规则：如果节点位于工作流内（其 path（来自 get-node）含 /，例如 wf_name/node_name），<PublishObjectId> 必须是工作流 ID，而非节点 ID —— API 会以 未找到发布对象 拒绝工作流内节点 ID。只有独立节点（根路径，无 /）才用自己的 ID。

第 1 步：查找节点

aliyun dataworks-public list-nodes --project-id $PID --name "my_node" --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

→ 记下响应中的节点 Id

第 2 步：更新（增量 —— 只有 id + 变更字段）

aliyun dataworks-public update-node --project-id $PID --id $NODE_ID \

--spec '{"version":"2.0.0","kind":"Node","spec":{"nodes":[{"id":"'$NODE_ID'","script":{"content":"SELECT 1;"}}]}}' \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

第 3 步：发布（见下方「发布与部署」）

重要：如果 $NODE_ID 的 `path`（来自 get-node）含 "/"，说明它在工作流内 ——

将下面的 $NODE_ID 替换为工作流 ID。独立节点（根路径）用自己的 ID。

aliyun dataworks-public create-pipeline-run --project-id $PID \

--type Online --object-ids $NODE_ID \

--user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop


> **`update-node` 后的常见错误路径**（全部禁止）：
> - ❌ `deploy-file` / `submit-file` —— 旧版 API，会失败或行为异常
> - ❌ `import-workflow-definition` —— 仅用于初始批量导入，不用于更新或发布
> - ❌ `list-files` / `get-file` —— 旧版文件模型，改用 `list-nodes` / `get-node`
> - ✅ `create-pipeline-run` → `get-pipeline-run` → `exec-pipeline-run-stage`

### 发布与部署

> **⚠️ 绝不使用 `deploy-file`、`submit-file`、`list-deployment-packages`、`get-deployment-package`、`list-files` 或 `get-file` 进行部署。** 这些都是旧版 API。只用：`create-pipeline-run` → `get-pipeline-run` → `exec-pipeline-run-stage`。

发布是一个异步多阶段流水线：

1. `create-pipeline-run --type Online --object-ids <ID>` → 从 `Id`

DataWorks 数据开发

❗ 5 秒速览 —— 先读这里

⚡ 强制要求：任何 API 调用前必读

安装插件（必需）

更新插件（定期运行）

AI-Mode：可用命令为 enable/disable/set-user-agent

本 Skill 关闭 AI-Mode（需要精确的参数控制）

不要运行：aliyun configure ai-mode enable

设置 user-agent 用于追踪

第 1 步：创建工作流容器

→ 返回 {"Id": "WORKFLOW_ID", ...}

第 2 步：在工作流内创建上游节点（Shell）

重要：创建前先确认输出名 "my_project.check_data" 未被其他节点占用（list-nodes）

→ 返回 {"Id": "NODE_A_ID", ...}

第 3 步：创建依赖上游的下游节点（SQL）

依赖说明："nodeId" 是当前节点的名称（自引用），"output" 是上游节点的输出

第 4 步：设置工作流调度（每天 00:30）

第 5 步：将工作流部署上线（必需 —— 未部署前工作流不会生效）

→ 返回 {"Id": "PIPELINE_RUN_ID", ...}

然后轮询 get-pipeline-run，并用 exec-pipeline-run-stage 推进阶段

（完整轮询流程见下方「发布与部署」章节）

核心工作流

环境发现（创建前必需）

创建节点

创建工作流

更新现有节点

第 1 步：查找节点

→ 记下响应中的节点 Id

第 2 步：更新（增量 —— 只有 id + 变更字段）

第 3 步：发布（见下方「发布与部署」）

重要：如果 $NODE_ID 的 path（来自 get-node）含 "/"，说明它在工作流内 ——

将下面的 $NODE_ID 替换为工作流 ID。独立节点（根路径）用自己的 ID。

◯ 评论 0

重要：如果 $NODE_ID 的 `path`（来自 get-node）含 "/"，说明它在工作流内 ——