DataWorks:代入ノード

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

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

Python 2 の print 文の出力は、カンマ (,) で分割され、1 次元配列として渡されます。

• 代入ノード

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

  print 'Electronics, Clothing, Books';

• Shell ノード

  代入ノードの outputs を参照する param という名前の入力パラメーターを追加します。次のスクリプトを使用してデータを読み取ります。

  次のスクリプトを使用してデータを読み取ります。 # 配列全体を出力 echo "Full result set: ${types}" # インデックスで特定の要素を出力 echo "Second element: ${types[1]}"

  DataWorks は静的な置き換えを実行します。出力： Full result set: Electronics,Clothing,Books Second element: Clothing

背景情報

ある大手インターネット企業のデータ開発者であると仮定します。あなたは、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. DataStudio に移動して、MaxCompute SQL ノードを作成します。

3. ソーステーブルの作成：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. 宛先テーブルの作成：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 ノードの設定

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

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

     

   • for-each ノードのループ本文で、[Create Internal 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 ノードで、[子孫ノードを作成] をクリックします。MaxCompute SQL ノードを作成し、次のコードを追加します。

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

デプロイと実行

ワークフローを本番環境にデプロイします。オペレーションセンターで、[自動トリガーノード運用保守] > [自動トリガーノード] に移動し、対象のワークフローを見つけ、テストを実行し、データタイムスタンプとして '20251010' を選択します。

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

メリット

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

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