DataWorks:Workflow acionado

Etapa 1: crie um workflow acionado

1. Acesse a página Workspaces no console do DataWorks. Na barra de navegação superior, selecione a região desejada. Localize o workspace desejado e escolha Shortcuts > Data Studio na coluna Actions.

2. No painel de navegação à esquerda, clique em e, em seguida, clique em > Create Workflow no lado direito de Project Directory para acessar a página Create Workflow.

3. Na caixa de diálogo exibida, na página Create Workflow, defina Scheduling Type como Triggered scheduling. Insira um Name para o workflow e clique em Confirm para criar o workflow.

Etapa 2: orquestre o workflow e desenvolva nós

1. Clique em + Add Node na barra de ferramentas para abrir a lista de nós. Arraste um nó Shell da lista de tipos de nós à esquerda para a tela, insira um nome e conclua a criação.

2. Clique duas vezes no nó Shell para acessar a página de edição de código e insira o seguinte código:

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

3. Clique no botão Save na barra de ferramentas.

Etapa 3: depure e execute (ambiente de desenvolvimento)

1. Retorne à tela do workflow e clique no ícone na barra de ferramentas superior.

2. Na caixa de diálogo exibida, insira o Value Used in This Run para o workflow (por exemplo, se hoje for 20260310, substitua bizdate por 20260309).

3. Após alguns instantes, no log de execução na parte inferior, visualize o status de execução do nó e a saída do comando echo.

Etapa 4: implante e execute (ambiente de produção)

1. Na tela do workflow, clique no botão Deploy e siga as instruções para concluir o processo de implantação.

2. Após a implantação bem-sucedida, acesse Operation Center > Manually Triggered Task O&M > Manually Triggered Task> Triggered Workflow.

3. Localize o workflow recém-implantado e clique em Run na coluna Operation.

4. Na caixa de diálogo exibida, clique em Run novamente para acionar uma instância do workflow no ambiente de produção. Visualize os detalhes desta execução na página Manual Instances.

Agora você domina o uso básico de workflows acionados. A seguir, exploraremos os recursos mais avançados de acionamento por eventos.

Etapa 1: crie um gatilho do OSS

1. Acesse Operation Center > Scheduling Settings > Trigger Management.

2. Clique em Create Trigger e configure conforme abaixo:

   Nota

   Para descrições detalhadas dos parâmetros, consulte OSS trigger.

   • Trigger Name: insira um nome personalizado, como oss_new_file_trigger.

   • Applicable Workspace: selecione o workspace de destino onde o workflow reside.

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

   • Trigger Event: selecione oss:ObjectCreated:PutObject (ou outro evento de upload).

   • Bucket Name: selecione seu bucket do OSS.

   • File Name: especifique o caminho e o formato do arquivo a serem monitorados. Há suporte para caracteres curinga. Por exemplo, para monitorar todos os arquivos .csv no diretório input/, insira input/*.csv.

   • Role Configuration: no primeiro uso, realize a One-Click Authorization e selecione a função chamada DataWorks-EventBridge-OSS-MNS-Role-***.

     *** representa um número de ID de 13 dígitos gerado aleatoriamente para garantir unicidade.

3. Clique em Confirm para concluir a criação do gatilho.

Etapa 2: crie e associe um workflow

1. Siga as etapas em Quick start: Create a manually triggered workflow para criar um novo workflow acionado chamado process_oss_file_workflow.

2. No painel à direita da tela do workflow, acesse Scheduling Settings > Scheduling Policy.

3. Na lista suspensa Trigger, selecione o gatilho oss_new_file_trigger recém-criado.

   Após selecionar um gatilho, utilize ${workflow.triggerMessage} nas tarefas internas para obter o corpo completo da mensagem, ou use ${workflow.triggerMessage.xxx} para obter o valor de um campo específico no corpo da mensagem.

Etapa 3: desenvolva um nó e analise parâmetros de evento

1. Clique em + Add Node na barra de ferramentas para abrir a lista de nós. Arraste um nó Shell da lista de tipos de nós à esquerda para a tela, insira um nome e conclua a criação.

2. Clique duas vezes no nó e escreva o código para recuperar e imprimir o caminho do arquivo a partir do evento de acionamento.

   1. Uso geral: o uso de parâmetros a seguir aplica-se à maioria das mensagens de evento.

      # 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: alguns nós, como os nós Shell, possuem regras de sintaxe próprias. Antes da execução do nó, o mecanismo de agendamento primeiro substitui o conteúdo da mensagem no texto do script e depois o passa para o interpretador Shell executar. Consequentemente, aspas, espaços e quebras de linha na mensagem participam diretamente da análise sintática do Shell. Isso pode fazer com que espaços ou quebras de linha no parâmetro quebrem a estrutura original do código após o nó recuperar o parâmetro. Nesse caso, utilize um heredoc para ler a mensagem, evitando a maioria dos problemas com caracteres especiais:

      # 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 o corpo completo da mensagem de evento no formato JSON. Obtenha o formato específico da mensagem para OSS em EventBridge > Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

      Visualize um exemplo de formato de mensagem do OSS, que corresponde ao 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"
      }

Etapa 4: depure e publique

1. Depuração:

   • Retorne à tela do workflow e clique no botão Run .

   • Na caixa de entrada Trigger Message Body, cole um JSON simulando um evento do OSS. Copie e modifique o valor de key do "Exemplo de Formato de Mensagem" na página de configuração do gatilho. Abaixo está um exemplo simples.

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

   • Clique em Run e verifique se input/test_file_20260310.csv foi impresso corretamente nos logs.

2. Publicação: após a depuração bem-sucedida, clique no botão Publish para implantar o workflow no ambiente de produção. Gatilhos de evento entram em vigor apenas no ambiente de produção.

Etapa 5: verifique em produção

1. Utilize o console do OSS ou uma ferramenta cliente para carregar um arquivo CSV no bucket e no caminho (por exemplo, o diretório input/) configurados no gatilho.

   Como verifico se o gatilho de evento funciona?

   Acesse https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing para consultar a lista de eventos acionados recentemente. Clique também em Event Details para visualizar a mensagem específica do gatilho (ou seja, workflow.triggerMessage).

   A lista de eventos exibe registros de acionamento com a fonte de evento Object Storage Service (OSS) / acs.oss e o tipo de evento oss:ObjectCreated:PostObject.

2. Acesse o Operation Center do DataWorks > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow. O workflow process_oss_file_workflow publicado com sucesso aparecerá na lista.

3. Aguarde alguns instantes e acesse o Operation Center do DataWorks > Manually Triggered Task O&M > Triggered Workflow Instance. Uma nova instância de workflow será acionada automaticamente. Clique para visualizar seus logs e confirmar que o caminho do arquivo foi processado corretamente.

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

Cenário 2: chegada de mensagem do Kafka impulsiona computação em tempo real

Objetivo: monitore logs de comportamento de usuário no Kafka. Quando novas mensagens chegarem, acione um workflow para analisá-las e execute lógicas diferentes com base no conteúdo.

Orquestração de workflow

O processo central de orquestração para workflows acionados é semelhante ao de workflows agendados. Para mais informações, consulte Node/workflow orchestration.

Parâmetros de agendamento

No painel Scheduling Settings no lado direito da tela do workflow, defina parâmetros globais para o workflow. Todos os nós dentro do workflow podem referenciar esses parâmetros.

• Método de referência: no código do nó, referencie parâmetros do workflow no formato ${workflow.parameter_name}.

• Prioridade de parâmetros: os parâmetros no DataWorks seguem uma regra hierárquica de substituição. A ordem de prioridade é: Parâmetros de nó > Parâmetros de workflow.

  Para mais informações sobre parâmetros, consulte Parameter design and flow .

Política de agendamento

Quando múltiplos workflows ou tarefas são acionados simultaneamente e os recursos do sistema se tornam um gargalo, utilize Priority e Weighting Strategy para implementar agendamento inteligente de recursos e garantir que as tarefas mais importantes sejam executadas primeiro.

• Proteja o negócio principal: defina uma prioridade mais alta para workflows de negócios essenciais, garantindo que sempre executem antes de outros workflows não essenciais.

• Reduza a duração do caminho crítico: dentro da mesma instância de workflow, ao usar a Priority weighting strategy, influencie a ordem de execução dos nós. Por exemplo, com a estratégia Downstream weighting, nós no caminho crítico que possuem mais dependências upstream recebem pesos dinâmicos maiores e são executados primeiro, reduzindo efetivamente a duração total da execução do workflow.

  Item de configuração	Descrição
  Priority	Define o nível absoluto de prioridade das instâncias de workflow na fila de agendamento. Os níveis disponíveis são 1, 3, 5, 7 e 8 (quanto maior o número, maior a prioridade). Tarefas ou workflows com prioridade mais alta sempre adquirem recursos de agendamento antes daqueles com prioridade mais baixa.
  Priority weighting strategy	Define como o peso dinâmico de cada nó (tarefa) dentro de um workflow é calculado no mesmo nível de prioridade. Nós com pesos maiores são executados primeiro. Sem ponderação: todos os nós têm o mesmo peso base fixo. Ponderação downstream: os pesos dos nós são ajustados dinamicamente. Quanto mais dependências upstream um nó tiver, maior será seu peso. Essa estratégia ajuda os nós no caminho crítico do DAG (grafo acíclico direcionado) a serem executados primeiro. O peso é calculado como: valor de peso inicial + soma das prioridades de todos os nós upstream.
  Máximo de Instâncias Paralelas para Tarefas Internas	Controla o número máximo de instâncias deste workflow que podem ser executadas simultaneamente. Usado para controle de concorrência e proteção de recursos. Quando o número de instâncias em execução atinge o limite, novas instâncias acionadas subsequentemente entram em estado de espera. Defina como Allowed ou especifique um valor máximo personalizado (até 100.000). Nota Se definir o limite com um valor que exceda a capacidade máxima do grupo de recursos, o gargalo real de concorrência será determinado pelo limite físico do grupo de recursos.

O sistema de prioridades no DataWorks segue uma regra hierárquica de substituição: Especificação em tempo de execução > Configuração no nível do nó > Configuração no nível do workflow.

1. Configuração no nível do workflow (linha de base): configurada na Scheduling Policy do workflow e serve como configuração padrão para todos os nós.

2. Configuração no nível do nó (local): em Scheduling Settings > Scheduling Policy de um nó individual dentro do workflow, defina uma Priority mais alta para um nó específico, substituindo a configuração no nível do workflow.

3. Especificação em tempo de execução (temporária): especificada através do alternador Runtime Priority Reset ao acionar manualmente uma execução no Operation and Maintenance Center. Esta configuração tem a maior prioridade, entra em vigor apenas para a execução atual e não modifica nenhuma configuração permanente.