DataWorks:for-eachノード

組み込み変数

for-eachループボディ内で、以下の組み込み変数を使用してループステータスとデータにアクセスします。

for(int i=0;i<data.length;i++) {
   print(data[i]);
}

アップストリームの出力が2次元配列 (SQLクエリ結果など) の場合、以下の構文を使用して正確な値を取得することもできます。

シナリオ 1: 1 次元配列の処理 (Shell/Python 出力)

• 代入ノード出力: 2025-11-01,2025-11-02,2025-11-03

• 反復回数: 3

• 2回目の反復で:

  • ${dag.foreach.current}の値は2025-11-02です。

  • ${dag.loopTimes}の値は2です。

シナリオ 2: 2 次元配列の処理 (SQL 出力)

• 代入ノード (MaxCompute SQL) 出力:

  +-----+----------+
  | id  | city     |
  +-----+----------+
  | 101 | beijing  |
  | 102 | shanghai |
  +-----+----------+

• 反復回数: 2

• 2回目の反復で:

  • ${dag.foreach.current}の値は102,shanghaiです。

  • ${dag.loopTimes}の値は2です。

  • ${dag.foreach.current[0]}の値は102です。

  • ${dag.foreach.current[1]}の値はshanghaiです。

背景情報

あなたは大手インターネット企業のデータ開発者であると仮定します。あなたは、E コマース、金融、物流という3つの主要事業部門のデータ処理を担当しています。今後、さらに多くの事業部門が追加される可能性があります。これら3つの事業部門のユーザー行動ログに対して、毎日同じ集約ロジックを実行する必要があります。このロジックは、各ユーザーのPVを計算し、結果を統合集計テーブルに保存します。

• アップストリームソーステーブル (DWD レイヤー):

  • dwd_user_behavior_ecom_d: E コマースユーザー行動テーブル。

  • dwd_user_behavior_finance_d: 金融ユーザー行動テーブル。

  • dwd_user_behavior_logistics_d: 物流ユーザー行動テーブル。

  • dwd_user_behavior_${line-of-business}_d: その他の潜在的な事業部門のユーザー行動テーブル。

  • これらのテーブルは同じスキーマを持ち、日 (dt) ごとにパーティション化されています。

• ダウンストリーム宛先テーブル (DWS レイヤー):

  • dws_user_summary_d: ユーザー集計テーブル。

  • このテーブルは、事業部門 (biz_line) と日 (dt) の両方でパーティション化されています。すべての事業部門の集計結果を保存するために使用されます。

各事業部門に対して個別のタスクを作成することは、維持コストが高く、エラーが発生しやすいです。for-eachノードを使用すると、1つの処理ロジックセットを維持するだけで済みます。システムはすべての事業部門を自動的に走査して計算を完了します。

データ準備

まず、サンプルテーブルを作成し、テストデータを挿入します。この例では、データタイムスタンプ20251010を使用します。

1. ワークスペースにMaxComputeコンピューティングリソースを関連付けます。

2. Data Studioに移動し、MaxCompute SQLノードを作成します。

3. ソーステーブルの作成: 以下のコードをMaxCompute SQLノードに追加し、選択して実行します。

   -- 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. 宛先テーブルの作成: 以下のコードをMaxCompute SQLノードに追加し、選択して実行します。

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

   重要

   ワークスペースが標準モードを使用している場合、このノードを本番環境にデプロイし、データバックフィルを実行する必要があります。

ワークフローの実装

1. ワークフローを作成します。右側の [スケジューリング] ペインで、スケジューリングパラメーター bizdate を前日の $[yyyymmdd-1] に設定します。

   

2. ワークフローで、get_biz_listという名前の代入ノードを作成します。MaxCompute SQLで以下のコードを記述します。このノードは、処理する事業部門のリストを出力します。

   -- 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. for-eachノードの設定

   • ワークフローキャンバスに戻り、代入ノードのダウンストリームにfor-eachノードを作成します。

   • for-each ノードの設定ページに移動します。右側の [スケジューリング] タブで、[スケジューリングパラメーター] > [スクリプトパラメーター] の下で、loopDataArray パラメーターの値を get_biz_list ノードの出力に設定します。

     

   • for-each ノードのループ本文で、[内部ノードの作成] をクリックします。 MaxCompute SQL ノードを作成し、ループ本文の処理ロジックを記述します。

     説明

     • このスクリプトはfor-eachノードによって駆動され、各事業部門に対して1回実行されます。

     • 組み込み変数${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;

デプロイと実行

ワークフローを本番環境にデプロイします。オペレーションセンターで、自動トリガーノードO&M > 自動トリガーノードに移動し、ターゲットワークフローを見つけてテストを実行し、データタイムスタンプとして'20251010'を選択します。

実行完了後、テストインスタンスで実行ログを表示します。最終ノードの期待される出力は以下のとおりです。

利点

• 高い拡張性: 新しい事業部門が追加された場合、代入ノードにSQLコードを1行追加するだけで済みます。処理ロジックを変更する必要はありません。

• 容易なメンテナンス: すべての事業部門が同じ処理ロジックを共有します。1箇所の変更がすべてに適用されます。