DataWorks:Triggered workflow

Step 1: Create a triggered workflow

1. Go to the Workspaces page in the DataWorks console. In the top navigation bar, select a desired region. Find the desired workspace and choose Shortcuts > Data Studio in the Actions column.

2. In the left-side navigation pane, click . Next to Business Flow, click > Create Workflow to open the Create Workflow page.

3. On the Create Workflow page, set Scheduling Type to Triggered Scheduling, enter a workflow Name, and click OK.

Step 2: Orchestrate the workflow and develop a node

1. In the toolbar, click + Add Node to open the node list. From the node type list on the left, drag a Shell node to the canvas and enter a name to create it.

2. Double-click the Shell node to open the code editor and enter the following code:

   echo "Hello, Trigger Workflow! Current time is ${bizdate}"

3. Click Save in the toolbar.

Step 3: Debug in the development environment

1. Return to the workflow canvas and click the icon in the top toolbar.

2. In the dialog box that appears, enter the Current run value for the workflow (for example, if the current date is 20260310, bizdate should be replaced with 20260309).

3. In the run log at the bottom, you can view the node's execution status and the output of the echo command.

Step 4: Publish and run in production

1. On the workflow canvas, click the Publish button and follow the prompts to publish the workflow.

2. After the workflow is published, go to Operation Center > Manually Triggered Node O&M > Manual Task> Triggered Workflow.

3. Find the workflow that you published and click Run in the Actions column.

4. In the dialog box that appears, click Run again to trigger an instance of the workflow in the production environment. You can view the details of this run on the Manual Instance page.

You have now mastered the basics of using triggered workflows. Next, we explore their more powerful event-driven capabilities.

Step 1: Create an OSS trigger

1. Go to Operation Center > Tenant Schedule Setting > Trigger Management.

2. Click Create Trigger and configure the following parameters:

   Note

   For detailed parameter descriptions, see OSS trigger.

   • Trigger name: Enter a custom name, such as oss_new_file_trigger.

   • Applicable workspace: Select the target workspace where the workflow is located.

   • Trigger event type: Select Object Storage Service (OSS).

   • Trigger event: Select oss:ObjectCreated:PutObject or another upload event.

   • Bucket name: Select your OSS bucket.

   • File name: Specify the file path and format to monitor. Wildcards are supported. For example, to monitor the input/ directory for all .csv files, enter input/*.csv.

   • Role configuration: The first time you use this feature, you must complete a one-click authorization. Select the generated role named DataWorks-EventBridge-OSS-MNS-Role-*************.

     The ************* represents a randomly generated 13-digit ID that ensures the role name is unique.

3. Click OK to create the trigger.

Step 2: Create and associate the workflow

1. Follow the steps in Quick start: Create a manually triggered workflow to create a triggered workflow named process_oss_file_workflow.

2. On the right side of the workflow canvas, in the Properties panel, select Scheduling Policy.

3. From the Trigger drop-down list, select the oss_new_file_trigger that you just created.

   

Step 3: Develop node and parse event parameters

1. In the toolbar, click + Add Node to open the node list. From the node type list on the left, drag a Shell node to the canvas and enter a name to create it.

2. Double-click the node and enter the following code to get and print the file path from the trigger event.

   # When a workflow is triggered, event information is passed via the built-in variable workflow.triggerMessage.
   # You can use ${workflow.triggerMessage.data.oss.object.key} to get the full path of the uploaded file.
   
   echo "========= Start Processing OSS File ========="
   message="${workflow.triggerMessage}"
   echo "Raw Value: ${message}"
   
   # Extract the file path from the event message.
   FILE_PATH="${workflow.triggerMessage.data.oss.object.key}"
   echo "A new file has arrived: ${FILE_PATH}"
   
   # Add your specific processing logic here.
   
   echo "========= Finish Processing OSS File ========="

   Note

   ${workflow.triggerMessage} retrieves the full event message body in JSON format. You can find the specific OSS message format in EventBridge by navigating to Event Buses > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracking > Event Details.

   Example OSS message format (the value of workflow.triggerMessage)

   {
       "datacontenttype": "application/json;charset=utf-8",
       "aliyunaccountid": "1***********9",
       "data": {
           "eventVersion": "1.0",
           "responseElements": {
               "requestId": "69B1***********C0A8"
           },
           "eventSource": "acs:oss",
           "eventTime": "2026-03-11T05:40:45.000Z",
           "requestParameters": {
               "sourceIPAddress": "***********"
           },
           "eventName": "ObjectCreated:PostObject",
           "userIdentity": {
               "principalId": "1***********9"
           },
           "region": "cn-hangzhou",
           "oss": {
               "bucket": {
                   "name": "******",
                   "arn": "acs:oss:cn-hangzhou:1***********9:******",
                   "virtualBucket": "",
                   "ownerIdentity": "1***********9"
               },
               "ossSchemaVersion": "1.0",
               "object": {
                   "size": 59537,
                   "objectMeta": {
                       "mimeType": "text/csv"
                   },
                   "deltaSize": 0,
                   "eTag": "63***********D32",
                   "key": "input/***********.csv"
               }
           }
       },
       "subject": "acs:oss:cn-hangzhou:1***********9:dwoss1024/input/******.csv",
       "aliyunoriginalaccountid": "1***********9",
       "source": "acs.oss",
       "type": "oss:ObjectCreated:PostObject",
       "aliyunpublishtime": "2026-03-11T05:40:45.682Z",
       "specversion": "1.0",
       "aliyuneventbusname": "DATAWORKS_TRIGGER_FOR_BUCKET_******",
       "id": "69B1***********0A8",
       "time": "2026-03-11T05:40:45.000Z",
       "aliyunregionid": "cn-hangzhou"
   }

Step 4: Debug and publish

1. Debug:

   • Return to the workflow canvas and click the Run button.

   • In the Trigger Message Body input box, paste a simulated OSS event in JSON format. You can copy an example message from the trigger configuration page and modify the value of the key field. The following is a minimal example:

     {
       "data": {
         "oss":{
           "object": {
             "key": "input/test_file_20260310.csv" 
           }
         } 
       }
     }

   • Click Run and check the logs to confirm that input/test_file_20260310.csv is printed successfully.

2. Publish: After successful debugging, click the Publish button to deploy the workflow to the production environment. Event-driven triggering is active only in the production environment.

Step 5: Verify in production

1. Using the OSS console or a client, upload a CSV file to the bucket and path that you configured in the trigger, such as the input/ directory.

   How to ensure the event is triggered successfully

   Go to https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing to query the list of recent trigger events. You can also click Event Details for a specific event to view its trigger message, which is the value of workflow.triggerMessage.

   

2. In DataWorks, go to Operation Center > Manually Triggered Node O&M > Manual Task > Triggered Workflow. The published process_oss_file_workflow appears in the list.

   

3. After a moment, a new workflow instance is automatically triggered and run. You can view it in DataWorks at Operation Center > Manually Triggered Node O&M > Manual Instance. Click the instance to view its logs and confirm that the file path was processed correctly.

Scenario 2: Process ApsaraMQ for Kafka messages

Objective: Monitor a Kafka topic for user behavior logs. When a new message arrives, trigger a workflow to parse it and execute different logic based on its content.

Workflow orchestration

Orchestrating a triggered workflow is similar to orchestrating a scheduled workflow. For more information, see Node and workflow orchestration.

Scheduling parameters

In the Properties panel on the right side of the workflow canvas, you can set global parameters for the workflow. All nodes within it can reference these parameters.

• Reference syntax: In node code, reference a workflow parameter by using the ${workflow.parameter_name} format.

• Parameter priority: Parameters in DataWorks have a hierarchical override relationship. The order of precedence is: node parameters > workflow parameters.

  For more information about parameters, see Parameter design and flow.

Scheduling policy

When multiple workflows trigger simultaneously and cause a resource bottleneck, use priority and priority weighting policies to manage resource scheduling and ensure that the most important tasks are executed first.

• Ensure core business continuity: Set a higher priority for critical business workflows so they always run before non-critical workflows.

• Reduce critical path runtime: Within a single workflow instance, you can use a priority weighting policy to influence the execution order of nodes. For example, using the downward weighting policy, nodes on the critical path with more upstream dependencies gain a higher dynamic weight. This prioritizes their execution, which shortens the entire workflow runtime.

  Parameter	Description
  Priority	Defines the absolute priority level of a workflow instance in the scheduling queue. Available levels are 1, 3, 5, 7, and 8, where a higher number indicates a higher priority. High-priority tasks or workflows are always allocated scheduling resources before low-priority ones.
  Priority weighting policy	Defines how the weights of individual nodes (tasks) are dynamically calculated within the same priority level. Nodes with higher weights are given execution priority. No weighting: All nodes have a fixed baseline weight. Downward weighting: The weight of a node is dynamically adjusted. The more upstream nodes it depends on, the higher its weight. This strategy helps prioritize the execution of nodes on the critical path in a directed acyclic graph (DAG). The weight is calculated as follows: Initial weight + Sum of the priorities of all its upstream nodes.
  Maximum parallel instances	Controls the maximum number of instances of this workflow that can run at the same time for concurrency control and resource protection. When this limit is reached, newly triggered instances enter a waiting state. You can set this to Unlimited or a custom value up to 100,000. Note If the specified limit exceeds the maximum capacity of the resource group, the actual concurrency is determined by the physical limit of the resource group.

The DataWorks priority system follows a hierarchical override rule: runtime specification > node-level configuration > workflow-level configuration.

1. Workflow-level configuration (baseline): Configured in the workflow's Scheduling Policy panel, this serves as the default setting for all nodes.

2. Node-level configuration (override): You can override the workflow-level setting for a specific node by setting a higher priority in its Properties > Scheduling Policy panel.

3. Runtime specification (temporary): When manually triggering a run in Operation Center, you can specify a priority by using the Override Priority at Runtime switch. This configuration has the highest precedence and applies only to the current run, without modifying any permanent settings.