DataWorks:Assignment node

Example 1: Pass a MaxCompute SQL query result

Example 2: Pass a Python 2 output result

The system splits the output of a Python 2 print statement by commas (,) and passes it to the downstream node as a one-dimensional array.

• Upstream node (Python 2 assignment node) configuration

  The Python 2 code is as follows:

  print 'Electronics, Clothing, Books';

• Downstream node (Shell node) configuration and output

  Add an input parameter named types to the Shell node and reference the outputs of the upstream assignment node.

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

  The code returns the following output: Full result set: Electronics,Clothing,Books Second element: Clothing

Background

Assume you are a data engineer at a large company responsible for processing data from three core lines of business: e-commerce (ecom), finance (finance), and logistics (logistics), with the possibility of adding more in the future. You need to apply the same aggregation logic to the user behavior logs of these lines of business every day to calculate the daily page views (PV) for each user and store the results in a unified summary 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 destination table (DWS layer):

  • dws_user_summary_d : User summary table.

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

Using a for-each node, you only need to maintain one set of processing logic, and the system automatically iterates through all lines of business to complete the calculations.

Data preparation

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

1. Associate a MaxCompute compute resource with the workspace.

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

3. Create the source tables (DWD layer). In the MaxCompute SQL node, add the following code, 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 'Event time as a millisecond-level Unix timestamp'
   ) 
   COMMENT 'E-commerce user behavior log detail table'
   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 'Event time as a millisecond-level Unix timestamp'
   ) 
   COMMENT 'Finance user behavior log detail table'
   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 'Event time as a millisecond-level Unix timestamp'
   ) 
   COMMENT 'Logistics user behavior log detail table'
   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 destination table (DWS layer). In the MaxCompute SQL node, add the following code, select it, and run it.

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

   Important

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

Workflow implementation

1. Create a workflow. In the Scheduling Parameters panel 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. Use the MaxCompute SQL language to write the following code. This node outputs a list of the lines of business to be processed.

   -- 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 for-each node settings panel. In the Scheduling panel on the right, under the Scheduling Parameters > Script Parameters tab, bind the loopDataArray parameter to the outputs of the get_biz_list node.

   • In the for-each node loop body, click Create Internal Node to create a MaxCompute SQL node, and then 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.

     • At runtime, the system dynamically replaces the built-in variable ${dag.foreach.current} with the current line of business name. 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 to create a MaxCompute SQL node, and then 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 Auto Triggered Task O&M > Auto Triggered Task O&M > Auto Triggered Task, find the workflow, and run a smoke test. Select '20251010' as the business date.

After the run completes, the final node produces the following output:

Benefits

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

• Easy maintenance: All lines of business share the same processing logic. A change in one place applies to all.