DataWorks:Nœud d'assignation

{
  "version": "1.1.0",
  "kind": "Node",
  "spec": {
    "nodes": [
      {
        "recurrence": "Normal",
        "script": {
          "runtime": {
            "command": "CONTROLLER_ASSIGNMENT"
          },
          "content": "print '10,20,30'"
        },
        "runtimeResource": {
          "resourceGroup": "S_res_group_XXX_XXXX"
        },
        "name": "assignment_node_demo"
      }
    ]
  }
}

Exemple 1 : Transmettre les résultats de requête MaxCompute SQL

Exemple 2 : Transmettre les résultats de sortie Python 2

La sortie de l'instruction print de Python 2 est divisée par des virgules (,) et transmise aux nœuds en aval sous forme de tableau unidimensionnel.

• Configuration du nœud en amont (nœud d'assignation - Python 2)

  Le code Python 2 est le suivant :

  print 'Electronics, Clothing, Books';

• Configuration et sortie du nœud en aval (nœud Shell)

  Dans le nœud Shell, ajoutez un paramètre d'entrée nommé types et référencez le paramètre outputs du nœud d'assignation en amont.

  Écrivez le code suivant pour lire les données : # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  DataWorks analyse directement le paramètre et effectue une substitution statique. La sortie est la suivante : Entire result set: Electronics,Clothing,Books Second element: Clothing

Contexte

Supposons que vous soyez ingénieur en développement de données au sein d'une entreprise Internet polyvalente, chargé de traiter les données de trois lignes métier principales : commerce électronique (ecom), finance (finance) et logistique (logistics), avec la possibilité d'en ajouter d'autres à l'avenir. Vous devez exécuter chaque jour la même logique d'agrégation sur les journaux de comportement utilisateur de ces trois lignes métier afin de calculer le nombre quotidien de pages vues (PV) par utilisateur et stocker les résultats dans une table d'agrégation unifiée.

• Tables sources en amont (couche DWD) :

  • dwd_user_behavior_ecom_d : Table de comportement utilisateur du commerce électronique.

  • dwd_user_behavior_finance_d : Table de comportement utilisateur de la finance.

  • dwd_user_behavior_logistics_d : Table de comportement utilisateur de la logistique.

  • dwd_user_behavior_${business_line}_d : Tables de comportement utilisateur pour d'autres lignes métier potentielles à l'avenir.

  • Ces tables partagent le même schéma et sont partitionnées par jour (dt).

• Table cible en aval (couche DWS) :

  • dws_user_summary_d : Table d'agrégation utilisateur.

  • Cette table est doublement partitionnée par ligne métier (biz_line) et par jour (dt) pour stocker de manière unifiée les résultats agrégés de toutes les lignes métier.

La création d'une tâche distincte pour chaque ligne métier entraîne des coûts de maintenance élevés et est source d'erreurs. Avec un nœud for-each, vous maintenez une seule logique de traitement et le système itère automatiquement sur toutes les lignes métier pour effectuer le calcul.

Préparation des données

Commencez par créer les tables d'exemple et insérer les données de test (en prenant la date commerciale 20251010 comme exemple).

1. Associez une ressource de calcul à l'espace de travail.

2. Accédez à Data Studio pour le développement de données et créez un nœud MaxCompute SQL.

3. Créez les tables sources (couche DWD) : Ajoutez le code suivant au nœud MaxCompute SQL et exécutez-le.

   -- 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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'E-commerce user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format 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
   -- Verify e-commerce user behavior table created successfully
   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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Finance user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format 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
   -- Verify finance user behavior table created successfully
   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 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Logistics user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format 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
   -- Verify logistics user behavior table created successfully
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Créez la table cible (couche DWS) : Ajoutez le code suivant au nœud MaxCompute SQL et exécutez-le.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily activity count'
   ) 
   COMMENT 'User daily activity summary table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition, format yyyymmdd',
       biz_line     STRING COMMENT 'Business line partition, e.g. ecom, finance, logistics'
   );

   Important

   Si l'espace de travail utilise le mode standard, vous devez déployer ce nœud dans l'environnement de production et remplir les données rétrospectives.

Implémentation du flux de travail

1. Créez un flux de travail. Dans la section Scheduling Parameters située à droite, définissez le paramètre de planification bizdate sur la veille : $[yyyymmdd-1].

2. Dans le flux de travail, créez un nœud d'assignation nommé get_biz_list et écrivez le code suivant en MaxCompute SQL. Ce nœud génère la liste des lignes métier à traiter :

   -- Output all business lines to be processed
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. Configurez le nœud for-each

   • Revenez à la page du flux de travail et créez un nœud for-each en aval pour le nœud d'assignation get_biz_list.

   • Ouvrez la page des paramètres du nœud for-each. Dans la section Scheduling Parameters > Script Parameters sous schedule settings à droite, liez le paramètre loopDataArray à la sortie outputs du nœud get_biz_list.

   • Dans le corps de boucle du nœud for-each, cliquez sur Create Internal Node et créez un nœud MaxCompute SQL. Écrivez la logique de traitement à l'intérieur du corps de boucle.

     Remarque

     • Ce script est piloté par le nœud for-each et s'exécute une fois pour chaque ligne métier.

     • La variable intégrée ${dag.foreach.current} est remplacée dynamiquement par le nom de la ligne métier actuelle à chaque itération. Les valeurs d'itération attendues sont : '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. Ajoutez un nœud de vérification

   Revenez au flux de travail. Cliquez sur Create Downstream sur le nœud for-each pour créer un nœud MaxCompute SQL et ajoutez le code suivant.

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

Déploiement et résultats

Déployez le flux de travail dans l'environnement de production. Accédez à la page Auto Triggered Task O&M > Auto Triggered Task dans Operation Center. Recherchez le flux de travail cible et effectuez un test de fumée en définissant la date commerciale sur '20251010'.

Une fois l'exécution terminée, consultez le journal d'exécution dans l'instance de test. La sortie attendue du nœud final est la suivante :

Avantages

• Haute évolutivité : Pour ajouter une nouvelle ligne métier, il suffit d'ajouter une ligne de SQL dans le nœud d'assignation sans modifier la logique de traitement.

• Maintenance simplifiée : Toutes les lignes métier partagent la même logique de traitement. Une modification unique s'applique à toutes.