alibabacloud-dataworks-datastudio-develop

DataWorks 数据开发

❗ 5 秒摘要 — 请先阅读

凭证：先运行 aliyun configure list。CLI 几乎总是已预配置（STS 令牌）。不要搜索凭证文件，只需检查 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 设置（首次使用前执行）：

# 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。

0. 第一步：检查 CLI 凭证。运行任何 aliyun 命令前，先运行 aliyun configure list。CLI 通常已预配置有效的 STS 令牌凭证——检查前不要搜索凭证文件（例如 testconfig.json）。如果 aliyun configure list 显示存在有效配置，则直接使用。如果存在多个配置，运行 aliyun configure switch --profile <name> 选择正确的配置。优先级：优先选择名称包含 dataworks（不区分大小写）的配置；否则使用 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 返回错误，请修正规范——不要回退到旧版 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. 切勿模拟、伪造或编造 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 字段；名称应放在 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 个依赖节点的定时工作流：

# 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
2. 处理已存在的对象：告知用户并询问如何处理（使用现有对象 / 重命名 / 更新现有对象）。禁止直接删除现有对象
3. 输出名称冲突检查（至关重要）：节点的 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 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 输入：

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

   build.py 会执行三项操作（不依赖第三方库；如果出错，请参照源代码手动执行）：

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

2. 提交前验证 spec：

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

3. 提交前验证（必须）——调用 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 中未记录的字段

4. 调用 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'])。

5. 要将节点放入工作流中，请添加 --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）必须在项目内全局唯一——如果其他节点（即使位于不同工作流中）已使用相同的输出名称，部署将失败，并报错 "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。

1. 创建工作流定义（最简 spec）：

   {"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）
3. 验证依赖项（所有节点创建完成后必须执行） — 对每个下游节点，调用 list-node-dependencies --id <NodeID>。如果 TotalCount 为 0，但该节点应有上游依赖项，则 create-node 已静默丢弃这些依赖项。请立即使用 update-node 和 spec.dependencies 修复（参见下文“更新依赖项”）。确认所有依赖项之前，不得继续部署
4. 设置调度 — 如果用户指定了调度，则使用 trigger 调用 update-workflow-definition
5. 在线部署（必需） — 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

更新现有节点

必须使用增量更新——只传入要修改的节点 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。示例：

{"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. 更新节点——使用增量 spec 调用 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> 选择规则：如果节点位于工作流内（其 path 来自 get-node 且包含 /，例如 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——旧版 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 字段获取 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. 当某个 Stage 的状态为 Init 且所有先前的 Stage 均为 Success 时 → 调用 exec-pipeline-run-stage --id <PipelineRunId> --code <Stage.Code> 推进阶段（参数是 --id，不是 --pipeline-run-id）
4. 直到 Pipeline 的总体状态变为 Success（部署成功）——任何其他终态（Fail、Termination、Cancel）都表示部署未成功；请参阅上文规则 #15

要点：Build 阶段会自动运行，但 Check 和 Deploy 阶段必须手动推进。详细的 CLI 示例、完整的状态决策矩阵和轮询脚本，请参阅 references/deploy-guide.md。

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

常见节点类型

用途	command	contentFormat	Extension	datasource
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（可按 command 名称、描述和类别搜索，并提供各类型详细文档的链接）

找不到节点类型时：

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. 此技能不支持删除：此技能不提供任何删除操作。创建或发布失败时，绝不要尝试通过删除现有对象来“修复”问题。正确做法：诊断失败原因 → 告知用户具体冲突 → 由用户决定如何处理（重命名 / 更新）
9. 创建前必须检查名称冲突：调用任何 Create API 前，使用对应的 List API 确认名称没有重复（参见“环境发现”）。名称冲突会导致创建失败；重复的节点输出名称（outputs.nodeOutputs[].data）会导致依赖项错误或发布失败
10. 修改操作必须获得用户确认：除 Create 和只读查询（Get/List）外，所有会修改现有对象的 OpenAPI 操作（Update、Move、Rename 等）都必须先向用户展示操作内容并明确获得确认，然后才能执行。确认信息应包含：操作类型、目标对象名称 / ID 和关键更改。获得用户确认前不得调用这些 API。此技能不支持 Delete 和 Abolish 操作
11. 仅使用 2024-05-18 版本的 API：此技能中的所有 API 均为 DataWorks 2024-05-18 版本。禁止使用旧版 API（create-file、create-folder、create-flow-project 等）。如果 API 调用返回错误，先检查 troubleshooting.md；不要回退到旧版 API
12. 遇到错误时停止，不要盲目重试：如果相同错误代码连续出现超过 2 次，说明方法有误。请停止并分析错误原因（检查 troubleshooting.md），不要不断使用不同参数重试同一个错误的 API。新 API 调用失败时，绝不要回退到旧版 API（create-file、create-business 等），请改为查看本文顶部的 FlowSpec 反模式表。失败陷阱 #1：如果遇到 0x5083000000000005（“Spec JSON parse failed”），不要尝试随机的 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. 写入操作的幂等性保护：创建前，执行创建前的冲突检查（List API）。Create 操作发生网络错误 / 超时后，重试前先通过 List/Get 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 的“状态决策矩阵”部分

API 快速参考

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

业务调用格式（插件模式，每次调用都必须带上 UA 标记）：aliyun dataworks-public {kebab-case-command} --parameter --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop — 插件安装和全局 UA 配置请参阅上方的“Aliyun 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	创建组件
get-component	获取组件详情
update-component	更新组件
list-components	列出组件

节点

命令	描述
create-node	创建数据开发节点。project-id + scene + spec，container-id 可选
update-node	更新节点信息。增量更新，仅传入 id 和要更改的字段
move-node	将节点移至指定路径
rename-node	重命名节点
get-node	获取节点详情，返回完整 spec
list-nodes	列出节点，支持按工作流筛选
list-node-dependencies	列出节点的依赖节点

工作流定义

命令	描述
create-workflow-definition	创建工作流。project-id + spec
import-workflow-definition	导入工作流（仅限初始批量导入——请勿用于更新或发布；请改用 update-node + create-pipeline-run）
update-workflow-definition	更新工作流信息，增量更新
move-workflow-definition	将工作流移至目标路径
rename-workflow-definition	重命名工作流
get-workflow-definition	获取工作流详情
list-workflow-definitions	列出工作流，按类型筛选

资源

命令	描述
create-resource	创建文件资源
update-resource	更新文件资源信息，增量更新
move-resource	将文件资源移至指定目录
rename-resource	重命名文件资源
get-resource	获取文件资源详情
list-resources	列出文件资源

函数

命令	描述
create-function	创建 UDF 函数
update-function	更新 UDF 函数信息，增量更新
move-function	将函数移至目标路径
rename-function	重命名函数
get-function	获取函数详情
list-functions	列出函数

发布流水线

命令	描述
create-pipeline-run	创建发布流水线。type=Online/Offline
exec-pipeline-run-stage	执行发布流水线的指定阶段；异步执行需要轮询
get-pipeline-run	获取发布流水线详情，返回 Stages 状态
list-pipeline-runs	列出发布流水线
list-pipeline-run-items	获取发布内容

辅助查询

命令	描述
get-project	根据 id 获取 projectIdentifier
list-data-sources	列出数据源
list-compute-resources	列出计算引擎绑定（EMR、Hologres、StarRocks 等）——补充 list-data-sources
list-resource-groups	列出资源组

参考文档

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