Alibabacloud Dataworks Datastudio Develop

Example 01: Shell Node

The simplest DataWorks node example. Creates a Shell script node that runs daily at midnight.

File Structure

hello/
├── hello.spec.json        # Node definition
├── hello.sh               # Shell script
└── dataworks.properties   # Configuration

Creation Steps

1. Create a spec file based on hello/hello.spec.json 2. Modify name, path, and content 3. Write hello.sh 4. Create dataworks.properties 5. Git Mode: git add && git commit 6. OpenAPI Mode: Call CreateNode API directly with minSpec

Example 02: MaxCompute SQL Node

SQL node with datasource. Demonstrates datasource configuration.

File Structure

dwd_user_info/
├── dwd_user_info.spec.json
├── dwd_user_info.sql
└── dataworks.properties

Example 03: SQL Inter-Node Dependency

Two ODPS SQL nodes where dwd_order_detail depends on ods_order.

File Structure

ods_order/
├── ods_order.spec.json
├── ods_order.sql
└── dataworks.properties
dwd_order_detail/
├── dwd_order_detail.spec.json
├── dwd_order_detail.sql
└── dataworks.properties

Example 04: DI Data Synchronization (MySQL to MaxCompute)

Offline data synchronization node that reads data from MySQL and writes to MaxCompute.

File Structure

sync_user/
├── sync_user.spec.json
├── sync_user.json          # DI task code (JSON format)
└── dataworks.properties

Deployment Command Sequence

PROJECT_ID=123456

# 1. Create workflow
# Note: Workflow spec must include script.runtime.command="WORKFLOW"
# Build spec JSON and call API
aliyun dataworks-public CreateWorkflowDefinition \
  --ProjectId $PROJECT_ID \
  --Spec "$(cat /tmp/wf.json)"
# -> Record the returned WorkflowId
WF_ID="<returned WorkflowId>"

# 2. Create extract node (root node, no upstream dependency)
# Build spec JSON and call API
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n1.json)"

# 3. Create transform node (depends on extract)
# Note: Dependencies are configured via spec.dependencies; ensure dependencies[*].nodeId exactly matches the node id
# Build spec JSON and call API
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n2.json)"

# 4. Create load node (depends on transform)
# Build spec JSON and call API
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n3.json)"

# 5. Publish and go live
aliyun dataworks-public CreatePipelineRun \
  --ProjectId $PROJECT_ID \
  --Type Online \
  --ObjectIds "[\"$WF_ID\"]"

Example 05: Scheduled Workflow

ETL workflow with 3 child nodes: extract (Shell) -> transform (SQL) -> load (SQL).

File Structure

my_etl/
├── my_etl.spec.json           # Workflow definition
├── dataworks.properties
├── extract/
│   ├── extract.spec.json
│   ├── extract.sh
│   └── dataworks.properties
├── transform/
│   ├── transform.spec.json
│   ├── transform.sql
│   └── dataworks.properties
└── load/
    ├── load.spec.json
    ├── load.sql
    └── dataworks.properties

Deployment Order

1. Create workflow -> obtain WorkflowId 2. Create extract node (ContainerId=WorkflowId) 3. Create transform node (ContainerId=WorkflowId) 4. Create load node (ContainerId=WorkflowId) 5. Publish and go live

Example 06: Manual Workflow

Manually triggered workflow with two steps.

File Structure

manual_task/
├── manual_task.spec.json
├── dataworks.properties
├── step1/
│   ├── step1.spec.json
│   ├── step1.sh
│   └── dataworks.properties
└── step2/
    ├── step2.spec.json
    ├── step2.py
    └── dataworks.properties

Deployment Command Sequence

PROJECT_ID=123456

# 1. Create workflow
# Note: Workflow spec must include script.runtime.command="WORKFLOW"
aliyun dataworks-public CreateWorkflowDefinition \
  --ProjectId $PROJECT_ID \
  --Spec "$(cat /tmp/wf.json)"
# -> Record the returned WorkflowId
WF_ID="<returned WorkflowId>"

# 2. Create prepare_data node (root node, depends on project_root)
# Note: Dependencies are configured via spec.dependencies; ensure dependencies[*].nodeId exactly matches the node id
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n1.json)"

# 3. Create process_orders node (fan-out, depends on prepare_data)
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n2.json)"

# 4. Create process_users node (fan-out, depends on prepare_data)
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n3.json)"

# 5. Create merge_report node (fan-in, depends on process_orders + process_users)
# Note: Multiple upstream dependencies are listed as multiple entries in the spec.dependencies depends array
aliyun dataworks-public CreateNode \
  --ProjectId $PROJECT_ID \
  --Scene DATAWORKS_PROJECT \
  --ContainerId $WF_ID \
  --Spec "$(cat /tmp/n4.json)"

# 6. Publish and go live
aliyun dataworks-public CreatePipelineRun \
  --ProjectId $PROJECT_ID \
  --Type Online \
  --ObjectIds "[\"$WF_ID\"]"

Example 07: Parallel Workflow (Fan-out + Fan-in)

Parallel ETL workflow with 4 child nodes, demonstrating fan-out and fan-in dependency patterns:

prepare_data (Shell)
    ├──→ process_orders (SQL)  ──┐
    └──→ process_users  (SQL)  ──┴──→ merge_report (SQL)

Dependency scenarios covered:

• Root node dependency (prepare_data <- project_root)
• Fan-out: multiple nodes depend on the same upstream (process_orders, process_users <- prepare_data)
• Fan-in: one node depends on multiple upstreams (merge_report <- process_orders + process_users)

File Structure

parallel_etl/
├── parallel_etl.spec.json         # Workflow definition
├── dataworks.properties
├── prepare_data/
│   ├── prepare_data.spec.json
│   ├── prepare_data.sh
│   └── dataworks.properties
├── process_orders/
│   ├── process_orders.spec.json
│   ├── process_orders.sql
│   └── dataworks.properties
├── process_users/
│   ├── process_users.spec.json
│   ├── process_users.sql
│   └── dataworks.properties
└── merge_report/
    ├── merge_report.spec.json
    ├── merge_report.sql
    └── dataworks.properties

Dependency Configuration Key Points

When creating nodes inside a workflow using CreateNode + ContainerId, dependencies are configured exclusively via spec.dependencies. Note: spec.dependencies[*].nodeId MUST exactly match the corresponding node's id, otherwise the dependency information will not be recognized.

Multi-upstream dependency (fan-in) example (merge_report depends on both process_orders and process_users):

"dependencies": [
  {
    "nodeId": "merge_report",
    "depends": [
      {"type": "Normal", "output": "${projectIdentifier}.process_orders"},
      {"type": "Normal", "output": "${projectIdentifier}.process_users"}
    ]
  }
]

Deployment Order

1. Create workflow -> obtain WorkflowId 2. Create prepare_data node (ContainerId=WorkflowId) 3. Create process_orders node (ContainerId=WorkflowId) 4. Create process_users node (ContainerId=WorkflowId) 5. Create merge_report node (ContainerId=WorkflowId) 6. Publish and go live

API Verification Status

This template has been verified through CreateNode + ContainerId API testing (cn-beijing, 2026-03-28). All dependency relationships were saved correctly.

Example 08: Assignment Node

Assignment node (CONTROLLER_ASSIGNMENT) executes a script and passes the result to downstream nodes via ${outputs}. Three language variants are provided, one template per language.

File Structure

08-assignment-node/
├── assignment_shell/                  # Shell 赋值节点
│   ├── assignment_shell.spec.json
│   ├── assignment_shell.sh
│   └── dataworks.properties
├── assignment_odps/                   # MaxCompute SQL 赋值节点
│   ├── assignment_odps.spec.json
│   ├── assignment_odps.sql
│   └── dataworks.properties
├── assignment_python/                 # Python 2 赋值节点
│   ├── assignment_python.spec.json
│   ├── assignment_python.py
│   └── dataworks.properties
└── README.md

Language Comparison

Key Points

• Code file使用语言原生文件（.sh/.py/.sql），build.py 自动嵌入 script.content
• datasource (ODPS type) is required for all three languages
• runtime.command is always CONTROLLER_ASSIGNMENT
• Output is automatically assigned to ${outputs} variable for downstream consumption

Creation Steps

1. Copy the desired language template directory 2. Rename node name/path in spec.json 3. Edit the code file (.sh/.py/.sql) with your actual code 4. Edit dataworks.properties with project-specific values 5. Build: python scripts/build.py ./assignment_shell

Downstream Parameter Passing

See CONTROLLER_ASSIGNMENT.md for full parameter passing configuration.

DataWorks Data Development Examples

Each example contains spec.json, code files, and dataworks.properties, demonstrating the local file structure in Git Mode.

Acceptance Criteria: DataWorks Data Development

Scenario: DataWorks node and workflow development Purpose: Skill test acceptance criteria

---

Correct CLI Command Patterns

1. Product -- Verify Product Name Exists

# CORRECT: dataworks-public is the correct product name
aliyun dataworks-public get-node --help

# INCORRECT: dataworks without the -public suffix
aliyun dataworks get-node --help

2. Command -- Verify Action Exists Under the Product

# CORRECT: Correct action names
aliyun dataworks-public create-node --help
aliyun dataworks-public list-nodes --help
aliyun dataworks-public get-node --help
aliyun dataworks-public update-node --help

aliyun dataworks-public create-workflow-definition --help
aliyun dataworks-public list-workflow-definitions --help
aliyun dataworks-public get-workflow-definition --help

aliyun dataworks-public create-pipeline-run --help
aliyun dataworks-public get-pipeline-run --help
aliyun dataworks-public exec-pipeline-run-stage --help

# INCORRECT: Wrong action names
aliyun dataworks-public create-task --help  # Should be create-node
aliyun dataworks-public list-task --help    # Should be list-nodes

3. Parameters -- Verify Each Parameter Name Exists

# CORRECT: Correct parameter names
aliyun dataworks-public create-node \
  --project-id 123456 \
  --scene DATAWORKS_PROJECT \
  --spec '{"version":"2.0.0",...}'

aliyun dataworks-public create-node \
  --project-id 123456 \
  --scene DATAWORKS_PROJECT \
  --container-id 789012 \
  --spec '{"version":"2.0.0",...}'

# INCORRECT: Wrong parameter names
aliyun dataworks-public create-node \
  --projectid 123456        # Should be --project-id (kebab-case)
  --Scene DATAWORKS_PROJECT  # Should be --scene (lowercase)

4. user-agent Identifier -- Must Be Included in Every Command

# CORRECT: Includes user-agent
aliyun dataworks-public get-node \
  --project-id 123456 \
  --id 789012 \
  --user-agent AlibabaCloud-Agent-Skills

# INCORRECT: Missing user-agent
aliyun dataworks-public get-node \
  --project-id 123456 \
  --id 789012

---

Correct FlowSpec Patterns

1. Node spec.json Basic Structure

// CORRECT: Correct node spec structure
{
  "version": "2.0.0",
  "kind": "Node",
  "spec": {
    "nodes": [{
      "name": "my_node",
      "script": {
        "path": "my_node",
        "language": "odps-sql",
        "runtime": {
          "command": "ODPS_SQL"
        }
      },
      "trigger": {
        "type": "Scheduler",
        "cron": "00 00 00 * * ?",
        "startTime": "1970-01-01 00:00:00",
        "endTime": "9999-01-01 00:00:00",
        "timezone": "Asia/Shanghai"
      }
    }],
    "dependencies": [{
      "nodeId": "my_node",
      "depends": [{
        "type": "Normal",
        "output": "${projectIdentifier}_root"
      }]
    }]
  }
}

// INCORRECT: Missing required fields
{
  "kind": "Node",  // Missing version
  "spec": {
    "nodes": [{
      "name": "my_node"
      // Missing script
    }]
  }
}

2. script.path Must Match name

// CORRECT: path matches name
{
  "name": "etl_daily",
  "script": {
    "path": "etl_daily",  // Matches name
    "runtime": { "command": "ODPS_SQL" }
  }
}

// INCORRECT: path does not match name
{
  "name": "etl_daily",
  "script": {
    "path": "other_path",  // API will return "script path not match name"
    "runtime": { "command": "ODPS_SQL" }
  }
}

3. Dependency Configuration (spec.dependencies)

// CORRECT: Configure dependencies in spec.dependencies, ensure nodeId exactly matches the node id
{
  "spec": {
    "nodes": [{
      "name": "downstream"
    }],
    "dependencies": [{
      "nodeId": "downstream",
      "depends": [{
        "type": "Normal",
        "output": "${projectIdentifier}.upstream"
      }]
    }]
  }
}

// INCORRECT: Using the legacy `flow` field instead of `spec.dependencies`
{
  "spec": {
    "nodes": [{ "name": "downstream" }],
    "flow": [{
      "nodeId": "downstream",
      "depends": [{
        "type": "Normal",
        "output": "${projectIdentifier}.upstream"
      }]
    }]
  }
}

4. Workflow spec Must Include command: WORKFLOW

// CORRECT: Workflow spec
{
  "version": "2.0.0",
  "kind": "CycleWorkflow",
  "spec": {
    "workflows": [{
      "name": "my_workflow",
      "script": {
        "path": "my_workflow",
        "runtime": {
          "command": "WORKFLOW"  // Must be set
        }
      },
      "trigger": {
        "type": "Scheduler",
        "cron": "00 00 00 * * ?"
      }
    }]
  }
}

// INCORRECT: Missing command
{
  "version": "2.0.0",
  "kind": "CycleWorkflow",
  "spec": {
    "workflows": [{
      "name": "my_workflow",
      "script": {
        "path": "my_workflow"
        // Missing runtime.command, API will return an error
      }
    }]
  }
}

5. Datasource Type Matching

// CORRECT: ODPS_SQL uses odps datasource
{
  "script": {
    "runtime": { "command": "ODPS_SQL" },
    "language": "odps-sql"
  },
  "datasource": {
    "name": "${spec.datasource.name}",
    "type": "odps"
  }
}

// CORRECT: HOLOGRES_SQL uses hologres datasource
{
  "script": {
    "runtime": { "command": "HOLOGRES_SQL" },
    "language": "hologres-sql"
  },
  "datasource": {
    "name": "${spec.datasource.name}",
    "type": "hologres"
  }
}

// INCORRECT: Type mismatch
{
  "script": {
    "runtime": { "command": "HOLOGRES_SQL" }
  },
  "datasource": {
    "name": "my_ds",
    "type": "odps"  // Should be hologres
  }
}

---

Correct dataworks.properties Patterns

# CORRECT: Correct properties format
projectIdentifier=my_project_name
spec.datasource.name=my_odps_datasource
spec.runtimeResource.resourceGroup=S_res_group_xxx
script.bizdate=20260101

# INCORRECT: Wrong key prefix
datasource.name=my_ds              # Should be spec.datasource.name
resource_group=S_res_group_xxx     # Should be spec.runtimeResource.resourceGroup

# INCORRECT: Value contains placeholder
spec.datasource.name=${datasource} # Value must not contain placeholders

---

Correct Python SDK Code Patterns

1. Import Patterns

# CORRECT
from alibabacloud_dataworks_public20240518.client import Client
from alibabacloud_dataworks_public20240518.models import CreateNodeRequest
from alibabacloud_tea_openapi.models import Config

# INCORRECT
from alibabacloud_dataworks.client import Client  # Wrong module name
from dataworks.models import CreateNodeRequest    # Wrong module name

2. Client Initialization

# CORRECT: Use CredentialClient
from alibabacloud_credentials.client import Client as CredentialClient

credential = CredentialClient()
config = Config(credential=credential)
config.endpoint = 'dataworks.cn-hangzhou.aliyuncs.com'
client = Client(config)

# INCORRECT: Hardcoded AK/SK (security risk)
config = Config(
    access_key_id='LTAI5tXXX',        # Do not hardcode
    access_key_secret='8dXXXXXXX'     # Do not hardcode
)

3. API Calls

# CORRECT
request = CreateNodeRequest(
    project_id=123456,
    scene='DATAWORKS_PROJECT',
    spec=spec_json
)
response = client.create_node(request)
node_id = response.body.id

# INCORRECT: Wrong parameter names
request = CreateNodeRequest(
    projectId=123456,    # Should be project_id
    Scene='xxx'          # Should be scene
)

---

Validation Commands

Each CLI command should be verified with --help:

# Verify product and action exist
aliyun dataworks-public create-node --help
aliyun dataworks-public update-node --help
aliyun dataworks-public get-node --help
aliyun dataworks-public list-nodes --help

aliyun dataworks-public create-workflow-definition --help
aliyun dataworks-public update-workflow-definition --help
aliyun dataworks-public get-workflow-definition --help
aliyun dataworks-public list-workflow-definitions --help

aliyun dataworks-public create-pipeline-run --help
aliyun dataworks-public get-pipeline-run --help
aliyun dataworks-public exec-pipeline-run-stage --help
aliyun dataworks-public list-pipeline-runs --help
aliyun dataworks-public list-pipeline-run-items --help
aliyun dataworks-public abolish-pipeline-run --help

aliyun dataworks-public get-project --help
aliyun dataworks-public list-data-sources --help
aliyun dataworks-public list-resource-groups --help

---

Critical Anti-Patterns to Avoid

1. Dependencies must only live in `spec.dependencies`: Do not place dependency declarations anywhere else on the node; dependencies[*].nodeId must exactly match the node id 2. Do not hardcode AK/SK: Use CredentialClient or environment variables 3. Do not forget user-agent: Every aliyun command must include --user-agent AlibabaCloud-Agent-Skills 4. Do not assume update-node works for all nodes: Hologres nodes cannot be updated 5. Do not skip validation: Always run validate.py after each modification 6. Do not echo AK/SK: Never print credential information 7. Do not execute write operations without confirmation: Except for Create and read-only queries (Get/List), all Delete, Update, Move, Rename, Abolish, and other APIs that modify existing objects must be confirmed with the user first

DataWorks Data Development API Call Templates

All APIs are based on the DataWorks OpenAPI 2024-05-18 version. Each operation provides both aliyun CLI and Python SDK methods.

Node Operations

Create Node

aliyun CLI:

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

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

Python SDK:

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_dataworks_public20240518.client import Client
from alibabacloud_dataworks_public20240518.models import CreateNodeRequest
from alibabacloud_tea_openapi.models import Config

credential = CredentialClient()
config = Config(credential=credential)
config.endpoint = 'dataworks.{{region}}.aliyuncs.com'
config.user_agent = 'AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop'
client = Client(config)

with open('/tmp/spec.json') as f:
    spec = f.read()

request = CreateNodeRequest(
    project_id={{project_id}},
    scene='DATAWORKS_PROJECT',
    spec=spec
)
response = client.create_node(request)
print(f"NodeId: {response.body.id}")

Create Node Within a Workflow

aliyun CLI:

$PYTHON $SKILL/scripts/build.py ./my_wf/step1 > /tmp/spec.json

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

Python SDK:

request = CreateNodeRequest(
    project_id=383839,
    scene='DATAWORKS_PROJECT',
    container_id='<workflow_id>',  # Create the node inside a workflow
    spec=spec
)
response = client.create_node(request)
print(f"NodeId: {response.body.id}")

Update Node

aliyun CLI:

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

aliyun dataworks-public update-node \
  --project-id {{project_id}} \
  --id {{node_id}} \
  --spec "$(cat /tmp/spec.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import UpdateNodeRequest

request = UpdateNodeRequest(
    project_id={{project_id}},
    id='{{node_id}}',
    spec=spec
)
response = client.update_node(request)

Get Node Details

aliyun CLI:

aliyun dataworks-public get-node \
  --project-id {{project_id}} \
  --id {{node_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetNodeRequest

request = GetNodeRequest(
    project_id={{project_id}},
    id='{{node_id}}'
)
response = client.get_node(request)
# response.body.spec contains the full FlowSpec JSON

List Nodes

aliyun CLI:

aliyun dataworks-public list-nodes \
  --project-id {{project_id}} \
  --scene DATAWORKS_PROJECT \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListNodesRequest

request = ListNodesRequest(
    project_id={{project_id}},
    scene='DATAWORKS_PROJECT',
    page_number=1,
    page_size=100
)
response = client.list_nodes(request)
for node in response.body.paging_info.nodes:
    print(f"{node.id}: {node.name}")

Workflow Operations

Create Workflow

The workflow spec must include script.runtime.command: "WORKFLOW", otherwise creation will fail. The correct spec format is as follows:

{
  "version": "2.0.0",
  "kind": "CycleWorkflow",
  "spec": {
    "workflows": [{
      "name": "my_workflow",
      "script": {
        "path": "my_workflow",
        "runtime": {"command": "WORKFLOW"}
      },
      "trigger": {
        "type": "Scheduler",
        "cron": "00 00 02 * * ?",
        "startTime": "1970-01-01 00:00:00",
        "endTime": "9999-01-01 00:00:00",
        "timezone": "Asia/Shanghai"
      }
    }]
  }
}

aliyun CLI:

$PYTHON $SKILL/scripts/build.py ./my_wf > /tmp/wf.json

aliyun dataworks-public create-workflow-definition \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/wf.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateWorkflowDefinitionRequest

with open('/tmp/wf.json') as f:
    spec = f.read()

request = CreateWorkflowDefinitionRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_workflow_definition(request)
print(f"WorkflowId: {response.body.id}")

Update Workflow

aliyun CLI:

aliyun dataworks-public update-workflow-definition \
  --project-id {{project_id}} \
  --id {{workflow_id}} \
  --spec "$(cat /tmp/wf.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import UpdateWorkflowDefinitionRequest

request = UpdateWorkflowDefinitionRequest(
    project_id={{project_id}},
    id='{{workflow_id}}',
    spec=spec
)
client.update_workflow_definition(request)

Get Workflow Details

aliyun CLI:

aliyun dataworks-public get-workflow-definition \
  --project-id {{project_id}} \
  --id {{workflow_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetWorkflowDefinitionRequest

request = GetWorkflowDefinitionRequest(
    project_id={{project_id}},
    id='{{workflow_id}}'
)
response = client.get_workflow_definition(request)

List Workflows

aliyun CLI:

aliyun dataworks-public list-workflow-definitions \
  --project-id {{project_id}} \
  --type CycleWorkflow \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListWorkflowDefinitionsRequest

request = ListWorkflowDefinitionsRequest(
    project_id={{project_id}},
    type='CycleWorkflow',
    page_number=1,
    page_size=100
)
response = client.list_workflow_definitions(request)

Resource File Operations

Create Resource

aliyun CLI:

$PYTHON $SKILL/scripts/build.py ./my_resource > /tmp/res.json

aliyun dataworks-public create-resource \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/res.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateResourceRequest

request = CreateResourceRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_resource(request)
print(f"ResourceId: {response.body.id}")

List Resources

aliyun CLI:

aliyun dataworks-public list-resources \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Function Operations

Create Function

aliyun CLI:

$PYTHON $SKILL/scripts/build.py ./my_func > /tmp/func.json

aliyun dataworks-public create-function \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/func.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateFunctionRequest

request = CreateFunctionRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_function(request)
print(f"FunctionId: {response.body.id}")

List Functions

aliyun CLI:

aliyun dataworks-public list-functions \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Node Dependency Configuration

Dependency Configuration

Inter-node dependencies are maintained exclusively in the spec.dependencies array:

• Upstream nodes must declare outputs.nodeOutputs (${projectIdentifier}.node_name)
• Downstream nodes reference upstream outputs in spec.dependencies
• spec.dependencies[*].nodeId must exactly match the corresponding node's id, otherwise dependencies will not be recognized

{
  "spec": {
    "nodes": [{
      "name": "downstream_node"
    }],
    "dependencies": [{
      "nodeId": "downstream_node",
      "depends": [{
        "type": "Normal",
        "output": "upstream_project.upstream_node_output"
      }]
    }]
  }
}

Deployment Process

Deployment is an asynchronous multi-stage pipeline. For the complete process and detailed instructions, see deploy-guide.md. Below are the API call templates.

Create Deployment (Online)

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreatePipelineRunRequest

# type: Online (deploy) or Offline (take offline)
# object_ids: Only the first entity and its child entities are processed
request = CreatePipelineRunRequest(
    project_id={{project_id}},
    type='Online',
    object_ids=['{{object_id}}']
)
response = client.create_pipeline_run(request)
run_id = response.body.id
print(f"PipelineRunId: {run_id}")

Query Deployment Status

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetPipelineRunRequest

response = client.get_pipeline_run(GetPipelineRunRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}'
))
pipeline = response.body.pipeline.to_map()
print(f"Status: {pipeline['Status']}")
# Status: Init / Running / Success / Fail / Termination / Cancel
for stage in pipeline.get('Stages', []):
    print(f"  {stage['Code']}({stage['Status']}): {stage['Name']}")

Advance Deployment Stage

Python SDK:

from alibabacloud_dataworks_public20240518.models import ExecPipelineRunStageRequest

# code: Stage code, obtained from Stages[].Code returned by get-pipeline-run
# Must advance in order; stages cannot be skipped
# Async trigger; continue polling to confirm results
client.exec_pipeline_run_stage(ExecPipelineRunStageRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}',
    code='{{stage_code}}'  # e.g., PROD_CHECK, PROD
))

View Deployment Items

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListPipelineRunItemsRequest

response = client.list_pipeline_run_items(ListPipelineRunItemsRequest(
    project_id={{project_id}},
    pipeline_run_id='{{pipeline_run_id}}',
    page_number=1,
    page_size=50
))
for item in response.body.paging_info.pipeline_run_items:
    m = item.to_map()
    print(f"{m['Name']}: {m.get('Status', 'N/A')}")

Query Deployment History

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListPipelineRunsRequest

response = client.list_pipeline_runs(ListPipelineRunsRequest(
    project_id={{project_id}},
    page_number=1,
    page_size=20
))
for run in response.body.paging_info.pipeline_runs:
    m = run.to_map()
    print(f"{m['Id']} [{m['Status']}]")

Cancel Deployment

Python SDK:

from alibabacloud_dataworks_public20240518.models import AbolishPipelineRunRequest

client.abolish_pipeline_run(AbolishPipelineRunRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}'
))

Helper Queries

Get Project Information (Convert Between projectId and projectIdentifier)

aliyun CLI:

# Get projectId by projectIdentifier
aliyun dataworks-public get-project \
  --project-identifier my_project_name \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetProjectRequest

request = GetProjectRequest(
    project_identifier='my_project_name'
)
response = client.get_project(request)
print(f"ProjectId: {response.body.id}")
print(f"ProjectIdentifier: {response.body.project_identifier}")

List Data Sources

aliyun CLI:

aliyun dataworks-public list-data-sources \
  --project-id {{project_id}} \
  --type odps \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListDataSourcesRequest

request = ListDataSourcesRequest(
    project_id={{project_id}},
    type='odps',
    page_number=1,
    page_size=100
)
response = client.list_data_sources(request)
for ds in response.body.paging_info.data_sources:
    print(f"{ds.name}: {ds.type}")

List Resource Groups

aliyun CLI:

aliyun dataworks-public list-resource-groups \
  --project-id {{project_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListResourceGroupsRequest

request = ListResourceGroupsRequest(
    project_id={{project_id}}
)
response = client.list_resource_groups(request)
for rg in response.body.resource_groups:
    print(f"{rg.identifier}: {rg.name}")

abolish-pipeline-run

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/AbolishPipelineRun/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Cancel Publishing

aliyun CLI:

aliyun dataworks-public abolish-pipeline-run \
  --project-id {{project_id}} \
  --id {{pipeline_run_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import AbolishPipelineRunRequest

client.abolish_pipeline_run(AbolishPipelineRunRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}'
))

create-component

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreateComponent/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether the component was created by calling list-components and searching by name. Only retry if the component does not exist. Always record the RequestId from the response for traceability.

Create Component

aliyun CLI:

aliyun dataworks-public create-component \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/component.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateComponentRequest

request = CreateComponentRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_component(request)
print(f"ComponentId: {response.body.id}")

create-function

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreateFunction/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether the function was created by calling list-functions and searching by name. Only retry if the function does not exist. Always record the RequestId from the response for traceability.

Create Function

aliyun CLI:

# Build spec JSON (replace placeholders in spec.json with actual values, embed function definition content)
aliyun dataworks-public create-function \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/func.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateFunctionRequest

request = CreateFunctionRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_function(request)
print(f"FunctionId: {response.body.id}")

create-node

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreateNode/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether the node was created by calling list-nodes --Name <node_name>. Only retry if the node does not exist. Always record the RequestId from the response for traceability.

Create Node

Prerequisite: Use build.py to merge the three files (spec.json + code file + properties) into the API input:

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

aliyun CLI:

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

Python SDK:

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_dataworks_public20240518.client import Client
from alibabacloud_dataworks_public20240518.models import CreateNodeRequest
from alibabacloud_tea_openapi.models import Config

credential = CredentialClient()
config = Config(credential=credential)
config.endpoint = 'dataworks.{{region}}.aliyuncs.com'
config.user_agent = 'AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop'
client = Client(config)

with open('/tmp/spec.json') as f:
    spec = f.read()

request = CreateNodeRequest(
    project_id={{project_id}},
    scene='DATAWORKS_PROJECT',
    spec=spec
)
response = client.create_node(request)
print(f"NodeId: {response.body.id}")

Create Node Inside a Workflow

Same as above, after merging with build.py, add --container-id:

aliyun CLI:

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

Python SDK:

request = CreateNodeRequest(
    project_id={{project_id}},
    scene='DATAWORKS_PROJECT',
    container_id='{{workflow_id}}',
    spec=spec
)
response = client.create_node(request)
print(f"NodeId: {response.body.id}")

create-pipeline-run

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreatePipelineRun/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether a pipeline run was already created by calling list-pipeline-runs and filtering by the target object. Only retry if no matching active pipeline run exists. Always record the RequestId from the response for traceability.

## ⚠️ --object-ids Format — Common Mistake

>

The CLI plugin (aliyun dataworks-public ...) accepts --object-ids as space-separated bare values — verified against aliyun dataworks-public create-pipeline-run --help, which prints format: --object-ids value1 value2 value3. Do NOT wrap it in JSON brackets.

>

| ✅ CORRECT | ❌ WRONG |

|---|---|

| --object-ids 7567482277219412494 (bare ID) | --object-ids '["7567482277219412494"]' (JSON array string) — CLI sends the literal text ["7567482277219412494"] as the ID and the API replies 未找到发布对象: [["7567482277219412494"]] |

| --object-ids id1 id2 id3 (space-separated; only `id1` and its children deploy — pass extra IDs as separate calls) | --object-ids [7567482277219412494] (unquoted brackets — shell glob hazard, also wrong) |

>

The IDs are strings on the wire even when they look numeric; pass them unquoted at the shell, the CLI handles the rest. The Python SDK takes a real Python list (object_ids=['ID']) — the JSON-array confusion only ever applied to the SDK style and was incorrectly back-ported into CLI docs.

Create Pipeline Run (Publish / Deploy)

aliyun CLI:

aliyun dataworks-public create-pipeline-run \
  --project-id {{project_id}} \
  --type Online \
  --object-ids {{object_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreatePipelineRunRequest

# type: Online (deploy) or Offline (undeploy)
# object_ids: only the first entity and its child entities will be processed
request = CreatePipelineRunRequest(
    project_id={{project_id}},
    type='Online',
    object_ids=['{{object_id}}']
)
response = client.create_pipeline_run(request)
run_id = response.body.id
print(f"PipelineRunId: {run_id}")

create-resource

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreateResource/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether the resource was created by calling list-resources and searching by name. Only retry if the resource does not exist. Always record the RequestId from the response for traceability.

Create Resource

aliyun CLI:

# Build spec JSON (replace placeholders in spec.json with actual values, embed resource file content)
aliyun dataworks-public create-resource \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/res.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateResourceRequest

request = CreateResourceRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_resource(request)
print(f"ResourceId: {response.body.id}")

create-workflow-definition

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/CreateWorkflowDefinition/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Idempotency Note

This API does not support ClientToken. If the call times out or returns a network error, do not blindly retry. First check whether the workflow was created by calling list-workflow-definitions and searching by name. Only retry if the workflow does not exist. Always record the RequestId from the response for traceability.

Create Workflow

The workflow spec must include script.runtime.command: "WORKFLOW", otherwise the creation will fail. The correct spec format is as follows:

{
  "version": "2.0.0",
  "kind": "CycleWorkflow",
  "spec": {
    "workflows": [{
      "name": "my_workflow",
      "script": {
        "path": "my_workflow",
        "runtime": {"command": "WORKFLOW"}
      },
      "trigger": {
        "type": "Scheduler",
        "cron": "00 00 02 * * ?",
        "startTime": "1970-01-01 00:00:00",
        "endTime": "9999-01-01 00:00:00",
        "timezone": "Asia/Shanghai"
      }
    }]
  }
}

Prerequisite: Use build.py to merge the three files (a workflow directory typically only has spec.json + properties, no code file):

python $SKILL/scripts/build.py ./my_workflow > /tmp/wf.json

aliyun CLI:

aliyun dataworks-public create-workflow-definition \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/wf.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import CreateWorkflowDefinitionRequest

with open('/tmp/wf.json') as f:
    spec = f.read()

request = CreateWorkflowDefinitionRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.create_workflow_definition(request)
print(f"WorkflowId: {response.body.id}")

exec-pipeline-run-stage

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ExecPipelineRunStage/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

## ⚠️ Parameter Name — Common Mistake

>

The pipeline-run identifier parameter is `--id`, NOT --pipeline-run-id.

>

| ❌ WRONG | ✅ CORRECT |

|----------|-----------|

| aliyun dataworks-public exec-pipeline-run-stage --pipeline-run-id <UUID> --code PROD_CHECK | aliyun dataworks-public exec-pipeline-run-stage --id <UUID> --code PROD_CHECK |

>

Calling with --pipeline-run-id returns Error: --id is required and exits non-zero.

>

Naming-convention inconsistency to memorize (the CLI is not uniform across pipeline APIs):

>

| API | Identifier flag |

|---|---|

| exec-pipeline-run-stage | --id |

| get-pipeline-run | --id |

| abolish-pipeline-run | --id |

| list-pipeline-run-items | --pipeline-run-id |

>

When in doubt, run aliyun dataworks-public <command> --help.

Advance Pipeline Run Stage

aliyun CLI:

aliyun dataworks-public exec-pipeline-run-stage \
  --project-id {{project_id}} \
  --id {{pipeline_run_id}} \
  --code {{stage_code}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ExecPipelineRunStageRequest

# code: stage code, obtained from Stages[].Code in the get-pipeline-run response
# stages must be advanced in order; skipping stages is not allowed
# triggered asynchronously; continue polling to confirm the result
client.exec_pipeline_run_stage(ExecPipelineRunStageRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}',
    code='{{stage_code}}'  # e.g., PROD_CHECK, PROD
))

get-component

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetComponent/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get Component Details

aliyun CLI:

aliyun dataworks-public get-component \
  --project-id {{project_id}} \
  --id {{component_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetComponentRequest

request = GetComponentRequest(
    project_id={{project_id}},
    id='{{component_id}}'
)
response = client.get_component(request)
print(response.body.spec)

get-function

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetFunction/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get Function Details

aliyun CLI:

aliyun dataworks-public get-function \
  --project-id {{project_id}} \
  --id {{function_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetFunctionRequest

request = GetFunctionRequest(
    project_id={{project_id}},
    id='{{function_id}}'
)
response = client.get_function(request)
print(response.body.spec)

get-node

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetNode/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get Node Details

aliyun CLI:

aliyun dataworks-public get-node \
  --project-id {{project_id}} \
  --id {{node_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetNodeRequest

request = GetNodeRequest(
    project_id={{project_id}},
    id='{{node_id}}'
)
response = client.get_node(request)
# response.body.spec contains the full FlowSpec JSON

get-pipeline-run

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetPipelineRun/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Query Pipeline Run Status

aliyun CLI:

aliyun dataworks-public get-pipeline-run \
  --project-id {{project_id}} \
  --id {{pipeline_run_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetPipelineRunRequest

response = client.get_pipeline_run(GetPipelineRunRequest(
    project_id={{project_id}},
    id='{{pipeline_run_id}}'
))
pipeline = response.body.pipeline.to_map()
print(f"Status: {pipeline['Status']}")
# Status: Init / Running / Success / Fail / Termination / Cancel
for stage in pipeline.get('Stages', []):
    print(f"  {stage['Code']}({stage['Status']}): {stage['Name']}")

get-project

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetProject/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get Project Information (retrieve projectIdentifier via projectId)

aliyun CLI:

# Retrieve project details by projectId (numeric); ProjectName is the projectIdentifier
aliyun dataworks-public get-project \
  --id {{project_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Note: get-project only accepts the numeric --Id parameter; reverse lookup by projectIdentifier is not supported.

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetProjectRequest

request = GetProjectRequest(
    id={{project_id}}
)
response = client.get_project(request)
# ProjectName in the response is the projectIdentifier

get-resource

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetResource/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get File Resource Details

aliyun CLI:

aliyun dataworks-public get-resource \
  --project-id {{project_id}} \
  --id {{resource_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetResourceRequest

request = GetResourceRequest(
    project_id={{project_id}},
    id='{{resource_id}}'
)
response = client.get_resource(request)
print(response.body.spec)

get-workflow-definition

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/GetWorkflowDefinition/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Get Workflow Details

aliyun CLI:

aliyun dataworks-public get-workflow-definition \
  --project-id {{project_id}} \
  --id {{workflow_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import GetWorkflowDefinitionRequest

request = GetWorkflowDefinitionRequest(
    project_id={{project_id}},
    id='{{workflow_id}}'
)
response = client.get_workflow_definition(request)

import-workflow-definition

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ImportWorkflowDefinition/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Import Workflow (including internal child nodes)

aliyun CLI:

# spec contains the workflow definition and all child node definitions
aliyun dataworks-public import-workflow-definition \
  --project-id {{project_id}} \
  --spec "$(cat /tmp/wf_with_nodes.json)" \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ImportWorkflowDefinitionRequest

request = ImportWorkflowDefinitionRequest(
    project_id={{project_id}},
    spec=spec
)
response = client.import_workflow_definition(request)

list-components

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListComponents/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Components

aliyun CLI:

aliyun dataworks-public list-components \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListComponentsRequest

request = ListComponentsRequest(
    project_id={{project_id}},
    page_number=1,
    page_size=100
)
response = client.list_components(request)

list-compute-resources

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListComputeResources/api.json

Query the list of compute resources bound to the project. Use this API to discover compute engine bindings (EMR Serverless Spark, Hologres, StarRocks, etc.) that may not appear in list-data-sources.

aliyun CLI:

aliyun dataworks-public list-compute-resources \
  --project-id {{project_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Key parameters:

• --project-id (Required) — DataWorks workspace ID
• --EnvType (Optional) — Dev or Prod
• --Types.1, --Types.2, ... (Optional) — Filter by compute resource type

Response fields of interest:

• ComputeResourceList[].Name — Compute resource name (can be used as datasource.name)
• ComputeResourceList[].Type — Compute engine type (e.g., EMR_Serverless, Hologres, StarRocks)
• ComputeResourceList[].EnvType — Environment type (Dev / Prod)

list-data-sources

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListDataSources/api.json

Query the list of data sources in the project.

aliyun CLI:

aliyun dataworks-public list-data-sources \
  --project-id {{project_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListDataSourcesRequest

response = client.list_data_sources(ListDataSourcesRequest(
    project_id={{project_id}}
))
# The response structure depends on the actual SDK version; use .to_map() to inspect

list-functions

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListFunctions/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Functions

aliyun CLI:

aliyun dataworks-public list-functions \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Node Dependency Configuration

list-node-dependencies

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListNodeDependencies/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Node Dependencies

aliyun CLI:

aliyun dataworks-public list-node-dependencies \
  --project-id {{project_id}} \
  --id {{node_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListNodeDependenciesRequest

request = ListNodeDependenciesRequest(
    project_id={{project_id}},
    id='{{node_id}}',
    page_number=1,
    page_size=100
)
response = client.list_node_dependencies(request)

list-nodes

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListNodes/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Nodes

aliyun CLI:

aliyun dataworks-public list-nodes \
  --project-id {{project_id}} \
  --scene DATAWORKS_PROJECT \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListNodesRequest

request = ListNodesRequest(
    project_id={{project_id}},
    scene='DATAWORKS_PROJECT',
    page_number=1,
    page_size=100
)
response = client.list_nodes(request)
for node in response.body.paging_info.nodes:
    print(f"{node.id}: {node.name}")

list-pipeline-run-items

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListPipelineRunItems/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Pipeline Run Items

aliyun CLI:

aliyun dataworks-public list-pipeline-run-items \
  --project-id {{project_id}} \
  --pipeline-run-id {{pipeline_run_id}} \
  --page-number 1 \
  --page-size 50 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListPipelineRunItemsRequest

response = client.list_pipeline_run_items(ListPipelineRunItemsRequest(
    project_id={{project_id}},
    pipeline_run_id='{{pipeline_run_id}}',
    page_number=1,
    page_size=50
))
for item in response.body.paging_info.pipeline_run_items:
    m = item.to_map()
    print(f"{m['Name']}: {m.get('Status', 'N/A')}")

list-pipeline-runs

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListPipelineRuns/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Pipeline Run History

aliyun CLI:

aliyun dataworks-public list-pipeline-runs \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 20 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListPipelineRunsRequest

response = client.list_pipeline_runs(ListPipelineRunsRequest(
    project_id={{project_id}},
    page_number=1,
    page_size=20
))
for run in response.body.paging_info.pipeline_runs:
    m = run.to_map()
    print(f"{m['Id']} [{m['Status']}]")

list-resource-groups

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListResourceGroups/api.json

Query the list of resource groups in the project.

aliyun CLI:

aliyun dataworks-public list-resource-groups \
  --project-id {{project_id}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListResourceGroupsRequest

response = client.list_resource_groups(ListResourceGroupsRequest(
    project_id={{project_id}}
))
# The response structure depends on the actual SDK version; use .to_map() to inspect

list-resources

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListResources/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Resources

aliyun CLI:

aliyun dataworks-public list-resources \
  --project-id {{project_id}} \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

list-workflow-definitions

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/ListWorkflowDefinitions/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

List Workflows

aliyun CLI:

aliyun dataworks-public list-workflow-definitions \
  --project-id {{project_id}} \
  --type CycleWorkflow \
  --page-number 1 \
  --page-size 100 \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import ListWorkflowDefinitionsRequest

request = ListWorkflowDefinitionsRequest(
    project_id={{project_id}},
    type='CycleWorkflow',
    page_number=1,
    page_size=100
)
response = client.list_workflow_definitions(request)

File Resource Operations

move-function

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/MoveFunction/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Move Function to Target Path

aliyun CLI:

aliyun dataworks-public move-function \
  --project-id {{project_id}} \
  --id {{function_id}} \
  --path {{target_path}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import MoveFunctionRequest

request = MoveFunctionRequest(
    project_id={{project_id}},
    id='{{function_id}}',
    path='{{target_path}}'
)
client.move_function(request)

move-node

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/MoveNode/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Move Node Path

aliyun CLI:

aliyun dataworks-public move-node \
  --project-id {{project_id}} \
  --id {{node_id}} \
  --path {{target_path}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import MoveNodeRequest

request = MoveNodeRequest(
    project_id={{project_id}},
    id='{{node_id}}',
    path='{{target_path}}'
)
client.move_node(request)

move-resource

Latest API definition: https://api.aliyun.com/meta/v1/products/dataworks-public/versions/2024-05-18/apis/MoveResource/api.json

If the call returns an error, you can obtain the latest parameter definitions from the URL above.

Move File Resource to Target Directory

aliyun CLI:

aliyun dataworks-public move-resource \
  --project-id {{project_id}} \
  --id {{resource_id}} \
  --path {{target_path}} \
  --user-agent AlibabaCloud-Agent-Skills/alibabacloud-dataworks-datastudio-develop

Python SDK:

from alibabacloud_dataworks_public20240518.models import MoveResourceRequest

request = MoveResourceRequest(
    project_id={{project_id}},
    id='{{resource_id}}',
    path='{{target_path}}'
)
client.move_resource(request)