DataWorks:割り当てノード

例 1：MaxCompute SQL クエリ結果の受け渡し

例 2：Python 2 出力結果の受け渡し

Python 2 の print 文の出力は、カンマ (,) で区切られ、1 次元の配列として下流ノードに渡されます。

• 上流ノード (割り当てノード - Python 2) の設定

  Python 2 のコードは次のとおりです。

  print 'Electronics, Clothing, Books';

• 下流ノード (シェルノード) の設定と出力

  Shell ノードで、types という名前の入力パラメーターを追加し、上流の割り当てノードの outputs パラメーターを参照します。

  次のコードを記述してデータを読み取ります。 # 一次元配列全体を直接出力 echo "Entire result set: ${types}" # インデックスで要素を出力 echo "Second element: ${types[1]}"

  DataWorks はパラメーターを直接解析し、静的な置換を実行します。出力は次のようになります。 Entire result set: Electronics, Clothing, Books Second element: Clothing

背景情報

ある総合インターネット企業でデータ開発エンジニアであると仮定します。あなたは3つの主要ビジネスライン (eコマース (ecom)、金融 (finance)、物流 (logistics)) のデータを処理しており、将来的にはさらに追加される可能性があります。毎日、これら 3 つのビジネスラインのユーザー行動ログに対して同じ集計ロジックを実行し、ユーザーごとの日次ページビュー (PV) を計算し、その結果を統一された集計テーブルに保存する必要があります。

• 上流のソーステーブル (DWD レイヤー)：

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

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

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

  • dwd_user_behavior_${business_line}_d：将来、さらにビジネスラインが追加される可能性に対応するユーザー行動テーブル。

  • これらのテーブルはスキーマが同じで、日単位 (dt) でパーティション化されています。

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

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

  • このテーブルは、すべてのビジネスラインからの集計結果を一元的に保存するために、ビジネスライン(biz_line)と日(dt)で二重にパーティション化されています。

ビジネスラインごとに個別のタスクを作成すると、メンテナンスコストが高くなり、エラーが発生しやすくなります。for-each ノードを使用すると、単一の処理ロジックを維持するだけで、システムが自動的にすべてのビジネスラインを反復処理して計算を完了します。

データ準備

まず、サンプルテーブルを作成し、テストデータを挿入します (業務日 20251010 を例として使用)。

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

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 '日次アクティビティ数'
   ) 
   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 ノードのループ本体で、[Create Inner Node] をクリックし、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 ノードで [Create Downstream] をクリックして MaxCompute SQL ノードを作成し、次のコードを追加します。

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

デプロイと結果

ワークフローを本番環境にデプロイします。運用センターの 定期タスクの運用保守 > 定期タスク ページに移動し、対象のワークフローを見つけ、ビジネス日付を '20251010' に設定してスモークテストを実行します。

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

メリット

• 高いスケーラビリティ：新しいビジネスラインを追加するには、処理ロジックを変更することなく、割り当てノードに SQL を 1 行追加するだけです。

• 容易なメンテナンス：すべてのビジネスラインが 1 つの処理ロジックを共有します。1 回の変更がすべてに適用されます。