ds_create_dag_workflow
Create a generic DAG workflow for any task type: SQL, Shell, Python, HTTP, or sub-process. Define task dependencies, retry policies, and optional cron schedules to automate data pipelines.
Instructions
Create a generic DAG workflow supporting any task type (SQL/SHELL/PYTHON/DEPENDENT/SUB_PROCESS/HTTP etc.).
Args:
project_name: Project name
name: Workflow name
tasks: Task definition list (see examples below)
relations: Dependency list [{"from": "taskA", "to": "taskB"}]; empty "from" = start node
description: Description
schedule: Whether to create a schedule
schedule_cron: Cron expression (7-field Quartz style)
locations: Optional node coordinates [{"task_name": "check", "x": 100, "y": 100}]
Leave empty for auto-layout (horizontal, 300x200 spacing)
tasks example:
[
# SHELL task (script required, resource_list optional)
{"name": "check", "type": "SHELL",
"script": "#!/bin/bash
python3 /public/check_partition.py table_name $[yyyyMMdd-1]", "resource_list": [67], # optional: referenced resource id or path "fail_retry_times": 3}, # optional: retry count on failure
# SQL task (datasource_id + sql required)
{"name": "sql1", "type": "SQL",
"datasource_id": 1, "sql": "SELECT 1",
"sql_type": "HIVE", # optional, default HIVE
"sql_type_select": 1}, # optional, 0=query 1=non-query (default 1)
# DEPENDENT task (wait for upstream workflow completion)
{"name": "wait", "type": "DEPENDENT",
"depend_items": [{"project_code": 123, "definition_code": 456,
"cycle": "day", "date_value": "today"}]},
# SUB_PROCESS task (invoke a sub-workflow)
{"name": "sub", "type": "SUB_PROCESS",
"sub_process_code": 21505676237440},
# HTTP task
{"name": "notify", "type": "HTTP",
"http_url": "https://api.example.com/callback",
"http_method": "POST"}
]
relations example:
[
{"from": "", "to": "check"}, # check is a start node (empty "from")
{"from": "check", "to": "sql1"},
{"from": "wait", "to": "sql1"} # wait and check run in parallel, both feed sql1
]
resource_list format:
- Recommended: resource_id (int): [67, 58]
- Or full path (str): ["/public/check_partition.py"]
- Paths must start with /
- Use ds_list_resources() to view all resources with their ids and paths
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| tasks | Yes | ||
| schedule | No | ||
| locations | No | ||
| relations | Yes | ||
| description | No | ||
| project_name | Yes | ||
| schedule_cron | No | 0 0 6 * * ? * |