DataWorks:Flujo de trabajo activado

Paso 1: Crear un flujo de trabajo activado

1. Vaya a la página Workspaces en la consola de DataWorks. En la barra de navegación superior, seleccione la región deseada. Busque el espacio de trabajo correspondiente y elija Shortcuts > Data Studio en la columna Actions.

2. En el panel de navegación izquierdo, haga clic en y luego haga clic en > Create Workflow en el lado derecho de Project Directory para ir a la página Create Workflow.

3. En el cuadro de diálogo que aparece, en la página Create Workflow, establezca Scheduling Type en Triggered scheduling. Introduzca un Name para el flujo de trabajo y haga clic en Confirm para crearlo.

Paso 2: Orquestar el flujo de trabajo y desarrollar nodos

1. Haga clic en + Add Node en la barra de herramientas para abrir la lista de nodos. Arrastre un nodo Shell desde la lista de tipos de nodo a la izquierda hacia el lienzo, introduzca un nombre y complete la creación.

2. Haga doble clic en el nodo Shell para ir a la página de edición de código e introduzca el siguiente código:

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

3. Haga clic en el botón Save en la barra de herramientas.

Paso 3: Depurar y ejecutar (entorno de desarrollo)

1. Regrese al lienzo del flujo de trabajo y haga clic en el icono en la barra de herramientas superior.

2. En el cuadro de diálogo que aparece, introduzca el Value Used in This Run para el flujo de trabajo (por ejemplo, si hoy es 20260310, bizdate debe reemplazarse con 20260309).

3. Después de un momento, en el registro de ejecución en la parte inferior, podrá ver el estado de ejecución del nodo y la salida del comando echo.

Paso 4: Desplegar y ejecutar (entorno de producción)

1. En el lienzo del flujo de trabajo, haga clic en el botón Deploy y siga las indicaciones para completar el proceso de despliegue.

2. Una vez completado el despliegue correctamente, vaya a Operation Center > Manually Triggered Task O&M > Manually Triggered Task> Triggered Workflow.

3. Busque el flujo de trabajo que acaba de desplegar y haga clic en Run en la columna Operation.

4. En el cuadro de diálogo que aparece, haga clic en Run nuevamente para activar una instancia del flujo de trabajo en el entorno de producción. Puede consultar los detalles de esta ejecución en la página Manual Instances.

Ahora domina el uso básico de los flujos de trabajo activados. A continuación, exploraremos las capacidades más avanzadas de activación por eventos.

Paso 1: Crear un activador de OSS

1. Vaya a Operation Center > Scheduling Settings > Trigger Management.

2. Haga clic en Create Trigger y configure lo siguiente:

   Nota

   Para obtener descripciones detalladas de los parámetros, consulte OSS trigger.

   • Trigger Name: Introduzca un nombre personalizado, como oss_new_file_trigger.

   • Applicable Workspace: Seleccione el espacio de trabajo destino donde reside el flujo de trabajo.

   • Trigger Event Type: Seleccione Object Storage Service (OSS).

   • Trigger Event: Seleccione oss:ObjectCreated:PutObject (u otro evento de carga).

   • Bucket Name: Seleccione su bucket de OSS.

   • File Name: Especifique la ruta y el formato del archivo que desea monitorear. Se admiten comodines. Por ejemplo, para monitorear todos los archivos .csv en el directorio input/, introduzca input/*.csv.

   • Role Configuration: Si es la primera vez que lo usa, realice la One-Click Authorization y seleccione el rol llamado DataWorks-EventBridge-OSS-MNS-Role-*************.

     ************* representa un número de identificación de 13 dígitos generado aleatoriamente para garantizar la unicidad.

3. Haga clic en Confirm para completar la creación del activador.

Paso 2: Crear y asociar un flujo de trabajo

1. Siga los pasos en Quick start: Create a manually triggered workflow para crear un nuevo flujo de trabajo activado llamado process_oss_file_workflow.

2. En el panel derecho del lienzo del flujo de trabajo, vaya a Scheduling Settings > Scheduling Policy.

3. En la lista desplegable Trigger, seleccione el activador oss_new_file_trigger que acaba de crear.

   Después de seleccionar un activador, puede usar ${workflow.triggerMessage} en las tareas internas para obtener el cuerpo completo del mensaje, o usar ${workflow.triggerMessage.xxx} para obtener el valor de un campo específico en el cuerpo del mensaje.

Paso 3: Desarrollar un nodo y analizar sintácticamente los parámetros del evento

1. Haga clic en + Add Node en la barra de herramientas para abrir la lista de nodos. Arrastre un nodo Shell desde la lista de tipos de nodo a la izquierda hacia el lienzo, introduzca un nombre y complete la creación.

2. Haga doble clic en el nodo y escriba código para recuperar e imprimir la ruta del archivo desde el evento de activación.

   1. Uso general: El siguiente uso de parámetros aplica a la mayoría de los mensajes de eventos.

      # 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. Caso especial: Algunos nodos, como los nodos Shell, tienen sus propias reglas de sintaxis. Antes de que se ejecute el nodo, el motor de programación primero sustituye el contenido del mensaje en el texto del script y luego lo pasa al intérprete de Shell para su ejecución. Como resultado, las comillas, espacios y saltos de línea en el mensaje participan directamente en el análisis sintáctico de la sintaxis de Shell. Esto puede provocar que los espacios o saltos de línea en el parámetro rompan la estructura original del código después de que el nodo recupere el parámetro. En este caso, puede usar un heredoc para leer el mensaje, lo cual evita la mayoría de los problemas con caracteres especiales:

      # 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"

      Nota

      ${workflow.triggerMessage}: Recupera el cuerpo completo del mensaje de evento en formato JSON. Puede obtener el formato específico del mensaje para OSS desde EventBridge > Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

      Consulte un ejemplo de formato de mensaje de OSS, que corresponde al formato 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"
      }

Paso 4: Depurar y publicar

1. Depuración:

   • Regrese al lienzo del flujo de trabajo y haga clic en el botón Run .

   • En el cuadro de entrada Trigger Message Body, pegue un JSON de evento de OSS simulado. Puede copiar y modificar el valor key del "Message Format Sample" en la página de configuración del activador. A continuación se muestra un ejemplo sencillo.

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

   • Haga clic en Run y compruebe si input/test_file_20260310.csv se imprime correctamente en los registros.

2. Publicación: Una vez que la depuración sea exitosa, haga clic en el botón Publish para desplegar el flujo de trabajo en el entorno de producción. Los activadores de eventos solo surten efecto en el entorno de producción.

Paso 5: Verificar en producción

1. Use la consola de OSS o una herramienta cliente para cargar un archivo CSV en el bucket y la ruta (por ejemplo, el directorio input/) que configuró en el activador.

   ¿Cómo verifico que el activador de eventos funciona?

   Vaya a https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing para consultar la lista de eventos activados recientemente. También puede hacer clic en Event Details para ver el mensaje de activación específico (es decir, workflow.triggerMessage).

   La lista de eventos muestra registros de activación con el origen de evento Object Storage Service (OSS) / acs.oss y el tipo de evento oss:ObjectCreated:PostObject.

2. Vaya a Operation Center de DataWorks > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow. Aparecerá el flujo de trabajo process_oss_file_workflow publicado correctamente.

3. Espere un momento y luego vaya a Operation Center de DataWorks > Manually Triggered Task O&M > Triggered Workflow Instance. Se activa automáticamente una nueva instancia del flujo de trabajo. Haga clic para consultar sus registros y verificar que la ruta del archivo se procesó correctamente.

   ===== 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 =========

Escenario 2: La llegada de mensajes de Kafka impulsa el cómputo en tiempo real

Objetivo: Monitorear registros de comportamiento de usuarios en Kafka. Cuando lleguen nuevos mensajes, active un flujo de trabajo para analizarlos sintácticamente y ejecutar diferentes lógicas según el contenido.

Orquestación del flujo de trabajo

El proceso central de orquestación para flujos de trabajo activados es similar al de los flujos de trabajo programados. Para obtener más información, consulte Node/workflow orchestration.

Parámetros de programación

En el panel Scheduling Settings en el lado derecho del lienzo del flujo de trabajo, puede establecer parámetros globales para el flujo de trabajo. Todos los nodos dentro del flujo de trabajo pueden hacer referencia a estos parámetros.

• Método de referencia: En el código del nodo, haga referencia a los parámetros del flujo de trabajo con el formato ${workflow.parameter_name}.

• Prioridad de parámetros: Los parámetros en DataWorks siguen una regla jerárquica de anulación. El orden de prioridad es: Parámetros de nodo > Parámetros de flujo de trabajo.

  Para obtener más información sobre los parámetros, consulte Parameter design and flow .

Directiva de programación

Cuando varios flujos de trabajo o tareas se activan simultáneamente y los recursos del sistema se convierten en un cuello de botella, puede usar Priority y Weighting Strategy para implementar una programación inteligente de recursos y garantizar que las tareas más importantes se ejecuten primero.

• Proteger el negocio principal: Establezca una prioridad más alta para los flujos de trabajo del negocio principal de modo que siempre se ejecuten antes que otros flujos de trabajo no esenciales.

• Reducir la duración de la ruta crítica: Dentro de la misma instancia de flujo de trabajo, al usar la Priority weighting strategy, puede influir en el orden de ejecución de los nodos. Por ejemplo, con la estrategia Downstream weighting, los nodos en la ruta crítica que tienen más dependencias ascendentes reciben pesos dinámicos más altos y se ejecutan primero, lo que reduce eficazmente la duración total de ejecución del flujo de trabajo.

  Elemento de configuración	Descripción
  Priority	Define el nivel de prioridad absoluta de las instancias del flujo de trabajo en la cola de programación. Los niveles disponibles son 1, 3, 5, 7 y 8 (cuanto mayor sea el número, mayor será la prioridad). Las tareas o flujos de trabajo con mayor prioridad siempre obtienen recursos de programación antes que aquellos con menor prioridad.
  Priority weighting strategy	Define cómo se calcula el peso dinámico de cada nodo (tarea) dentro de un flujo de trabajo en el mismo nivel de prioridad. Los nodos con pesos más altos se ejecutan primero. Sin ponderación: Todos los nodos tienen el mismo peso base fijo. Ponderación descendente: Los pesos de los nodos se ajustan dinámicamente. Cuantas más dependencias ascendentes tenga un nodo, mayor será su peso. Esta estrategia ayuda a que los nodos en la ruta crítica del DAG (grafo acíclico dirigido) se ejecuten primero. El peso se calcula como: valor de peso inicial + suma de prioridades de todos los nodos ascendentes.
  Maximum Parallel Instances for Internal Tasks	Controla el número máximo de instancias de este flujo de trabajo que pueden ejecutarse simultáneamente. Se utiliza para el control de concurrencia y la protección de recursos. Cuando el número de instancias en ejecución alcanza el límite, las nuevas instancias activadas posteriormente entran en estado de espera. Puede establecerlo en Allowed o especificar un valor máximo personalizado (hasta 100,000). Nota Si establece el límite en un valor que excede la capacidad máxima del resource group, el cuello de botella real de concurrencia estará determinado por el límite físico del resource group.

El sistema de prioridades en DataWorks sigue una regla jerárquica de anulación: Especificación en tiempo de ejecución > Configuración a nivel de nodo > Configuración a nivel de flujo de trabajo.

1. Configuración a nivel de flujo de trabajo (base): Se configura en la Scheduling Policy del flujo de trabajo y sirve como configuración predeterminada para todos los nodos.

2. Configuración a nivel de nodo (local): En Scheduling Settings > Scheduling Policy de un nodo individual dentro del flujo de trabajo, puede establecer una Priority más alta para un nodo específico, lo cual anula la configuración a nivel de flujo de trabajo.

3. Especificación en tiempo de ejecución (temporal): Se especifica mediante el interruptor Runtime Priority Reset cuando activa manualmente una ejecución en el Operation and Maintenance Center. Esta configuración tiene la prioridad más alta, solo surte efecto para la ejecución actual y no modifica ninguna configuración permanente.