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 , and then click > Create Workflow on the right side of Project Directory to go to the Create Workflow page.

3. In the dialog box that appears, on the Create Workflow page, set Scheduling Type to Triggered scheduling. Enter a workflow Name and click Confirm to create the workflow.

Step 2: Orchestrate the workflow and develop nodes

1. Click + Add Node on the toolbar to open the node list. Drag a Shell node from the node type list on the left onto the canvas, enter a name, and complete the creation.

2. Double-click the Shell node to go to the code editing page, and enter the following code:

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

3. Click the Save button on the toolbar.

Step 3: Debug and run (development environment)

1. Go back to the workflow canvas and click the icon on the top toolbar.

2. In the dialog box that appears, enter the Value Used in This Run for the workflow (for example, if today is 20260310, bizdate should be replaced with 20260309).

3. After a moment, in the run log at the bottom, you can see the running status of the node and the output of the echo command.

Step 4: Deploy and run (production environment)

1. On the workflow canvas, click the Deploy button and follow the prompts to complete the deployment process.

2. After successful deployment, go to Operation Center > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow.

3. Find the workflow you just deployed, and click Run in the Operation 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 Instances page.

You have now mastered the basic usage of triggered workflows. Next, we will explore the more powerful event-triggered capabilities.

Step 1: Create an OSS trigger

1. Go to Operation Center > Scheduling Settings > Trigger Management.

2. Click Create Trigger and configure as follows:

   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 resides.

   • 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 all .csv files in the input/ directory, enter input/*.csv.

   • Role Configuration: For first-time use, perform One-Click Authorization and select the role named DataWorks-EventBridge-OSS-MNS-Role-*************.

     ************* represents a randomly generated 13-digit ID number used for uniqueness.

3. Click Confirm to complete the trigger creation.

Step 2: Create and associate a workflow

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

2. In the right-side panel of the workflow canvas, navigate to Scheduling Settings > Scheduling Policy.

3. In the Trigger drop-down list, select the oss_new_file_trigger trigger you just created.

   After selecting a trigger, you can use ${workflow.triggerMessage} in internal tasks to obtain the complete message body, or use ${workflow.triggerMessage.xxx} to obtain the value of a specific field in the message body.

Step 3: Develop a node and parse event parameters

1. Click + Add Node on the toolbar to open the node list. Drag a Shell node from the node type list on the left onto the canvas, enter a name, and complete the creation.

2. Double-click the node and write code to retrieve and print the file path from the trigger event.

   1. General usage: The following parameter usage applies to most event messages.

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

   2. Special case: Some nodes, such as Shell nodes, have their own syntax rules. Before the node runs, the scheduling engine first substitutes the message content into the script text and then passes it to the Shell interpreter for execution. As a result, the quotation marks, spaces, and line breaks in the message directly participate in Shell syntax parsing. This may cause the spaces or line breaks in the parameter to break the original code structure after the node retrieves the parameter. In this case, you can use a heredoc to read the message, which avoids most special-character issues:

      # Use a heredoc to safely capture the raw text substituted from ${workflow.triggerMessage}.
      message=$(cat <<'__DW_TRIGGER_MESSAGE__'
      ${workflow.triggerMessage}
      __DW_TRIGGER_MESSAGE__
      )
      
      # Scenario 1: Get the entire raw message
      printf 'Raw Value: %s\n' "$message"
      
      # Scenario 2: If you need only a specific field, use Python to parse the JSON and extract the value
      dt=$(printf '%s' "$message" | /home/tops/bin/python3 -c 'import json,sys; print(json.load(sys.stdin).get("body",{}).get("dt",""))')
      printf 'dt=%s\n' "$dt"

      Note

      ${workflow.triggerMessage}: Retrieves the complete event message body in JSON format. You can obtain the specific message format for OSS from EventBridge > Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

      View a sample OSS message format, which is the format 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 JSON. You can copy and modify the key value from the "Message Format Sample" on the trigger configuration page. The following is a simple example.

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

   • Click Run and check whether input/test_file_20260310.csv is successfully printed in the logs.

2. Publish: After debugging is successful, click the Publish button to deploy the workflow to the production environment. Event triggers take effect only in the production environment.

Step 5: Verify in production

1. Use the OSS console or a client tool to upload a CSV file to the bucket and path (for example, the input/ directory) that you configured in the trigger.

   How do I verify that the event trigger works?

   Go to https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing to query the list of recently triggered events. You can also click Event Details to view the specific trigger message (that is, workflow.triggerMessage).

   The event list displays trigger records with the event source Object Storage Service (OSS) / acs.oss and the event type oss:ObjectCreated:PostObject.

2. Go to DataWorks Operation Center > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow. The successfully published process_oss_file_workflow appears.

3. Wait a moment and then go to DataWorks Operation Center > Manually Triggered Task O&M > Triggered Workflow Instance. A new workflow instance is automatically triggered. Click to view its logs and verify that the file path is correctly processed.

   ===== Start Processing 0SS File =====
   Raw Value: {"datacontenttype":"application/json;charset=utf-8","aliyunaccountid":"1162423445433459","data":{"eventVersion":"1.0","responseElements":{"requestId":"69B1083F7A439F343040ABCD"},
   "eventSource":"acs:oss","eventTime":"2026-03-11T06:14:23.000Z","requestParameters":{"sourceIPAddress":"140.205.11.13"},"eventName":"0bjectCreated:Post0bject","userIdentity":{"principalId":"1162423445433459"},
   "region":"cn-hangzhou","oss":{"bucket":{"name":"dwoss1024","arn":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024","virtualBucket":"","ownerIdentity":"1162423445433459"},"ossSchemaVersion":"1.0","object":{"size":59537,"objectMeta":
   {"mimeType":"text/csv"},"deltaSize":0,"eTag":"63B4BA5A45AEFC679B9A917E8DDF0D32","key":"input/2013-2020-global-PS4-game-sales.csv"}}},"subject":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024/input/2013-2020-global-PS4-game-sales.csv","aliyunoriginalaccountid":"1162423445433459","source":"acs.oss","type":"oss:0bjectCreated:Post0bject","aliyunpublishtime":"2026-03-11T06:14:23.959Z","specversion":"1.0","aliyuneventbusname":"DATAW0RKS_TRIGGER_F0R_BUCKET_dwoss1024","id":"69B1083F7A439F343040ABCD","time":"2026-03-11T06:14:23.000Z","aliyunregionid":"cn-hangzhou"}A new file has arrived: input/2013-2020-global-PS4-game-sales.csv
   ====== Finish Processing 0SS File =========

Scenario 2: Kafka message arrival drives real-time computation

Objective: Monitor user behavior logs in Kafka. When new messages arrive, trigger a workflow to parse them and execute different logic based on the content.

Workflow orchestration

The core orchestration process for triggered workflows is similar to that of scheduled workflows. For more information, see Node/workflow orchestration.

Scheduling parameters

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

• Reference method: In node code, reference workflow parameters in the format ${workflow.parameter_name}.

• Parameter priority: Parameters in DataWorks follow a hierarchical override rule. The priority order is: Node parameters > Workflow parameters.

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

Scheduling policy

When multiple workflows or tasks are triggered simultaneously and system resources become a bottleneck, you can use Priority and Weighting Strategy to implement intelligent resource scheduling and ensure that the most important tasks are executed first.

• Safeguard core business: Set a higher priority for core business workflows so that they always run before other non-core workflows.

• Reduce critical path duration: Within the same workflow instance, by using Priority weighting strategy, you can influence the execution order of nodes. For example, with the Downstream weighting strategy, nodes on the critical path that have more upstream dependencies receive higher dynamic weights and are executed first, effectively reducing the overall workflow run duration.

  Configuration item	Description
  Priority	Defines the absolute priority level of workflow instances in the scheduling queue. Available levels are 1, 3, 5, 7, and 8 (the larger the number, the higher the priority). Tasks or workflows with higher priority always acquire scheduling resources before those with lower priority.
  Priority weighting strategy	Defines how the dynamic weight of each node (task) within a workflow is calculated at the same priority level. Nodes with higher weights are executed first. No weighting: All nodes have the same fixed baseline weight. Downstream weighting: Node weights are dynamically adjusted. The more upstream dependencies a node has, the higher its weight. This strategy helps nodes on the critical path of the DAG (directed acyclic graph) to be executed first. The weight is calculated as: initial weight value + sum of priorities of all upstream nodes.
  Maximum Parallel Instances for Internal Tasks	Controls the maximum number of instances of this workflow that can run concurrently. This is used for concurrency control and resource protection. When the number of running instances reaches the limit, subsequently triggered new instances enter a waiting state. You can set it to Allowed or specify a custom maximum value (up to 100,000). Note If you set the limit to a value that exceeds the maximum capacity of the resource group, the actual concurrency bottleneck is determined by the physical limit of the resource group.

The priority system in DataWorks follows a hierarchical override rule: Runtime specification > Node-level configuration > Workflow-level configuration.

1. Workflow-level configuration (baseline): Configured in the Scheduling Policy of the workflow and serves as the default setting for all nodes.

2. Node-level configuration (local): In the Scheduling Settings > Scheduling Policy of an individual node within the workflow, you can set a higher Priority for a specific node, which overrides the workflow-level setting.

3. Runtime specification (temporary): Specified through the Runtime Priority Reset toggle when you manually trigger a run in Operation and Maintenance Center. This configuration has the highest priority, takes effect only for the current run, and does not modify any permanent configuration.