DataWorks:Flux de travail déclenché

Étape 1 : Créer un flux de travail déclenché

1. Accédez à la page Workspaces dans la console DataWorks. Dans la barre de navigation supérieure, sélectionnez la région souhaitée. Recherchez l'espace de travail cible et choisissez Shortcuts > Data Studio dans la colonne Actions.

2. Dans le volet de navigation de gauche, cliquez sur , puis cliquez sur > Create Workflow à droite de Project Directory pour accéder à la page Create Workflow.

3. Dans la boîte de dialogue qui s'affiche, sur la page Create Workflow, définissez le paramètre Scheduling Type sur Triggered scheduling. Saisissez un Name pour le flux de travail et cliquez sur Confirm pour créer le flux de travail.

Étape 2 : Orchestrer le flux de travail et développer les nœuds

1. Cliquez sur + Add Node dans la barre d'outils pour ouvrir la liste des nœuds. Faites glisser un nœud Shell depuis la liste des types de nœuds sur la gauche vers le canevas, saisissez un nom et finalisez la création.

2. Double-cliquez sur le nœud Shell pour accéder à la page d'édition du code et saisissez le code suivant :

   echo "Hello, Trigger Workflow! Current time is ${bizdate}"

3. Cliquez sur le bouton Save dans la barre d'outils.

Étape 3 : Déboguer et exécuter (environnement de développement)

1. Revenez au canevas du flux de travail et cliquez sur l'icône dans la barre d'outils supérieure.

2. Dans la boîte de dialogue qui s'affiche, saisissez la Value Used in This Run pour le flux de travail (par exemple, si nous sommes le 10/03/2026, bizdate doit être remplacé par 20260309).

3. Après quelques instants, consultez le journal d'exécution en bas pour voir le statut d'exécution du nœud et la sortie de la commande echo.

Étape 4 : Déployer et exécuter (environnement de production)

1. Sur le canevas du flux de travail, cliquez sur le bouton Deploy et suivez les instructions pour terminer le processus de déploiement.

2. Une fois le déploiement réussi, accédez à Operation Center > Manually Triggered Task O&M > Manually Triggered Task> Triggered Workflow.

3. Recherchez le flux de travail que vous venez de déployer et cliquez sur Run dans la colonne Operation.

4. Dans la boîte de dialogue qui s'affiche, cliquez à nouveau sur Run pour déclencher une instance du flux de travail dans l'environnement de production. Consultez les détails de cette exécution sur la page Manual Instances.

Vous maîtrisez désormais l'utilisation de base des flux de travail déclenchés. Nous allons ensuite explorer les fonctionnalités plus avancées du déclenchement par événement.

Étape 1 : Créer un déclencheur OSS

1. Accédez à Operation Center > Scheduling Settings > Trigger Management.

2. Cliquez sur Create Trigger et configurez les paramètres comme suit :

   Remarque

   Pour une description détaillée des paramètres, consultez la rubrique OSS trigger.

   • Trigger Name : Saisissez un nom personnalisé, par exemple oss_new_file_trigger.

   • Applicable Workspace : Sélectionnez l'espace de travail cible où réside le flux de travail.

   • Trigger Event Type : Sélectionnez Object Storage Service (OSS).

   • Trigger Event : Sélectionnez oss:ObjectCreated:PutObject (ou un autre événement de téléchargement).

   • Bucket Name : Sélectionnez votre compartiment OSS.

   • File Name : Spécifiez le chemin et le format des fichiers à surveiller. Les caractères génériques sont pris en charge. Par exemple, pour surveiller tous les fichiers .csv dans le répertoire input/, saisissez input/*.csv.

   • Configuration du rôle : Pour une première utilisation, effectuez une One-Click Authorization et sélectionnez le rôle nommé DataWorks-EventBridge-OSS-MNS-Role-***.

     *** représente un numéro ID aléatoire de 13 chiffres utilisé pour garantir l'unicité.

3. Cliquez sur Confirm pour terminer la création du déclencheur.

Étape 2 : Créer et associer un flux de travail

1. Suivez les étapes décrites dans la section Démarrage rapide : Créer un flux de travail déclenché manuellement pour créer un nouveau flux de travail déclenché nommé process_oss_file_workflow.

2. Dans le panneau de droite du canevas du flux de travail, accédez à Scheduling Settings > Scheduling Policy.

3. Dans la liste déroulante Trigger, sélectionnez le déclencheur oss_new_file_trigger que vous venez de créer.

   Après avoir sélectionné un déclencheur, utilisez ${workflow.triggerMessage} dans les tâches internes pour obtenir le corps complet du message, ou ${workflow.triggerMessage.xxx} pour obtenir la valeur d'un champ spécifique du corps du message.

Étape 3 : Développer un nœud et analyser les paramètres d'événement

1. Cliquez sur + Add Node dans la barre d'outils pour ouvrir la liste des nœuds. Faites glisser un nœud Shell depuis la liste des types de nœuds sur la gauche vers le canevas, saisissez un nom et finalisez la création.

2. Double-cliquez sur le nœud et écrivez du code pour récupérer et afficher le chemin du fichier à partir de l'événement déclencheur.

   1. Utilisation générale : L'utilisation des paramètres suivante s'applique à la plupart des messages d'événement.

      # When a trigger fires the workflow, event info is passed in via the built-in variable workflow.triggerMessage
      # We can get the full path of the uploaded file via ${workflow.triggerMessage.data.oss.object.key}
      echo "========= Start Processing OSS File ========="
      message='${workflow.triggerMessage}'
      echo "Raw Value: ${message}"
      # Extract the file name from the event message
      FILE_PATH='${workflow.triggerMessage.data.oss.object.key}'
      echo "A new file has arrived: ${FILE_PATH}"
      # Add specific processing logic here
      echo "========= Finish Processing OSS File ========="

   2. Cas particulier : Certains nœuds, tels que les nœuds Shell, possèdent leurs propres règles de syntaxe. Avant l'exécution du nœud, le moteur de planification substitue d'abord le contenu du message dans le texte du script, puis le transmet à l'interpréteur Shell pour exécution. Par conséquent, les guillemets, les espaces et les sauts de ligne du message participent directement à l'analyse syntaxique Shell. Cela peut entraîner une rupture de la structure originale du code après la récupération du paramètre par le nœud, en raison des espaces ou des sauts de ligne contenus dans le paramètre. Dans ce cas, utilisez un heredoc pour lire le message, ce qui évite la plupart des problèmes liés aux caractères spéciaux :

      # Use a heredoc to safely capture the raw text substituted from ${workflow.triggerMessage}.
      message=$(cat <<'__DW_TRIGGER_MESSAGE__'
      ${workflow.triggerMessage}
      __DW_TRIGGER_MESSAGE__
      )
      
      # Scenario 1: Get the entire raw message
      printf 'Raw Value: %s\n' "$message"
      
      # Scenario 2: If you need only a specific field, use Python to parse the JSON and extract the value
      dt=$(printf '%s' "$message" | /home/tops/bin/python3 -c 'import json,sys; print(json.load(sys.stdin).get("body",{}).get("dt",""))')
      printf 'dt=%s\n' "$dt"

      Remarque

      ${workflow.triggerMessage} : Récupère le corps complet du message d'événement au format JSON. Vous pouvez obtenir le format de message spécifique pour OSS depuis EventBridge > Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

      Afficher un exemple de format de message OSS, qui correspond au format de workflow.triggerMessage.

      {
          "datacontenttype": "application/json;charset=utf-8",
          "aliyunaccountid": "1***********9",
          "data": {
              "eventVersion": "1.0",
              "responseElements": {
                  "requestId": "69B1***********C0A8"
              },
              "eventSource": "acs:oss",
              "eventTime": "2026-03-11T05:40:45.000Z",
              "requestParameters": {
                  "sourceIPAddress": "***********"
              },
              "eventName": "ObjectCreated:PostObject",
              "userIdentity": {
                  "principalId": "1***********9"
              },
              "region": "cn-hangzhou",
              "oss": {
                  "bucket": {
                      "name": "******",
                      "arn": "acs:oss:cn-hangzhou:1***********9:******",
                      "virtualBucket": "",
                      "ownerIdentity": "1***********9"
                  },
                  "ossSchemaVersion": "1.0",
                  "object": {
                      "size": 59537,
                      "objectMeta": {
                          "mimeType": "text/csv"
                      },
                      "deltaSize": 0,
                      "eTag": "63***********D32",
                      "key": "input/***********.csv"
                  }
              }
          },
          "subject": "acs:oss:cn-hangzhou:1***********9:dwoss1024/input/******.csv",
          "aliyunoriginalaccountid": "1***********9",
          "source": "acs.oss",
          "type": "oss:ObjectCreated:PostObject",
          "aliyunpublishtime": "2026-03-11T05:40:45.682Z",
          "specversion": "1.0",
          "aliyuneventbusname": "DATAWORKS_TRIGGER_FOR_BUCKET_******",
          "id": "69B1***********0A8",
          "time": "2026-03-11T05:40:45.000Z",
          "aliyunregionid": "cn-hangzhou"
      }

Étape 4 : Déboguer et publier

1. Débogage :

   • Revenez au canevas du flux de travail et cliquez sur le bouton Run .

   • Dans la zone de saisie Trigger Message Body , collez un JSON d'événement OSS simulé. Copiez et modifiez la valeur key à partir de l'« Exemple de format de message » sur la page de configuration du déclencheur. Voici un exemple simple.

     {
       "data": {
         "oss":{
           "object": {
             "key": "input/test_file_20260310.csv" 
           }
         } 
       }
     }

   • Cliquez sur Run et vérifiez si input/test_file_20260310.csv s'affiche correctement dans les journaux.

2. Publication : Une fois le débogage réussi, cliquez sur le bouton Publish pour déployer le flux de travail dans l'environnement de production. Les déclencheurs d'événements ne prennent effet que dans l'environnement de production.

Étape 5 : Vérifier en production

1. Utilisez la console OSS ou un outil client pour télécharger un fichier CSV dans le compartiment et le chemin (par exemple, le répertoire input/) que vous avez configurés dans le déclencheur.

   Comment vérifier que le déclencheur d'événement fonctionne ?

   Accédez à https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing pour interroger la liste des événements récemment déclenchés. Cliquez également sur Event Details pour afficher le message de déclenchement spécifique (c'est-à-dire workflow.triggerMessage).

   La liste des événements affiche les enregistrements de déclenchement avec la source d'événement Object Storage Service (OSS) / acs.oss et le type d'événement oss:ObjectCreated:PostObject.

2. Accédez à Operation Center de DataWorks > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow. Le flux de travail process_oss_file_workflow publié avec succès apparaît.

3. Attendez quelques instants, puis accédez à Operation Center de DataWorks > Manually Triggered Task O&M > Triggered Workflow Instance. Une nouvelle instance de flux de travail est automatiquement déclenchée. Cliquez dessus pour afficher ses journaux et vérifier que le chemin du fichier est correctement traité.

   ===== Start Processing 0SS File =====
   Raw Value: {"datacontenttype":"application/json;charset=utf-8","aliyunaccountid":"1162423445433459","data":{"eventVersion":"1.0","responseElements":{"requestId":"69B1083F7A439F343040ABCD"},
   "eventSource":"acs:oss","eventTime":"2026-03-11T06:14:23.000Z","requestParameters":{"sourceIPAddress":"140.205.11.13"},"eventName":"0bjectCreated:Post0bject","userIdentity":{"principalId":"1162423445433459"},
   "region":"cn-hangzhou","oss":{"bucket":{"name":"dwoss1024","arn":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024","virtualBucket":"","ownerIdentity":"1162423445433459"},"ossSchemaVersion":"1.0","object":{"size":59537,"objectMeta":
   {"mimeType":"text/csv"},"deltaSize":0,"eTag":"63B4BA5A45AEFC679B9A917E8DDF0D32","key":"input/2013-2020-global-PS4-game-sales.csv"}}},"subject":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024/input/2013-2020-global-PS4-game-sales.csv","aliyunoriginalaccountid":"1162423445433459","source":"acs.oss","type":"oss:0bjectCreated:Post0bject","aliyunpublishtime":"2026-03-11T06:14:23.959Z","specversion":"1.0","aliyuneventbusname":"DATAW0RKS_TRIGGER_F0R_BUCKET_dwoss1024","id":"69B1083F7A439F343040ABCD","time":"2026-03-11T06:14:23.000Z","aliyunregionid":"cn-hangzhou"}A new file has arrived: input/2013-2020-global-PS4-game-sales.csv
   ====== Finish Processing 0SS File =========

Scénario 2 : L'arrivée d'un message Kafka pilote le calcul en temps réel

Objectif : Surveiller les journaux de comportement des utilisateurs dans Kafka. Lorsque de nouveaux messages arrivent, déclencher un flux de travail pour les analyser et exécuter différentes logiques en fonction du contenu.

Orchestration du flux de travail

Le processus d'orchestration principal des flux de travail déclenchés est similaire à celui des flux de travail planifiés. Pour plus d'informations, consultez la rubrique Node/workflow orchestration.

Paramètres de planification

Dans le panneau Scheduling Settings situé à droite du canevas du flux de travail, vous pouvez définir des paramètres globaux pour le flux de travail. Tous les nœuds du flux de travail peuvent référencer ces paramètres.

• Méthode de référence : Dans le code du nœud, référencez les paramètres du flux de travail au format ${workflow.parameter_name}.

• Priorité des paramètres : Les paramètres dans DataWorks suivent une règle de substitution hiérarchique. L'ordre de priorité est : Paramètres du nœud > Paramètres du flux de travail.

  Pour plus d'informations sur les paramètres, consultez la rubrique Parameter design and flow .

Politique de planification

Lorsque plusieurs flux de travail ou tâches sont déclenchés simultanément et que les ressources système deviennent un goulot d'étranglement, utilisez les options Priority et Weighting Strategy pour mettre en œuvre une planification intelligente des ressources et garantir que les tâches les plus importantes soient exécutées en premier.

• Protection des activités métier critiques : Définissez une priorité plus élevée pour les flux de travail métier essentiels afin qu'ils s'exécutent toujours avant les autres flux de travail non essentiels.

• Réduction de la durée du chemin critique : Au sein d'une même instance de flux de travail, en utilisant la Priority weighting strategy , vous pouvez influencer l'ordre d'exécution des nœuds. Par exemple, avec la stratégie Downstream weighting, les nœuds situés sur le chemin critique qui possèdent davantage de dépendances en amont reçoivent des poids dynamiques plus élevés et sont exécutés en premier, réduisant ainsi efficacement la durée totale d'exécution du flux de travail.

  Élément de configuration	Description
  Priority	Définit le niveau de priorité absolu des instances de flux de travail dans la file d'attente de planification. Les niveaux disponibles sont 1, 3, 5, 7 et 8 (plus le chiffre est élevé, plus la priorité est haute). Les tâches ou flux de travail ayant une priorité plus élevée acquièrent toujours les ressources de planification avant ceux ayant une priorité plus faible.
  Priority weighting strategy	Définit comment le poids dynamique de chaque nœud (tâche) au sein d'un flux de travail est calculé au même niveau de priorité. Les nœuds ayant des poids plus élevés sont exécutés en premier. Pas de pondération : Tous les nœuds ont le même poids de base fixe. Pondération descendante : Les poids des nœuds sont ajustés dynamiquement. Plus un nœud possède de dépendances en amont, plus son poids est élevé. Cette stratégie permet aux nœuds situés sur le chemin critique du DAG (graphe orienté acyclique) d'être exécutés en premier. Le poids est calculé comme suit : valeur de poids initiale + somme des priorités de tous les nœuds en amont.
  Maximum Parallel Instances for Internal Tasks	Contrôle le nombre maximal d'instances de ce flux de travail pouvant s'exécuter simultanément. Cela sert au contrôle de la concurrence et à la protection des ressources. Lorsque le nombre d'instances en cours d'exécution atteint la limite, les nouvelles instances déclenchées ultérieurement passent à l'état d'attente. Vous pouvez le définir sur Allowed ou spécifier une valeur maximale personnalisée (jusqu'à 100 000). Remarque Si vous définissez la limite sur une valeur dépassant la capacité maximale du groupe de ressources, le goulot d'étranglement réel en matière de concurrence est déterminé par la limite physique du groupe de ressources.

Le système de priorité dans DataWorks suit une règle de substitution hiérarchique : Spécification d'exécution > Configuration au niveau du nœud > Configuration au niveau du flux de travail.

1. Configuration au niveau du flux de travail (référence) : Configurée dans la Scheduling Policy du flux de travail et sert de paramètre par défaut pour tous les nœuds.

2. Configuration au niveau du nœud (local) : Dans les Scheduling Settings > Scheduling Policy d'un nœud individuel au sein du flux de travail, vous pouvez définir une Priority plus élevée pour un nœud spécifique, ce qui remplace le paramètre au niveau du flux de travail.

3. Spécification d'exécution (temporaire) : Spécifiée via l'option Runtime Priority Reset lorsque vous déclenchez manuellement une exécution dans le Operation and Maintenance Center. Cette configuration a la priorité la plus élevée, ne prend effet que pour l'exécution en cours et ne modifie aucune configuration permanente.