DataWorks 数据开发

❗ 5 秒摘要——请先阅读

凭证：请先运行 aliyun configure list。CLI 几乎总是已预配置（STS 令牌）。不得搜索凭证文件，只需检查 CLI 配置。

请先安装插件：运行 aliyun plugin install --names dataworks-public。所有命令均使用插件模式（kebab-case）：aliyun dataworks-public create-node ...

APIs：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 前必须阅读

以下绝对规则并非可选——违反任何一条都将导致任务失败：

预检查：必须使用阿里云 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 获取安装说明。

阿里云 CLI 设置（首次使用前执行）：

# Plugin install (required)
aliyun plugin install --names dataworks-public
# Plugin update (run periodically)
aliyun plugin update --names dataworks-public
# AI-Mode: available commands are enable/disable/set-user-agent
# This skill DISABLES AI-Mode (exact parameter control required)
aliyun configure ai-mode disable
# Do NOT run: aliyun configure ai-mode enable
# Set user-agent for tracking
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。

1. 首先：检查 CLI 凭证。 在执行任何 aliyun 命令前，运行 aliyun configure list。CLI 通常已预配置有效的 STS 令牌凭证——检查前不得搜索凭证文件（例如 testconfig.json）。如果 aliyun configure list 显示有效的配置文件，请直接使用。如果存在多个配置文件，请运行 aliyun configure switch --profile <name> 选择正确的配置文件。优先级：优先选择名称中包含 dataworks 的配置文件（不区分大小写）；否则使用 default。不得跳过此步骤。完成切换前，不得运行任何 aliyun dataworks-public 命令。 严禁读取、回显或打印 AK/SK 值。
2. 首次使用前安装插件。 运行 aliyun plugin install --names dataworks-public。如果已安装，请运行 aliyun plugin update --names dataworks-public 以确保使用最新版本。该插件提供 kebab-case 命令（create-node、create-workflow-definition 等），必须以此形式调用。
3. 仅使用插件模式（kebab-case）。 每次 DataWorks API 调用都必须采用如下形式：aliyun dataworks-public create-node --project-id ... --spec '...'。禁止使用 PascalCase RPC（CreateNode、CreateWorkflowDefinition）——始终使用插件模式。
4. 创建时只能使用以下命令： create-workflow-definition → create-node（每个节点执行一次，并使用 --container-id）→ create-pipeline-run（部署）。
5. 更新时只能使用以下命令： 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——这些都是会失败的旧版 APIs。⚠️ --object-ids 必须是以空格分隔的裸 ID（例如 --object-ids 7567482277219412494），不得是 JSON 数组字符串。若将其包装为 '["ID"]'，将产生 未找到发布对象: [["ID"]]，因为 CLI 会将方括号字面文本作为 ID 传递。

1. 如果 create-workflow-definition 或 create-node 返回错误，必须修正 Spec——不得回退使用旧版 APIs。 错误 58014884415 / 0x5083000000000005（“Spec JSON 解析失败”）表示 FlowSpec JSON 格式错误（例如，使用了 "kind":"Workflow" 而不是 "kind":"CycleWorkflow"，或使用了 "apiVersion" 而不是 "version"，或使用了扁平的 {"type":"SHELL","content":"..."} 结构，而不是 {"version":"2.0.0","kind":"Node","spec":{"nodes":[...]}} 结构）。停止猜测，从下方的快速入门中复制完全相同的 Spec，然后仅修改所需的值。
2. 直接运行 CLI 命令——不得创建包装脚本。 严禁创建 .sh 脚本来批量执行 API 调用。直接在 shell 中逐条运行 aliyun 命令。包装脚本会增加复杂性并掩盖错误。
3. 将文件保存在本地并不代表完成。 只有当 API 返回成功响应时，任务才算完成（例如，create-workflow-definition/create-node 返回 {"Id": "..."}）。仅将 JSON 文件写入磁盘而不调用 API，意味着工作流/节点并未创建。没有真实的 API 响应时，禁止声称成功。
4. 严禁模拟、仿制或伪造 API 响应。 如果缺少凭证、CLI 配置错误或 API 调用返回错误，请向用户报告确切的错误消息并停止。不得生成虚假的 JSON 响应、编写模拟文档、回显硬编码输出或以任何形式声称成功。模拟成功比明确失败更糟糕。
5. 凭证失败 = 必须停止。 如果 aliyun configure list 显示凭证为空或无效，或者任何 CLI 调用返回 InvalidAccessKeyId、access_key_id must be assigned 或类似的身份验证错误，请立即停止。告知用户在此会话之外配置有效凭证。不得尝试变通方法（手动编写 config.json、使用占位凭证、未经身份验证继续操作）。在凭证经验证有效前，不得尝试任何后续 API 调用。
6. 只能使用本文列出的 APIs。 调用的每个 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 反模式

Agents 经常臆造错误的 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 字段；名称应放在 spec.workflows[0] 或 spec.nodes[0] 中 | | "type": "SHELL"（位于节点级别） | "script": {"runtime": {"command": "DIDE_SHELL"}} | 节点类型应放在 script.runtime.command 中 | | "schedule": {"cron": "..."} | "trigger": {"cron": "...", "type": "Scheduler"} | 调度使用 trigger，而不是 schedule | | 缺少 path 的 "script": {"content": "..."} | "script": {"path": "node_name", ...} | script.path 始终为必填项 |

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

完整的可运行示例——创建一个包含 2 个具有依赖关系的节点的定时工作流：

# Step 1: Create the workflow container
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
# → Returns {"Id": "WORKFLOW_ID", ...}

# Step 2: Create upstream node (Shell) inside the workflow
# IMPORTANT: Before creating, verify output name "my_project.check_data" is not already used by another node (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
# → Returns {"Id": "NODE_A_ID", ...}

# Step 3: Create downstream node (SQL) with dependency on upstream
# NOTE on dependencies: "nodeId" is the CURRENT node's name (self-reference), "output" is the UPSTREAM node's 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

# Step 4: Set workflow schedule (daily at 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

# Step 5: Deploy the workflow online (REQUIRED — workflow is not active until deployed)
aliyun dataworks-public create-pipeline-run \
  --project-id 585549 \
  --type Online --object-ids WORKFLOW_ID \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop
# → Returns {"Id": "PIPELINE_RUN_ID", ...}
# Then poll get-pipeline-run and advance stages with exec-pipeline-run-stage
# (see "Publishing and Deploying" section below for full polling flow)

关键模式：create-workflow-definition → create-node（带 --container-id + outputs.nodeOutputs）→ update-workflow-definition（添加触发器）→ 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 令牌凭据——在运行此命令之前，不得声称“我没有凭据”或搜索凭据文件。如果输出中显示一个 Valid 配置文件，说明凭据可用——立即继续。如果存在多个配置文件，请运行 aliyun configure switch --profile <name>（优先选择名称包含 dataworks 的配置文件，否则选择 default）。在此之前不得运行任何 aliyun dataworks-public 命令。

如果凭据为空或无效，请在此处停止。不得继续进行任何 API 调用。向用户报告错误，并指示他们在此会话之外（通过 aliyun configure 或环境变量）配置有效凭据。不得尝试手动写入配置文件或使用占位值等变通方法。

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

子 Agent 任务：

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

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

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

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

1. 名称重复检查：创建任何对象前，使用相应的列表命令检查是否已存在同名对象：

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

1. 现有对象的处理：告知用户并询问如何继续（使用现有对象 / 重命名 / 更新现有对象）。禁止直接删除现有对象
2. 输出名称冲突检查（关键）：节点的 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 模式还是 Git 模式，都生成相同的本地文件结构。

⚠️ 必须先创建本地文件，再调用 API

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

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

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

my_node/
├── my_node.spec.json          # FlowSpec node definition
├── my_node.sql                # Code file (extension based on contentFormat)
└── dataworks.properties       # Runtime configuration (actual values)

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=<actual project identifier>
spec.datasource.name=<actual datasource name>
spec.runtimeResource.resourceGroup=<actual resource group identifier>

不得填写不确定的值——如果省略，服务器会自动使用项目默认值。

参考示例：assets/templates/

步骤 2：提交

默认为 OpenAPI（除非用户明确表示“提交到 Git”）：

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

``bash python $SKILL/scripts/build.py ./my_node > /tmp/spec.json `` build.py 执行三项操作（无第三方依赖；如果发生错误，请参考源代码手动执行）：

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

1. 提交前验证 Spec：

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

2. 提交前验证（强制要求） — 调用 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 完全匹配。每个工作流节点都必须具有依赖关系：根节点（无上游节点）必须依赖 ${projectIdentifier}_root（使用下划线，而不是点号）；下游节点依赖上游输出。没有依赖关系条目的工作流节点将成为孤立节点
• [ ] 不得虚构字段——对照上方的 FlowSpec 反模式表；移除所有未在 references/flowspec-guide.md 中记录的字段

1. 调用 API 提交（请参阅 [references/api/CreateNode.md](references/api/CreateNode.md)）：

``bash 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 插件（请参阅上方的阿里云 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'])。

1. 若要放入工作流，请添加 --container-id $WorkflowId

Git 模式（用户明确请求时）：执行 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）必须在项目内全局唯一——如果另一个节点（即使位于不同工作流中）已使用相同的输出名称，部署将失败并提示“无法将多个节点导出到同一输出”。创建前务必使用 list-nodes 检查

创建工作流

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

1. 创建工作流定义（最小 Spec）：

``json {"version":"2.0.0","kind":"CycleWorkflow","spec":{"workflows":[{ "name":"workflow_name","script":{"path":"workflow_name","runtime":{"command":"WORKFLOW"}} }]}} ` 调用 create-workflow-definition` → 返回 WorkflowId

2. 按依赖顺序创建节点（每个节点均传入 --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）

1. 验证依赖关系（创建所有节点后必须执行）——对每个下游节点调用 list-node-dependencies --id <NodeID>。如果 TotalCount 为 0，但该节点应有上游依赖关系，则 create-node 已静默丢弃这些依赖关系。立即修复：使用 update-node 的 spec.dependencies（请参阅下文“更新依赖关系”）。在确认所有依赖关系之前，不得继续部署
2. 设置调度——如果用户指定了调度，请使用 update-workflow-definition 设置 trigger
3. 上线部署（强制要求）——create-pipeline-run --type Online --object-ids <WorkflowId> → 轮询 get-pipeline-run --id <PipelineRunId> → 使用 exec-pipeline-run-stage --id <PipelineRunId> --code <StageCode> 推进各阶段。工作流在部署前不会生效。 不得跳过此步骤，也不得让用户手动执行。

详细指南和可复制的完整节点 Spec 示例（包括输出和依赖关系）：[references/workflow-guide.md](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"}]}]}}

```

更新并重新发布工作流

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

1. 查找节点——list-nodes(--name xxx) → 获取节点 ID
2. 更新节点——使用增量配置调用 update-node（kind:Node，仅包含 id + 已更改的字段）
3. 发布——create-pipeline-run --type Online --object-ids <PublishObjectId> → 轮询 get-pipeline-run --id <PipelineRunId> → 使用 exec-pipeline-run-stage --id <PipelineRunId> --code <StageCode> 推进各阶段。⚠️ <PublishObjectId> 选择规则：如果节点位于工作流内（get-node 返回的 path 中包含 /，例如 wf_name/node_name），则 <PublishObjectId> 必须是工作流 ID，不得是节点 ID——API 会拒绝工作流内部节点 ID，并返回 未找到发布对象。只有独立节点（根路径中不含 /）才使用自身的 ID。

# Step 1: Find the node
aliyun dataworks-public list-nodes --project-id $PID --name "my_node" --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop
# → Note the node Id from the response

# Step 2: Update (incremental — only id + changed fields)
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

# Step 3: Publish (see "Publishing and Deploying" below)
# IMPORTANT: if $NODE_ID's `path` from get-node contains a "/", it is in a workflow —
# replace $NODE_ID below with the workflow ID. Standalone nodes (root path) take their own 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——旧版 APIs，将失败或产生意外行为

- ❌ 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。 这些均为旧版 APIs。只能使用：create-pipeline-run → get-pipeline-run → exec-pipeline-run-stage。

发布采用异步多阶段流水线：

1. create-pipeline-run --type Online --object-ids <ID> → 从 Id 字段获取 PipelineRunId。⚠️ --object-ids 接收的是以空格分隔的裸 ID，不得传入 JSON 数组字符串。直接传入 ID：--object-ids 7567482277219412494。将其包装为 '["..."]' 会导致 CLI 将带方括号的文本作为字面 ID 发送，从而产生 未找到发布对象: [["..."]]。API 仅发布第一个 ID 及其子实体——对于相互独立的对象，请分别调用 create-pipeline-run。对于工作流内的节点：传入工作流 ID，不得传入节点 ID（工作流内部节点 ID 会因同样的 未找到发布对象 错误而被拒绝）
2. 轮询 get-pipeline-run --id <PipelineRunId> → 检查 Pipeline.Status 和 Pipeline.Stages
3. 当某个阶段的状态为 Init，且此前所有阶段均为 Success 时 → 调用 exec-pipeline-run-stage --id <PipelineRunId> --code <Stage.Code> 推进该阶段（参数为 --id，不是 --pipeline-run-id）
4. 直至流水线整体状态变为 Success（部署成功）— 其他任何终态（Fail、Termination、Cancel）均表示部署未成功；请参见上方规则 #15

要点：构建阶段会自动运行，但检查和部署阶段必须手动推进。详细的 CLI 示例、完整的状态决策矩阵和轮询脚本位于 [references/deploy-guide.md](references/deploy-guide.md)。

CLI 说明：aliyun CLI 返回的 JSON 以 Pipeline 为顶层键（而不是 SDK 的 resp.body.pipeline）；各阶段位于 Pipeline.Stages 中。

常用节点类型

| 使用场景 | 命令 | contentFormat | 扩展名 | 数据源 | |------|---------|--------------|------|------------| | Shell 脚本 | DIDE_SHELL | shell | .sh | — | | MaxCompute SQL | ODPS_SQL | sql | .sql | odps | | Python 脚本 | PYTHON | python | .py | — | | 离线数据同步 | DI | json | .json | — | | Hologres SQL | HOLOGRES_SQL | sql | .sql | hologres | | Flink 流式 SQL | FLINK_SQL_STREAM | sql | .json | flink | | Flink 批处理 SQL | FLINK_SQL_BATCH | sql | .json | flink | | EMR Hive | EMR_HIVE | sql | .sql | emr | | EMR Spark SQL | EMR_SPARK_SQL | sql | .sql | emr | | Serverless Spark SQL | SERVERLESS_SPARK_SQL | sql | .sql | emr | | StarRocks SQL | StarRocks | sql | .sql | starrocks | | ClickHouse SQL | CLICK_SQL | sql | .sql | clickhouse | | 虚拟节点 | VIRTUAL | empty | .vi | — |

完整列表（130+ 种类型）：[references/nodetypes/index.md](references/nodetypes/index.md)（可按命令名称、描述和类别搜索，并包含每种类型的详细文档链接）

找不到节点类型时：

1. 检查 references/nodetypes/index.md 并按关键字匹配
2. 使用 Glob("**/{keyword}*.md", path="references/nodetypes") 直接定位文档
3. 使用 get-node 命令从实际环境中获取类似节点的配置作为参考
4. 如果以上方法均无效 → 回退到 DIDE_SHELL，并在 Shell 中使用命令行工具完成任务

核心约束

1. script.path 为必填项：脚本路径必须以节点名称结尾。创建时可以仅传入节点名称；服务器会自动添加工作流前缀
2. 通过 spec.dependencies 配置依赖关系：在 spec.dependencies 中，nodeId 是自引用——必须是当前节点自身的 name（正在创建的节点），不得是上游节点。depends[].output 是上游节点的输出（${projectIdentifier}.UpstreamNodeName）。上游节点的 outputs.nodeOutputs[].data 与下游节点的 depends[].output 必须逐字符完全一致。上游节点必须声明 outputs.nodeOutputs。⚠️ 输出名称（${projectIdentifier}.NodeName）必须在项目内全局唯一——重复会导致部署失败
3. 不可变属性：节点的 command（节点类型）在创建后无法更改；如果该类型不正确，请告知用户并建议使用正确类型创建新节点
4. 更新必须采用增量方式：仅传入 id + 需要修改的字段；不要传入 datasource/runtimeResource 等未更改的字段
5. datasource.type 可能由服务器更正：例如，flink → flink_serverless；创建时使用通用类型
6. 节点可以独立存在：节点可以在根层级创建（不传入 --container-id），也可以属于某个工作流（传入 --container-id WorkflowId）。是否将节点放入工作流由用户决定
7. 工作流命令始终为 WORKFLOW：script.runtime.command 必须为 "WORKFLOW"
8. 此 skill 不支持删除操作：此 skill 不提供任何删除操作。创建或发布失败时，禁止尝试通过删除现有对象来“修复”问题。正确方法：诊断失败原因 → 告知用户具体冲突 → 让用户决定如何处理（重命名 / 更新现有对象）
9. 创建前必须检查名称冲突：调用任何创建类 API 前，使用对应的列表类 API 确认名称没有重复（参见“环境发现”）。名称冲突会导致创建失败；节点输出名称（outputs.nodeOutputs[].data）重复会导致依赖错误或发布失败
10. 变更操作需要用户确认：除创建和只读查询（获取/列出）外，所有修改现有对象的 OpenAPI 操作（更新、移动、重命名等）必须在执行前向用户展示，并获得用户明确确认。确认信息应包括：操作类型、目标对象名称/ID 和关键变更。在用户确认之前，不得调用这些 APIs。此 skill 不支持删除和下线操作
11. 仅使用 2024-05-18 版本的 APIs：此 skill 中的所有 APIs 均为 DataWorks 2024-05-18 版本。禁止使用旧版 APIs（create-file、create-folder、create-flow-project 等）。如果 API 调用返回错误，请先查阅 [troubleshooting.md](references/troubleshooting.md)；不得回退到旧版 APIs
12. 发生错误时停止，不要暴力重试：如果同一错误码连续出现超过 2 次，说明当前方法有误。停止操作并分析错误原因（查阅 [troubleshooting.md](references/troubleshooting.md)），不要使用不同参数反复重试同一个错误的 API。新 API 失败时，禁止回退到旧版 APIs（create-file、create-business 等），而应查看本文顶部的 FlowSpec 反模式表。#1 失败陷阱：如果遇到 0x5083000000000005（“Spec JSON 解析失败”），不得尝试随意构造的 FlowSpec 结构，而应转到快速入门部分，逐字复制其中可用的 JSON。#2 失败陷阱：如果找不到命令，请确保已安装插件（aliyun plugin install --names dataworks-public）
13. 必须在文档中核对 CLI 参数名称，禁止猜测：调用 API 前，必须先查阅 references/api/{APIName}.md 以确认参数名称。常见错误：get-project 的 ID 参数是 --id（而不是 --project-id）；update-node 需要 --id。不确定时，使用 aliyun dataworks-public {command} --help 验证
14. 写操作的幂等性保护：创建前，执行创建前冲突检查（列表类 API）。创建操作发生网络错误/超时后，先通过列表/获取 API 检查资源是否已创建，然后再重试。记录每个响应中的 RequestId
15. 除非 Pipeline.Status == 'Success'，否则禁止声称部署成功：调用 create-pipeline-run 后，只有 get-pipeline-run 返回的最终 Pipeline.Status 为 Success，才表示“已部署”。状态为 Termination、Fail、Cancel，或任何阶段仍处于 Init/Fail，均表示部署未成功，即使部分阶段成功、版本号已更改，或 exec-pipeline-run-stage 针对较早阶段返回了 Success: true。在这些情况下，不得写入声称成功的本地结果文件（例如 publishing_result.json），不得告诉用户“已部署”，也不得静默忽略 400 流水线不是正在运行 错误。在编写面向用户的摘要前，再获取一次 get-pipeline-run；报告实际终止流水线的阶段。决策矩阵：[references/deploy-guide.md](references/deploy-guide.md) 的“状态决策矩阵”章节

API 快速参考

API 版本：下列所有 APIs 均为 DataWorks 2024-05-18 版本。仅使用下表列出的 APIs；不得搜索或使用其他 DataWorks APIs。

业务调用格式（插件模式，每次调用都必须带 UA 标志）：aliyun dataworks-public {kebab-case-command} --parameter --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop——有关插件安装和全局 UA 配置，请参阅上文的“阿里云 CLI 设置”部分。

每个 API 的详细参数和代码模板均位于 references/api/{APIName}.md 中。如果调用返回错误，可从 https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/{APIName}/api.json 获取最新定义。

组件

| 命令 | 说明 | |-----|------| | [create-component](references/api/CreateComponent.md) | 创建组件 | | [get-component](references/api/GetComponent.md) | 获取组件详情 | | [update-component](references/api/UpdateComponent.md) | 更新组件 | | [list-components](references/api/ListComponents.md) | 列出组件 |

节点

| 命令 | 说明 | |-----|------| | [create-node](references/api/CreateNode.md) | 创建数据开发节点。project-id + scene + spec，container-id 可选 | | [update-node](references/api/UpdateNode.md) | 更新节点信息。增量更新，仅传入 id + 待修改字段 | | [move-node](references/api/MoveNode.md) | 将节点移动到指定路径 | | [rename-node](references/api/RenameNode.md) | 重命名节点 | | [get-node](references/api/GetNode.md) | 获取节点详情，返回完整 spec | | [list-nodes](references/api/ListNodes.md) | 列出节点，支持按工作流筛选 | | [list-node-dependencies](references/api/ListNodeDependencies.md) | 列出节点的依赖节点 |

工作流定义

| 命令 | 说明 | |-----|------| | [create-workflow-definition](references/api/CreateWorkflowDefinition.md) | 创建工作流。project-id + spec | | [import-workflow-definition](references/api/ImportWorkflowDefinition.md) | 导入工作流（仅限首次批量导入——不得用于更新或发布；应改用 update-node + create-pipeline-run） | | [update-workflow-definition](references/api/UpdateWorkflowDefinition.md) | 更新工作流信息，增量更新 | | [move-workflow-definition](references/api/MoveWorkflowDefinition.md) | 将工作流移动到目标路径 | | [rename-workflow-definition](references/api/RenameWorkflowDefinition.md) | 重命名工作流 | | [get-workflow-definition](references/api/GetWorkflowDefinition.md) | 获取工作流详情 | | [list-workflow-definitions](references/api/ListWorkflowDefinitions.md) | 列出工作流，可按类型筛选 |

资源

| 命令 | 说明 | |-----|------| | [create-resource](references/api/CreateResource.md) | 创建文件资源 | | [update-resource](references/api/UpdateResource.md) | 更新文件资源信息，增量更新 | | [move-resource](references/api/MoveResource.md) | 将文件资源移动到指定目录 | | [rename-resource](references/api/RenameResource.md) | 重命名文件资源 | | [get-resource](references/api/GetResource.md) | 获取文件资源详情 | | [list-resources](references/api/ListResources.md) | 列出文件资源 |

函数

| 命令 | 说明 | |-----|------| | [create-function](references/api/CreateFunction.md) | 创建 UDF 函数 | | [update-function](references/api/UpdateFunction.md) | 更新 UDF 函数信息，增量更新 | | [move-function](references/api/MoveFunction.md) | 将函数移动到目标路径 | | [rename-function](references/api/RenameFunction.md) | 重命名函数 | | [get-function](references/api/GetFunction.md) | 获取函数详情 | | [list-functions](references/api/ListFunctions.md) | 列出函数 |

发布流水线

| 命令 | 说明 | |-----|------| | [create-pipeline-run](references/api/CreatePipelineRun.md) | 创建发布流水线。type=Online/Offline | | [exec-pipeline-run-stage](references/api/ExecPipelineRunStage.md) | 执行发布流水线的指定阶段，异步操作需要轮询 | | [get-pipeline-run](references/api/GetPipelineRun.md) | 获取发布流水线详情，返回 Stages 状态 | | [list-pipeline-runs](references/api/ListPipelineRuns.md) | 列出发布流水线 | | [list-pipeline-run-items](references/api/ListPipelineRunItems.md) | 获取发布内容 |

辅助查询

| 命令 | 说明 | |-----|------| | [get-project](references/api/GetProject.md) | 根据 id 获取 projectIdentifier | | [list-data-sources](references/api/ListDataSources.md) | 列出数据源 | | [list-compute-resources](references/api/ListComputeResources.md) | 列出计算引擎绑定（EMR、Hologres、StarRocks 等）——作为 list-data-sources 的补充 | | [list-resource-groups](references/api/ListResourceGroups.md) | 列出资源组 |

参考文档

| 场景 | 文档 | |------|------| | APIs 和 CLI 命令的完整列表 | [references/related-apis.md](references/related-apis.md) | | RAM 权限策略配置 | [references/ram-policies.md](references/ram-policies.md) | | 操作验证方法 | [references/verification-method.md](references/verification-method.md) | | 验收标准和测试用例 | [references/acceptance-criteria.md](references/acceptance-criteria.md) | | CLI 安装和配置指南 | [references/cli-installation-guide.md](references/cli-installation-guide.md) | | 节点类型索引（130+ 种类型） | [references/nodetypes/index.md](references/nodetypes/index.md) | | FlowSpec 字段参考 | [references/flowspec-guide.md](references/flowspec-guide.md) | | 工作流开发 | [references/workflow-guide.md](references/workflow-guide.md) | | 调度配置 | [references/scheduling-guide.md](references/scheduling-guide.md) | | 发布与取消发布 | [references/deploy-guide.md](references/deploy-guide.md) | | DI 数据集成 | [references/di-guide.md](references/di-guide.md) | | 故障排查 | [references/troubleshooting.md](references/troubleshooting.md) | | 完整示例 | [assets/templates/README.md](assets/templates/README.md) |