Tous les produits
Search

Ce produit

    DataWorks:Nœud For-each

    Centre de documentation

    DataWorks:Nœud For-each

    Dernière mise à jour :Sep 09, 2026

    Un nœud for-each itère sur un ensemble de résultats en amont, tel qu'une liste de noms de fichiers ou de partitions, et exécute la même sous-tâche pour chaque élément. Cela évite de créer manuellement des tâches individuelles et permet des workflows dynamiques et automatisés.

    Cas d'utilisation

    Un nœud for-each permet une exécution paramétrée lorsque vous devez appliquer la même logique d'analyse ou de traitement à différentes unités commerciales, gammes de produits ou éléments de configuration. Par exemple, si votre entreprise possède plusieurs gammes de produits et que vous devez générer un rapport quotidien distinct pour chacune, la logique de traitement est identique ; seules les données cibles diffèrent.

    À l'instar d'une boucle for dans un langage de programmation, un nœud for-each parcourt une liste, telle que des noms de tables, des noms de partitions ou des noms de fichiers, et exécute un sous-workflow prédéfini pour chaque élément.

    Remarques d'utilisation

    • Exigences de version : Disponible uniquement dans DataWorks Standard Edition et versions ultérieures.

    • Autorisations : Votre compte RAM doit être ajouté au workspace cible et recevoir le rôle developer ou workspace administrator. Pour plus d'informations, consultez la rubrique Ajouter des membres à un workspace.

    Fonctionnement

    Le nœud for-each agit comme un conteneur qui encapsule un sous-workflow personnalisable, appelé corps de boucle. Il fonctionne comme suit :

    image

    1. Entrée des données : Le nœud for-each dépend d'un nœud d'assignation ou d'un autre nœud assignable en amont (tel qu'un nœud EMR Hive). Il récupère l'ensemble de résultats au format tableau en le liant au paramètre loopDataArray.

    2. Exécution de la boucle : Lorsque le nœud démarre, il parcourt chaque élément de l'ensemble de résultats dans l'ordre. Pour chaque élément, il exécute entièrement le corps de boucle interne une fois, du nœud Start au nœud End.

      Remarque

      Les nœuds Start et End ne sont pas modifiables. Ils marquent uniquement le début et la fin du corps de boucle.

    3. Transmission des données : Lors de chaque itération, la valeur de l'élément actuel est transmise aux nœuds situés à l'intérieur du corps de boucle via des variables intégrées. Les nœuds métier internes utilisent ${dag.foreach.current} pour accéder à l'élément de données en cours de traitement.

    Paramètres intégrés

    Important

    Les variables au format ${...} constituent une syntaxe de modèle spécifique à DataWorks. DataWorks analyse directement ces paramètres et les remplace par leurs valeurs avant l'exécution.

    Les nœuds situés dans le corps de la boucle for-each peuvent utiliser les variables intégrées suivantes pour accéder à l'état de la boucle et aux données :

    Paramètre intégré

    Description

    Analogie avec la boucle for

    ${dag.loopDataArray}

    L'ensemble de résultats complet transmis par le nœud d'assignation en amont.

    Considérez le code de boucle for suivant :

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

    • ${dag.loopDataArray} correspond à data.

    • ${dag.foreach.current} correspond à data[i].

    • ${dag.offset} correspond à i.

    • ${dag.loopTimes} correspond à i+1.

    ${dag.foreach.current}

    L'élément de données traité lors de l'itération actuelle.

    ${dag.offset}

    Le décalage actuel de la boucle (indexé à partir de 0).

    ${dag.loopTimes}

    Le compteur actuel de la boucle (indexé à partir de 1).

    Si la sortie en amont est un tableau à deux dimensions, tel qu'un résultat de requête SQL, vous pouvez également utiliser la syntaxe suivante pour accéder à des valeurs spécifiques :

    Autres paramètres

    Description

    ${dag.foreach.current}

    Obtient une chaîne en séparant les éléments de la ligne de données actuelle (un tableau à une dimension) par une virgule ,.

    ${dag.foreach.current[n]}

    Le n-ième élément de la ligne de données actuelle.

    ${dag.loopDataArray[i][j]}

    Les données de la i-ème ligne et de la j-ème colonne de l'ensemble de résultats entier.

    Le nœud for-each ne prend actuellement pas en charge les boucles imbriquées. Cet exemple sert uniquement à démontrer la récupération de valeurs.

    Limitations

    • Mécanisme d'exécution : La boucle prend en charge à la fois l'exécution sérielle et l'exécution parallèle. Choisissez l'exécution parallèle lorsque les itérations sont indépendantes les unes des autres.

    • Limite de boucle : Le nombre maximal de boucles par défaut est de 128, pouvant être ajusté jusqu'à 1024.

    • Contraintes de débogage : Vous ne pouvez pas exécuter directement un nœud for-each dans Data Studio. Vous devez déployer la tâche, puis la tester dans Operation Center en utilisant la fonctionnalité de test de fumée.

    • Contraintes d'exécution : Un nœud for-each ne peut pas être exécuté de manière isolée. Cela inclut les tests de fumée, les opérations de backfill et les exécutions manuelles.

    • Contrôle de flux dans le corps de boucle : Si vous utilisez un nœud de branche à l'intérieur d'un corps de boucle for-each, vous devez vous assurer que toutes les branches convergent vers un seul nœud de fusion avant de se connecter au nœud End. Cela garantit l'intégrité logique du corps de boucle.

    • Contraintes de réexécution : Après le déploiement d'un nœud, une réexécution automatique en cas d'échec reprend à partir du point d'échec. Cependant, une réexécution manuelle déclenche une réexécution complète de l'ensemble du nœud for-each.

    Procédure

    Cette procédure utilise un nœud d'assignation comme nœud en amont et un nœud Shell à l'intérieur du corps de boucle pour afficher les résultats. Elle détaille la configuration d'une tâche for-each complète :

    1. Préparer les données en amont (configurer un nœud d'assignation)

      Créez et configurez un nœud d'assignation pour fournir un ensemble de résultats itérable au nœud for-each en aval.

      1. Dans le workflow, créez un nœud d'assignation (par exemple, assign) et placez-le en amont du nœud for-each.

      2. Double-cliquez sur le nœud d'assignation et sélectionnez un environnement Python 2. Par exemple, utilisez Python 2 pour produire un tableau contenant quatre éléments :

        Le nœud transmet [10,20,30,40] aux nœuds en aval en divisant automatiquement la dernière ligne de sortie en un tableau à chaque virgule.
        print "10,20,30,40"

      3. Le nœud d'assignation génère automatiquement un paramètre de sortie nommé outputs, qui représente son ensemble de résultats.

      4. Enregistrez le nœud d'assignation.

    2. Configurer le nœud for-each pour consommer les données

      Configurez le nœud for-each pour recevoir les données en amont et les utiliser dans son corps de boucle.

      1. Double-cliquez sur le nœud for-each pour ouvrir son canevas interne.

      2. Dans le panneau Scheduling à droite, localisez le paramètre loopDataArray sous Scheduling Parameters et cliquez sur Bind.

        Sélectionnez le paramètre outputs du nœud assign pour créer la liaison. Une fois la liaison terminée, la valeur du paramètre loopDataArray reflète son état lié.

      3. Dans la boîte de dialogue qui s'affiche, définissez la Value Source sur le nœud d'assignation en amont (assign) et sélectionnez son paramètre outputs. Cette action crée automatiquement une dépendance entre les deux nœuds.

      4. Dans le corps de boucle for-each, cliquez sur Create Internal Node et créez un nœud Shell.

        Dans un scénario réel, vous pouvez configurer n'importe quel type de nœud.

      5. Double-cliquez sur le nouveau nœud Shell et utilisez des variables intégrées dans le code pour récupérer et afficher les informations sur la boucle :

        #!/bin/bash
# Use ${dag.loopTimes} to get the current loop count
echo "Current loop number is: ${dag.loopTimes}"
# Use ${dag.foreach.current} to get the data item for the current iteration
echo "Current item is: ${dag.foreach.current}"

      6. (Facultatif) Dans le panneau Scheduling Settings à droite, configurez les propriétés sous Scheduling Policy.

        • Maximum Number of Loops : La valeur par défaut est 128 et le maximum est 1024.

          Important

          Ce paramètre détermine le nombre maximal d'itérations pour le corps de boucle. Si le nombre d'éléments de données en amont est important, augmentez cette valeur pour garantir que tous les éléments soient traités.

        • Execute Policy : Sélectionnez Serial pour cet exemple.

          • Serial : Exécute les itérations séquentiellement.

          • Parallel : Exécute les itérations de boucle simultanément pour améliorer l'efficacité des tâches. En mode Parallel, si une itération échoue, cela n'affecte pas les autres itérations. Le planificateur tente d'exécuter toutes les itérations jusqu'à leur terme. La concurrence par défaut est de 5 et le maximum est de 20.

      7. Enregistrez le nœud Shell.

    3. Déployer, exécuter et vérifier

      Déployez le workflow dans Operation Center pour l'exécution et vérifiez les résultats du nœud for-each.

      1. Revenez au canevas principal du workflow et cliquez sur le bouton Deploy dans la barre d'outils pour publier l'ensemble du workflow.

      2. Accédez à Node O&M > Auto Triggered Task O&M > Auto Triggered Task et effectuez un test de fumée sur le workflow cible.

        Important

        N'effectuez pas de test de fumée sur le nœud for-each individuellement. Étant donné que le nœud for-each dépend de la sortie du nœud d'assignation en amont, vous devez démarrer le test à partir du nœud d'assignation pour garantir que la lignée des données soit complète.

      3. Une fois que l'instance de test s'est exécutée avec succès, recherchez l'instance du nœud for-each dans la liste, ouvrez-la et faites un clic droit pour sélectionner View Internal Nodes.

      4. Dans la vue des nœuds internes, vérifiez les instances de nœud Shell générées par chaque boucle. Ouvrez le journal d'exécution de n'importe quelle instance pour afficher la sortie de cette itération et vérifier que la sortie est correcte.

        Le panneau de gauche indique que les quatre itérations de boucle sont terminées. Le journal d'exécution de la quatrième itération affiche Current loop number is: 4 et Current item is: 40, et la commande Shell se termine avec le code 0, ce qui indique une exécution réussie.

    Remarque

    Outre l'utilisation d'un nœud d'assignation traditionnel comme nœud en amont, un nœud for-each prend également en charge l'obtention du même effet d'itération via la fonctionnalité assignment parameter d'un nœud SQL en amont. Pour les types de nœuds qui prennent en charge les paramètres d'assignation, tels que EMR Hive, Hologres SQL, EMR Spark SQL, AnalyticDB for PostgreSQL, ClickHouse SQL et MySQL, vous pouvez ajouter un paramètre d'assignation dans la section Node Context Parameters > Output Parameters of This Node.

    Cas d'utilisation : Traiter différents formats de données

    Scénario 1 : Traiter un tableau à une dimension

    • Sortie du nœud d'assignation : 2025-11-01,2025-11-02,2025-11-03

    • Nombre d'itérations : 3

    • Lors de la deuxième itération :

      • La valeur de ${dag.foreach.current} est 2025-11-02.

      • La valeur de ${dag.loopTimes} est 2.

    Scénario 2 : Traiter un tableau à deux dimensions

    • Sortie du nœud d'assignation (MaxCompute SQL) :

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

    • Nombre d'itérations : 2

    • Lors de la deuxième itération :

      • La valeur de ${dag.foreach.current} est 102,shanghai.

      • La valeur de ${dag.loopTimes} est 2.

      • La valeur de ${dag.foreach.current[0]} est 102.

      • La valeur de ${dag.foreach.current[1]} est shanghai.

    Scénario : Traitement par lots des données de table de partition sur plusieurs lignes métier

    Cet exemple montre comment utiliser un nœud d'assignation et un nœud for-each pour traiter par lots les données de comportement utilisateur sur plusieurs lignes métier, en automatisant le traitement des données avec une seule logique qui sert plusieurs gammes de produits.

    image

    Contexte

    Supposons que vous soyez ingénieur en développement de données au sein d'une entreprise internet globale, chargé du traitement des 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 la même logique d'agrégation sur les journaux de comportement utilisateur de ces trois lignes métier chaque jour pour calculer les pages vues quotidiennes (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 e-commerce.

      • dwd_user_behavior_finance_d : Table de comportement utilisateur finance.

      • dwd_user_behavior_logistics_d : Table de comportement utilisateur logistique.

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

      • Ces tables ont 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 les résultats agrégés de toutes les lignes métier de manière unifiée.

    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 un seul ensemble de 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 utilisant la date commerciale 20251010 comme exemple).

    1. Associer une ressource de calcul au workspace.

    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 le workspace utilise le mode standard, vous devez déployer ce nœud dans l'environnement de production et effectuer un backfill des données.

    Implémentation du workflow

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

    2. Dans le workflow, créez un nœud d'assignation nommé get_biz_list et écrivez le code suivant dans MaxCompute SQL. Ce nœud produit 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. Configurer le nœud for-each

      • Revenez à la page du workflow et créez un nœud for-each en aval du 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 aux 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 est exécuté 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. Ajouter un nœud de vérification

      Revenez au workflow. 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 workflow dans l'environnement de production. Accédez à la page Auto Triggered Task O&M > Auto Triggered Task dans Operation Center. Recherchez le workflow 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 :

    user_id

    pv

    dt

    biz_line

    user001

    2

    20251010

    ecom

    user002

    1

    20251010

    ecom

    user003

    3

    20251010

    finance

    user004

    1

    20251010

    finance

    user001

    1

    20251010

    logistics

    user005

    1

    20251010

    logistics

    Avantages

    • Haute évolutivité : Pour ajouter une nouvelle ligne métier, ajoutez 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 un même ensemble de logique de traitement. Une seule modification s'applique à toutes.

    FAQ

    • Q : Pourquoi ne puis-je pas exécuter directement un nœud for-each dans Data Studio pour le tester ?

      R : C'est intentionnel. Le nœud nécessite un environnement de planification complet pour résoudre le contexte du nœud et ses dépendances ; il ne prend donc pas en charge l'exécution directe dans Data Studio. Vous devez déployer la tâche dans Operation Center et la tester en utilisant un backfill ou en déclenchant une exécution planifiée.

    • Q : Pourquoi un test de fumée sur un nœud for-each individuel échoue-t-il ou ne fait-il rien ?

      R : Les données de boucle d'un nœud for-each proviennent de son paramètre d'entrée loopDataArray, qui doit être lié au paramètre outputs d'un nœud d'assignation en amont. Si vous exécutez le nœud for-each seul, il échouera ou sera ignoré car il ne peut pas recevoir d'ensemble de résultats en entrée.

    • Q : Pourquoi ma boucle ne s'exécute-t-elle qu'une seule fois ?

      R : Cela se produit généralement parce que la sortie du nœud d'assignation en amont est analysée comme un seul élément. Vérifiez votre sortie :

      • S'agit-il d'une seule chaîne sans délimiteur ?

      • Si vous prévoyez d'itérer sur plusieurs éléments, assurez-vous qu'ils sont séparés par des virgules (,). Par exemple, 'item1,item2,item3' entraîne trois boucles, tandis que 'item1 item2 item3' n'en entraîne qu'une.