: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 コマース (ecom)、金融 (finance)、物流 (logistics) という3つのコア LOB のデータ処理を担当しています。将来的にはさらに多くの LOB が追加される可能性があります。毎日、これら3つの LOB のユーザー行動ログに対して同じ集約ロジックを実行する必要があります。このロジックは、各ユーザーの1日あたりの人気度 (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：他の潜在的な LOB のユーザー行動テーブル。

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

• 下流ターゲットテーブル (DWS レイヤー)：

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

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

LOB ごとに個別のタスクを作成すると、メンテナンスコストが高くなり、エラーが発生しやすくなります。for-each ノードを使用すれば、1セットの処理ロジックを維持するだけで済みます。システムは自動的にすべての LOB を走査して計算を完了します。

データ準備

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

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

2. DataStudio に移動してデータ開発を行い、MaxCompute SQL ノードを作成します。

3. ソーステーブル (DWD レイヤー) を作成します：MaxCompute SQL ノードに次のコードを追加し、それを選択して実行します。

   -- E コマースユーザー行動テーブル
   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 'E コマースユーザー行動ログの詳細'
   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
   -- E コマースユーザー行動テーブルが作成されたことを確認します。
   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 '1日あたりの人気度',
   ) 
   COMMENT '1日あたりのユーザー人気度集計テーブル'
   PARTITIONED BY (
       dt           STRING COMMENT '日付パーティション (yyyymmdd 形式)',
       biz_line     STRING COMMENT 'LOB パーティション (ecom, finance, logistics など)'
   );

   重要

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

ワークフローの実装

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

   

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

   -- 処理するすべての LOB を出力します
   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 ノードによって駆動され、LOB ごとに1回実行されます。

     • 組み込み変数 ${dag.foreach.current} は、実行時に現在の LOB 名に動的に置き換えられます。期待される反復値は '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' を選択します。

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

このソリューションの利点

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

• 容易なメンテナンス：すべての LOB が同じ処理ロジックを共有します。1か所の変更がすべてに反映されます。