DataWorks:Assignment node

Example 1: Pass a MaxCompute SQL query result

Example 2: Pass a Python 2 output

The output of a Python 2 print statement is split by commas (,) and passed as a one-dimensional array.

• Assignment node

  The Python 2 code is as follows:

  print 'Electronics, Clothing, Books';

• Shell node

  Add an input parameter named param that references the assignment node's outputs. Use the following script to read the data:

  Use the following script to read the data: # Output the full array echo "Full result set: ${types}" # Output a specific element by index echo "Second element: ${types[1]}"

  DataWorks performs static replacement. Output: Full result set: Electronics,Clothing,Books Second element: Clothing

Background

Assume that you are a data developer for a large internet company. You are responsible for processing data from three core lines of business: E-commerce (ecom), finance (finance), and logistics (logistics). More lines of business may be added in the future. You need to run the same aggregation logic on the user behavioral logs of these three lines of business every day. The logic calculates the daily popularity (PV) for each user and stores the results in a unified aggregate table.

• Upstream source tables (DWD layer):

  • dwd_user_behavior_ecom_d: E-commerce user behavior table.

  • dwd_user_behavior_finance_d: Finance user behavior table.

  • dwd_user_behavior_logistics_d: Logistics user behavior table.

  • dwd_user_behavior_${line-of-business}_d: User behavior tables for other potential lines of business.

  • These tables have the same schema and are partitioned by day (dt).

• Downstream target table (DWS layer):

  • dws_user_summary_d: User aggregate table.

  • This table is partitioned by both line of business (biz_line) and day (dt). It is used to store the aggregated results for all lines of business.

Creating a separate task for each line of business is costly to maintain and prone to errors. If you use a for-each node, you only need to maintain one set of processing logic. The system automatically traverses all lines of business to complete the calculations.

Data preparation

First, create the example tables and insert test data. This example uses the data timestamp 20251010.

1. Associate a MaxCompute computing resource to the workspace.

2. Go to DataStudio to perform data development and create a MaxCompute SQL node.

3. Create the source tables (DWD layer): Add the following code to the MaxCompute SQL node, select it, and run it.

   -- E-commerce user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of E-commerce user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_ecom_d PARTITION (dt='20251010') VALUES
   ('user001', 'click',        1760004060000), -- 2025-10-10 10:01:00.000
   ('user002', 'browse',       1760004150000), -- 2025-10-10 10:02:30.000
   ('user001', 'add_to_cart',  1760004300000); -- 2025-10-10 10:05:00.000
   -- Verify that the E-commerce user behavior table is created.
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   
   -- Finance user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of finance user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_finance_d PARTITION (dt='20251010') VALUES
   ('user003', 'open_app',      1760020200000), -- 2025-10-10 14:30:00.000
   ('user003', 'transfer',      1760020215000), -- 2025-10-10 14:30:15.000
   ('user003', 'check_balance', 1760020245000), -- 2025-10-10 14:30:45.000
   ('user004', 'open_app',      1760020300000); -- 2025-10-10 14:31:40.000
   -- Verify that the finance user behavior table is created.
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   
   -- Logistics user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of logistics user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_logistics_d PARTITION (dt='20251010') VALUES
   ('user001', 'check_status',    1760032800000), -- 2025-10-10 18:00:00.000
   ('user005', 'schedule_pickup', 1760032920000); -- 2025-10-10 18:02:00.000
   
   -- Verify that the logistics user behavior table is created.
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Create the target table (DWS layer): Add the following code to the MaxCompute SQL node, select it, and run it.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily popularity',
   ) 
   COMMENT 'Daily user popularity aggregate table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition in yyyymmdd format',
       biz_line     STRING COMMENT 'Line-of-business partition, such as ecom, finance, logistics'
   );

   Important

   If the workspace uses the standard environment, you must publish this node to the production environment and perform a data backfill.

Workflow implementation

1. Create a workflow. In the Scheduling Parameters pane on the right, set the scheduling parameter bizdate to the previous day $[yyyymmdd-1].

   

2. In the workflow, create an assignment node named get_biz_list. Write the following code in MaxCompute SQL. This node outputs a list of lines of business to process.

   -- Output all lines of business to be processed
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. Configure the for-each node

   • Return to the workflow canvas and create a downstream for-each node for the get_biz_list assignment node.

   • Go to the settings page of the for-each node. On the Schedule tab on the right, under Scheduling Parameters > Script Parameters, set the value of the loopDataArray parameter to the outputs of the get_biz_list node.

     

   • In the for-each node loop body, click Create Inner Node. Create a MaxCompute SQL node and write the processing logic for the loop body.

     Note

     • This script is driven by the for-each node and runs once for each line of business.

     • The built-in variable ${dag.foreach.current} is dynamically replaced with the current line-of-business name at runtime. The expected iteration values are 'ecom', 'finance', and 'logistics'.

     SET odps.sql.allow.dynamic.partition=true;
     
     INSERT OVERWRITE TABLE dws_user_summary_d PARTITION (dt='${bizdate}', biz_line)
     SELECT
         user_id,
         COUNT(*) AS pv,
         '${dag.foreach.current}' AS biz_line
     FROM
         dwd_user_behavior_${dag.foreach.current}_d
     WHERE
         dt = '${bizdate}'
     GROUP BY
         user_id;

4. Add a verification node

   Return to the workflow canvas. For the for-each node, click Create Downstream Node. Create a MaxCompute SQL node and add the following code.

   SELECT * FROM dws_user_summary_d WHERE dt='20251010' ORDER BY biz_line, user_id;

Publish and run

Publish the workflow to the production environment. In Operation Center, navigate to Auto Triggered Node O&M > Auto Triggered Nodes, find the target workflow, run a smoke test, and select '20251010' as the data timestamp.

After the run is complete, view the run log in the test instance. The expected output of the final node is:

Advantages of this solution

• High extensibility: If a new line of business is added, you only need to add one line of SQL code in the assignment node. You do not need to modify the processing logic.

• Easy maintenance: All lines of business share the same processing logic. A change in one place takes effect for all of them.