DataWorks：賦值節點

樣本一：傳遞 MaxCompute SQL 查詢結果

樣本二：傳遞 Python 2 輸出結果

Python 2的print語句輸出結果會基於逗號,分割，並作為一個一維數組傳遞給下遊。

• 上遊節點（賦值節點-Python 2）配置

  Python 2代碼如下：

  print 'Electronics, Clothing, Books';

• 下遊節點（Shell節點）配置和輸出

  在Shell節點中添加名為types的輸入參數，並引用上遊賦值節點的outputs。

  編寫如下代碼讀取資料： # 直接輸出整個一維數組 echo "整個結果集: ${types}" # 按索引輸出數組中的元素 echo "第二個元素: ${types[1]}"

  DataWorks將直接解析參數，並做靜態替換。運行輸出如下： 整個結果集: Electronics,Clothing,Books 第二個元素: Clothing

業務背景

假設您是一家綜合性互連網公司的資料開發工程師，負責處理三個核心業務線的資料：電商（ecom）、金融（finance）和物流（logistics），且後續存在增加業務線的可能。您需要每天對這三個業務線的使用者行為日誌執行相同的彙總邏輯，計算每個使用者的日活躍度（PV），並將結果存入統一的匯總表。

• 上遊源表（DWD層）：

  • dwd_user_behavior_ecom_d：電商使用者行為表。

  • dwd_user_behavior_finance_d ：金融使用者行為表。

  • dwd_user_behavior_logistics_d ：物流使用者行為表。

  • dwd_user_behavior_${業務線}_d ：後續更多可能的業務線使用者行為表。

  • 這些表結構相同，都按天分區（dt）。

• 下遊目標表（DWS）：

  • dws_user_summary_d ：使用者匯總表。

  • 該表按業務線（biz_line）和天（dt）雙重分區，用於統一儲存所有業務線的彙總結果。

若為每個業務線建立獨立的任務，維護成本高且容易出錯。使用 for-each-節點後，只需維護一份處理邏輯，系統會自動遍曆所有業務線完成計算。

資料準備

首先，建立樣本表並插入測試資料（以業務日期 20251010 為例）：

1. 為工作空間綁定MaxCompute計算資源。

2. 進入Data Studio資料開發，建立MaxCompute SQL節點。

3. 建立源表（DWD層）：在MaxCompute SQL節點添加如下代碼，並選中運行。

   -- 電商使用者行為表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT '使用者ID',
       action_type STRING COMMENT '行為類型',
       event_time  BIGINT COMMENT '事件發生的毫秒級Unix時間戳記'
   ) 
   COMMENT '電商使用者行為日誌明細表'
   PARTITIONED BY (dt STRING COMMENT '日期分區，格式 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
   -- 驗證電商使用者行為表建立成功
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   
   -- 金融使用者行為表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT '使用者ID',
       action_type STRING COMMENT '行為類型',
       event_time  BIGINT COMMENT '事件發生的毫秒級Unix時間戳記'
   ) 
   COMMENT '金融使用者行為日誌明細表'
   PARTITIONED BY (dt STRING COMMENT '日期分區，格式 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
   -- 驗證金融使用者行為表建立成功
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   
   -- 物流使用者行為表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT '使用者ID',
       action_type STRING COMMENT '行為類型',
       event_time  BIGINT COMMENT '事件發生的毫秒級Unix時間戳記'
   ) 
   COMMENT '物流使用者行為日誌明細表'
   PARTITIONED BY (dt STRING COMMENT '日期分區，格式 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
   
   -- 驗證物流使用者行為表建立成功
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. 建立目標表（DWS層）：在MaxCompute SQL節點上添加如下代碼，並選中運行。

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT '使用者ID',
       pv          BIGINT COMMENT '日活躍度',
   ) 
   COMMENT '使用者日活躍度匯總表'
   PARTITIONED BY (
       dt           STRING COMMENT '日期分區，格式 yyyymmdd',
       biz_line     STRING COMMENT '業務線分區，如 ecom, finance, logistics'
   );

   重要

   若工作空間使用標準環境，需要將此節點發布至生產環境，並執行補資料。

工作流程實現

1. 建立一個工作流程，並在右側調度參數設定調度參數bizdate為上一天$[yyyymmdd-1]。

   

2. 在工作流程中，建立名為get_biz_list的賦值節點，並使用MaxCompute SQL語言編寫如下代碼。該節點輸出需要處理的業務線列表：

   -- 輸出所有需要處理的業務線
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. 配置 for-each-節點

   • 回到工作流程介面，給賦值節點get_biz_list建立一個下遊的for-each節點。

   • 進入for-each節點設定介面，在右側調度配置的調度參數 > 指令碼參數，將 loopDataArray 參數綁定到get_biz_list節點的outputs。

     

   • 在for-each節點迴圈體內單擊建立內部節點，建立 MaxCompute SQL節點，編寫迴圈體內的處理邏輯。

     說明

     • 該指令碼由 for-each-節點驅動，會針對每個業務線執行一次。

     • 內建變數 ${dag.foreach.current} 在每次運行時，會動態地替換為當前的業務線名稱。預期的迭代值為：'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. 添加驗證節點

   回到工作流程，在for-each節點單擊建立下遊，建立一個MaxCompute SQL節點並添加如下代碼。

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

發布和運行結果

將工作流程發布至生產環境，前往營運中心的周期任務營運 > 周期任務，找到目標工作流程並執行煙霧測試 (Smoke Test)，選擇業務日期為'20251010'。

完成運行後，在測試執行個體中查看作業記錄。最終節點預期輸出為：

方案優勢

• 高擴充性：新增業務線時，只需在賦值節點添加一行 SQL，無需修改處理邏輯。

• 易維護：所有業務線共用同一套處理邏輯，修改一處即可生效於全部。