DataWorks:Assignment node

Example 1: Pass MaxCompute SQL query results

Example 2: Pass Python 2 output results

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

• Upstream node (assignment node - Python 2) configuration

  The Python 2 code is as follows:

  print 'Electronics, Clothing, Books';

• Downstream node (Shell node) configuration and output

  In the Shell node, add an input parameter named types and reference the outputs parameter of the upstream assignment node.

  Write the following code to read the data: # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  DataWorks directly parses the parameter and performs static substitution. The output is as follows: Entire result set: Electronics,Clothing,Books Second element: Clothing

Background

Assume you are a data development engineer at a comprehensive internet company and are responsible for processing data from three core business lines: e-commerce (ecom), finance (finance), and logistics (logistics), with the possibility of adding more business lines in the future. You need to run the same aggregation logic on the user behavior logs of these three business lines every day to calculate the daily page views (PV) per user and store 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_${business_line}_d: User behavior tables for more potential business lines in the future.

  • 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 double-partitioned by business line (biz_line) and day (dt) to store the aggregated results from all business lines in a unified manner.

Creating a separate task for each business line results in high maintenance costs and is error-prone. By using a for-each node, you only need to maintain a single set of processing logic, and the system automatically iterates through all business lines to complete the computation.

Data preparation

First, create the sample tables and insert test data (using business date 20251010 as an example):

1. Associate a compute resource with the workspace.

2. Go to Data Studio for data development and create a MaxCompute SQL node.

3. Create the source tables (DWD layer): Add the following code to the MaxCompute SQL node 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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'E-commerce user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   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 e-commerce user behavior table created successfully
   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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Finance user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   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 finance user behavior table created successfully
   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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Logistics user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   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 logistics user behavior table created successfully
   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 and run it.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily activity count'
   ) 
   COMMENT 'User daily activity summary table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition, format yyyymmdd',
       biz_line     STRING COMMENT 'Business line partition, e.g. ecom, finance, logistics'
   );

   Important

   If the workspace uses the standard mode, you need to deploy this node to the production environment and backfill data.

Workflow implementation

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

2. In the workflow, create an assignment node named get_biz_list and write the following code in MaxCompute SQL. This node outputs the list of business lines to be processed:

   -- Output all business lines 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

   • Go back to the workflow page and create a downstream for-each node for the assignment node get_biz_list.

   • Open the for-each node settings page. In the Scheduling Parameters > Script Parameters section under schedule settings on the right side, bind the loopDataArray parameter to the outputs of the get_biz_list node.

   • In the loop body of the for-each node, click Create Internal Node and create a MaxCompute SQL node. Write the processing logic inside the loop body.

     Note

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

     • The built-in variable ${dag.foreach.current} is dynamically replaced with the current business line name at each iteration. The expected iteration values are: 'ecom', 'finance', '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

   Go back to the workflow. Click Create Downstream on the for-each node to 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;

Deployment and results

Deploy the workflow to the production environment. Go to the Auto Triggered Task O&M > Auto Triggered Task page in Operation Center. Find the target workflow and perform smoke testing with the business date set to '20251010'.

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

Advantages

• High scalability: When you add a new business line, you only need to add one line of SQL in the assignment node without modifying the processing logic.

• Easy maintenance: All business lines share the same set of processing logic. A single modification applies to all business lines.