DataWorks:Triggered workflows

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:

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

   • Workspace: Select the workspace where the workflow is located.

   • Trigger Type: Select OSS.

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

   • Bucket name: Select the OSS bucket to monitor.

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

   • Configure Role: For first-time use, click authorization and select the generated role named **DataWorks-EventBridge-OSS-MNS-Role-\*\*\*\*\*\*\*\*\*\*\*\*\*** (the 13-digit suffix ensures uniqueness).

   For detailed parameter descriptions, see OSS trigger.

3. Click OK to create the trigger.

Step 2: Create and associate the workflow

1. Follow Step 1: Create a triggered workflow to create a triggered workflow named process_oss_file_workflow.

2. In the right-side panel of the workflow canvas, select Properties > Scheduling Policies.

3. From the Trigger dropdown list, select the oss_new_file_trigger you just created.

   

Step 3: Parse the event payload

When a trigger fires, DataWorks passes the event information through the built-in variable ${workflow.triggerMessage}. This variable contains the complete event message in JSON format. Use ${workflow.triggerMessage.data.oss.object.key} to access the path of the uploaded file.

1. Click + Add Node in the toolbar, drag a Shell node to the canvas, and enter a name.

2. Double-click the node and enter the following code:

   To inspect the full OSS event message format, go to EventBridge and navigate to Event Buses > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracking > Event Detail. An example message is shown below.

   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 ========="

   Example OSS message format

   {
       "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 deploy

1. Debug: Return to the workflow canvas and click the Run button. In the Trigger Message Body input box, paste a sample OSS event in JSON format. Copy the message format from the trigger configuration page and modify the key value. The following is a minimal example:

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

   Click Run and confirm that input/test_file_20260310.csv is printed in the logs.

2. Deploy: After successful debugging, click Deploy to deploy the workflow to the production environment.

   Important

   Event triggering only takes effect 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 configured in the trigger (for example, the input/ directory).

   How do I ensure that event triggering is successful?

   To confirm the event was received, go to https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing and check the list of recently triggered events. Click Event Detail to view the full message payload (workflow.triggerMessage).

   

2. Go to DataWorks Operation Center > Manually Triggered Node O&M > Manually Triggered Node > Triggered Workflow. The deployed process_oss_file_workflow is listed there.

   

3. After a short wait, go to Operation Center > Manually Triggered Node O&M > Triggered Workflow Instance. A new workflow instance should appear. Click to view its logs and confirm that the file path was processed correctly.

Step 1: Create a Kafka trigger

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

2. Configure the following parameters:

   • Trigger Name: kafka_user_action_trigger.

   • Trigger Type: Select ApsaraMQ for Kafka.

   • Kafka Instance and Topic: Select the instance and topic to monitor.

   • ConsumerGroupId: Select Quick Create to have the system auto-generate a consumer group ID and avoid conflicts with other applications.

   • Key (Optional): Specify a message key. Only messages with an exact key match trigger the workflow.

3. Click OK.

Step 2: Create and associate the workflow

1. Follow Step 1: Create a triggered workflow to create a triggered workflow named handle_user_action_workflow.

2. In the right-side panel of the workflow canvas, select Properties > Scheduling Policies.

3. In the Trigger dropdown list, select the kafka_user_action_trigger you just created.

   

4. Because Kafka messages can arrive at a high frequency, set Maximum Parallel Instances for Internal Tasks to a value such as 100 to prevent message spikes from overwhelming the scheduling resources.

Step 3: Parse the event payload

The workflow.triggerMessage variable holds the complete Kafka event in JSON format. However, the value field inside the message is a JSON-encoded string, not a JSON object. This means your code must parse the value field a second time after retrieving it.

For example, if the value field contains {"user_id": "1001", "action_type": "login", "timestamp": 1688888888}:

1. Click + Add Node in the toolbar, drag a Python node to the canvas, and enter a name.

2. Enter the following code:

   import json
   
   # Step 1: Retrieve the 'value' field from the Kafka message.
   # This is a JSON-encoded string, not a JSON object.
   message_value_str = '${workflow.triggerMessage.value}'
   
   print(f'Received raw message value string: {message_value_str}')
   
   try:
       # Step 2: Parse the string into a Python dictionary.
       message_data = json.loads(message_value_str)
   
       user_id = message_data.get("user_id")
       action_type = message_data.get("action_type")
       print(f"Successfully parsed message. User ID: {user_id}, Action: {action_type}")
   
       # Step 3: Execute different logic based on action_type.
       if action_type == 'login':
           # o.run_sql(f"INSERT OVERWRITE TABLE user_login_record PARTITION(ds='{bizdate}') VALUES ('{user_id}');")
           print("Processing login action...")
       elif action_type == 'purchase':
           print("Processing purchase action...")
       else:
           print("Unknown action type.")
   
   except json.JSONDecodeError as e:
       print(f"Error decoding JSON: {e}")
       # Write the error to a dedicated log table for troubleshooting.
       raise e  # Re-raise to mark the node as failed.

Step 4: Debug and deploy

1. Debug: Return to the workflow canvas and click the Run button. In the Trigger Message Body input box, paste the following simulated Kafka event. Note that the value field is an escaped JSON string:

   {
     "topic": "user-behavior-topic",
     "key": "some-key",
     "value": "{\"user_id\": \"1001\", \"action_type\": \"login\", \"timestamp\": 1688888888}"
   }

   Run and check the logs to confirm that the Python node correctly parses user_id and action_type.

2. Deploy: After successful debugging, deploy the workflow to the production environment.

Step 5: Verify in production

1. Send a message in the correct format to the configured Kafka topic.

   

2. Go to Operation Center > Manually Triggered Node O&M > Manually Triggered Node > Triggered Workflow. The deployed handle_user_action_workflow is listed there.

   

3. Go to Operation Center > Manually Triggered Node O&M > Manually Triggered Node Instance > Triggered Workflow Instance to confirm that a new instance was triggered, and check the runtime log.