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 e escolha Shortcuts > Data Studio na coluna Actions.

2. Na barra de navegação à esquerda, clique em . Em seguida, ao lado de Project Directory, clique em > Create Workflow para abrir a página Create Workflow.

3. Na caixa de diálogo Create Workflow, defina Scheduling Type como Triggered scheduling, insira um Name para o workflow e clique em Confirm.

Etapa 2: orquestre e desenvolva nós

1. Na barra de ferramentas, clique em + Add Node para abrir a lista de nós. Na lista de tipos de nó à esquerda, arraste um nó Shell para a tela e insira um nome para criá-lo.

2. Clique duas vezes no nó Shell para abrir o editor 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

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 esta execução (por exemplo, se hoje for 20260310, substitua bizdate por 20260309).

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

Etapa 4: publique e execute

1. Acima da tela do workflow, clique no botão Publish e siga as instruções para publicá-lo.

2. Após publicar o workflow, acesse Operation Center > Manually Triggered Task O&M > Manually Triggered Task> Triggered Workflow.

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

4. Na janela exibida, clique em Run novamente para disparar uma instância do workflow no ambiente de produção. A página Triggered Workflow Instance exibe os detalhes desta execução.

Agora você conhece os conceitos básicos sobre workflows acionados. A seguir, exploraremos os recursos mais avançados de acionamento por eventos.

Etapa 1: crie um trigger do OSS

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

2. Clique em Create Trigger e configure-o da seguinte forma:

   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 está localizado.

   • Trigger Event Type: selecione Object Storage 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 monitorar. Há suporte para wildcards. Por exemplo, para monitorar todos os arquivos .csv no diretório input/, insira input/*.csv.

   • Role Configuration: no primeiro uso, realize a autorização com um clique. Selecione a função gerada chamada DataWorks-EventBridge-OSS-MNS-Role-***.

     O *** representa um ID de 13 dígitos gerado aleatoriamente para garantir unicidade.

3. Clique em Confirm para criar o trigger.

Etapa 2: crie e associe o workflow

1. Siga as etapas em Início rápido: crie um workflow acionado manualmente para criar um novo workflow acionado chamado process_oss_file_workflow.

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

3. Na lista suspensa Trigger, selecione o oss_new_file_trigger criado anteriormente.

   Após selecionar um trigger, use a variável ${workflow.triggerMessage} nas tarefas internas para obter o corpo completo da mensagem ou utilize ${workflow.triggerMessage.xxx} para obter um campo específico.

Etapa 3: analise os parâmetros do evento

1. Na barra de ferramentas, clique em + Add Node para abrir a lista de nós. Na lista de tipos de nó à esquerda, arraste um nó Shell para a tela e insira um nome para criá-lo.

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

   # When a trigger starts a workflow, the event information is passed through the built-in variable workflow.triggerMessage
   # We can get the full path of the uploaded file using ${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}"
   # You can write specific processing logic here
   echo "========= Finish Processing OSS File ========="

   Nota

   ${workflow.triggerMessage}: obtém o corpo completo da mensagem de evento no formato JSON. Encontre o formato específico da mensagem para OSS no EventBridge navegando até Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracking> Event Detail.

   Exemplo de formato de mensagem do OSS (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 o exemplo de formato de mensagem da página de configuração do trigger e modifique o valor de key. Veja um exemplo mínimo abaixo.

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

   • Clique em Run e verifique os logs para confirmar se input/test_file_20260310.csv foi impresso com sucesso.

2. Publicação: após depurar com sucesso, clique no botão Publish para implantar o workflow no ambiente de produção. O acionamento por eventos só entra em vigor no ambiente de produção.

Etapa 5: verifique em produção

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

   Como garantir que o evento foi acionado com sucesso

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

   A lista de eventos mostrará um registro de acionamento com a fonte de evento como Object Storage Service (OSS) / acs.oss e o tipo de evento como oss:ObjectCreated:PostObject.

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

3. Após uma breve espera, acesse o Operation Center do DataWorks > Manually Triggered Task O&M > Triggered Workflow Instance. Uma nova instância do workflow será acionada automaticamente. Clique para visualizar seu log e confirmar se o caminho do arquivo foi processado corretamente.

Cenário 2: processe novas mensagens do Kafka

Objetivo: monitore um tópico do Kafka em busca de logs de comportamento do usuário. Quando uma nova mensagem chegar, acione um workflow para analisá-la e executar lógicas diferentes com base no conteúdo.

Orquestração de workflow

O processo principal para orquestrar um workflow acionado é semelhante ao de um workflow agendado. Para mais informações, consulte Orchestrate nodes and workflows.

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 um parâmetro do workflow usando o formato ${workflow.parameter_name}.

• Prioridade de parâmetros: os parâmetros no DataWorks possuem uma relação hierárquica de substituição. A prioridade efetiva é: parâmetros de nó > parâmetros de workflow.

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

Políticas de agendamento

Quando múltiplos workflows ou tarefas são acionados simultaneamente e causam gargalo de recursos, utilize Priority e priority weighting policies para alcançar agendamento inteligente de recursos e garantir que as tarefas mais importantes executem primeiro.

• Garantia de negócios essenciais: defina uma prioridade maior para workflows de negócios críticos para assegurar que sempre executem antes de outros workflows não essenciais.

• Redução do tempo de processos críticos: dentro de uma única instância de workflow, influencie a ordem de execução dos nós usando a Priority weighting strategy. Por exemplo, usar a política de ponderação descendente atribui um peso dinâmico maior aos nós no caminho crítico que possuem mais dependências upstream. Isso faz com que executem primeiro, reduzindo o tempo total de execução do workflow.

  Parâmetro	Descrição
  Priority	Define o nível absoluto de prioridade de uma instância 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 de alta prioridade sempre obtêm recursos de agendamento antes dos de baixa prioridade.
  Priority weighting strategy	Define como os pesos de nós individuais (tarefas) dentro de um workflow são calculados dinamicamente no mesmo nível de prioridade. Nós com pesos maiores recebem prioridade de execução. Sem ponderação: todos os nós têm um peso base fixo. Ponderação descendente: o peso de um nó é ajustado dinamicamente. Quanto mais nós upstream um nó depender, maior será seu peso. Esta política ajuda a priorizar nós no caminho crítico de um Grafo Acíclico Direcionado (DAG). O peso é calculado como: Peso inicial + Soma das prioridades de todos os nós upstream.
  	Controla o número máximo de instâncias deste workflow que podem executar simultaneamente, usado para controle de concorrência e proteção de recursos. Quando o número de instâncias em execução atinge esse limite, novas instâncias acionadas entram em fila. Defina como Allowed ou um valor máximo personalizado (até 100.000). Nota Se o limite definido exceder a capacidade do grupo de recursos, a concorrência real será limitada pelos limites físicos do grupo de recursos.

O sistema de prioridades do 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 (base): configurada na Scheduling Policy do workflow como configuração padrão para todos os nós.

2. Configuração no nível do nó (local): nas Scheduling Settings > Scheduling Policy de um único nó, defina uma Priority maior para um nó específico, substituindo a configuração no nível do workflow.

3. Especificação em tempo de execução (temporária): ao acionar manualmente uma execução no Operation and Maintenance Center, use a opção Runtime Priority Reset para especificar uma configuração temporária. Esta configuração tem a maior precedência e aplica-se apenas à execução atual, sem modificar configurações permanentes.