DataWorks:Assignment node

Step 1: Configure the assignment node

In your workflow, create an assignment node. Select MaxCompute SQL, Python 2, or Shell, then write the script.

Step 2: Configure the downstream node

The following steps use a Shell node as the downstream node.

1. Create a Shell node.

2. In the Scheduling panel on the right, select the Input and Output Parameters tab.

3. In the Input Parameters section, click Create Parameter.

4. Set the parameter name (for example, param) and set the value to the outputs parameter from the assignment node.

   DataWorks automatically creates a scheduling dependency between the Shell node and the assignment node after this configuration.

5. Reference the passed value in your Shell script using ${param}.

Step 3: Deploy and verify

1. On the workflow canvas, click Deploy in the toolbar and select Full Deployment.

2. Go to Operation Center > Auto Triggered Node O&M > Auto Triggered Nodes.

3. Run a smoke test on the workflow and verify the results.

Pass a MaxCompute SQL query result

The result of a SQL query is passed to downstream nodes as a two-dimensional array.

Assignment node

The following SQL returns two rows and two columns:

SELECT 'beijing', '1001'
UNION ALL
SELECT 'hangzhou', '1002';

Shell node

Add an input parameter named param (or any name you choose) that references the assignment node's outputs. DataWorks performs static replacement at runtime.

Pass a Python 2 output

The last print statement is split by commas into a one-dimensional array.

Assignment node

print 'Electronics, Clothing, Books';

Shell node

Add an input parameter named param that references outputs. DataWorks performs static replacement at runtime.

Shell nodes follow the same processing logic as Python 2 nodes.

Background

You are a data developer responsible for processing user behavioral data from three lines of business: e-commerce, finance, and logistics. The same aggregation logic (calculating daily page views per user) runs every day, and more lines of business may be added in the future.

Source tables (DWD layer):

• dwd_user_behavior_ecom_d: E-commerce user behavior, partitioned by day (dt)

• dwd_user_behavior_finance_d: Finance user behavior, partitioned by day (dt)

• dwd_user_behavior_logistics_d: Logistics user behavior, partitioned by day (dt)

• 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).

Destination table (DWS layer):

• dws_user_summary_d: Aggregate table, partitioned by biz_line and dt

Creating a separate task for each line of business is hard to maintain. By combining an assignment node and a for-each node, you maintain one set of processing logic that runs automatically for each line of business.

Data preparation

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: add the following code to the 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 destination table: add the following code to the node, select it, and run it.

   Important

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

   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'
   );

Workflow implementation

1. Create a workflow. In the Scheduling 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 MaxCompute SQL code. This node outputs the 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 for-each node downstream of the assignment node.

   • Go to the for-each node settings. On the Scheduling tab, under Scheduling Parameters > Script Parameters, set loopDataArray to the outputs of the get_biz_list node.

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

     Note

     This script runs once for each line of business. The built-in variable ${dag.foreach.current} is replaced with the current line-of-business name at runtime (ecom, finance, or 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. Create a downstream MaxCompute SQL node for the for-each node and add the following code.

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

Deploy and run

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

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

Advantages

• High extensibility: To add a new line of business, add one line of SQL in the assignment node. No changes to the processing logic are needed.

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