DataWorks:Nó de atribuição

Criar e usar um nó de atribuição no console

O exemplo a seguir demonstra o procedimento geral de transmissão do resultado de um nó de atribuição para um nó Shell. Na prática, qualquer tipo de nó pode atuar como downstream.

1. Configure o nó de atribuição upstream

   No workflow de destino, crie e edite um nó de atribuição. Selecione MaxCompute SQL, Python 2 ou Shell e escreva o código que produz o resultado desejado para transmissão a um nó downstream.

   print '10,20,30,40'

2. Configure o nó Shell downstream

   Crie um nó Shell. Na página de edição do nó Shell, referencie o resultado upstream:

   1. No painel Scheduling Settings à direita, selecione a aba Node Context Parameters.

   2. Na seção Input Parameters, clique em Add parameters.

   3. Na caixa de diálogo exibida, defina o parâmetro de saída do nó upstream como o parâmetro outputs do nó de atribuição configurado na etapa anterior e especifique um Parameter Name personalizado para o parâmetro de entrada do nó atual (por exemplo, param).

      Nota

      Após a configuração, o nó downstream estabelece automaticamente uma dependência do nó de atribuição upstream.

   4. Após configurar os parâmetros, utilize o valor transmitido pelo nó upstream no código do nó Shell downstream com o formato ${param}.

3. Configure um nó de Data Integration downstream

   Na prática, um nó de sincronização offline do Data Integration também pode atuar como downstream de um nó de atribuição:

   1. Crie um nó de sincronização offline do Data Integration e conecte-o como downstream do nó de atribuição.

   2. Na página de configuração de agendamento do nó de sincronização offline, clique na aba Node Context Parameters.

   3. Clique em Add parameters para adicionar um parâmetro de entrada e defina sua origem de valor como o parâmetro outputs do nó de atribuição upstream.

   4. Referencie o parâmetro no código do nó: coloque o valor entre aspas duplas ("") se o parâmetro for do tipo string. Não adicione aspas se o parâmetro não for do tipo string.

4. Verifique o resultado

   1. Retorne ao workflow e clique em Deploy na barra de ferramentas. Selecione a implantação completa.

   2. Acesse a página Node O&M > Auto Triggered Task O&M > Auto Triggered Task no Operation Center e execute smoke testing.

   3. Na instância de teste, verifique se o resultado final atende às suas expectativas.

Criar um nó de atribuição usando a OpenAPI

Também é possível criar um nó de atribuição chamando a operação CreateNode da OpenAPI do DataWorks.

Para criar um nó em um workflow existente, chame ListWorkflowDefinitions para consultar o workflow. Em seguida, use o Id de nível superior do objeto workflow na resposta como ContainerId. Não use Script.Id.

Configure os seguintes parâmetros de solicitação ao chamar CreateNode:

• ProjectId: ID do workspace do DataWorks.

• Scene: defina este parâmetro como DATAWORKS_PROJECT ao criar um nó em um projeto do DataWorks.

• ContainerId: Id de nível superior do workflow de destino. Especifique este parâmetro apenas quando quiser criar o nó em um workflow existente.

• Spec: string JSON serializada do FlowSpec.

O exemplo de FlowSpec a seguir cria um nó de atribuição MaxCompute SQL. Certifique-se de que a linguagem do script, a fonte de dados e o conteúdo do script estejam consistentes.

{
  "version": "1.1.0",
  "kind": "Node",
  "spec": {
    "nodes": [
      {
        "recurrence": "Normal",
        "instanceMode": "T+1",
        "datasource": {
          "name": "<MAXCOMPUTE_DATASOURCE_NAME>",
          "type": "odps"
        },
        "script": {
          "language": "odps-sql",
          "runtime": {
            "command": "CONTROLLER_ASSIGNMENT",
            "commandTypeId": 1100
          },
          "content": "SELECT '10' AS value UNION ALL SELECT '20';"
        },
        "trigger": {
          "type": "Scheduler",
          "cron": "00 00 00 * * ?",
          "startTime": "1970-01-01 00:00:00",
          "endTime": "9999-01-01 00:00:00",
          "timezone": "Asia/Shanghai",
          "delaySeconds": 0
        },
        "runtimeResource": {
          "resourceGroup": "<RESOURCE_GROUP_IDENTIFIER>"
        },
        "name": "assignment_node_demo",
        "owner": "<OWNER_ID>"
      }
    ]
  }
}

O parâmetro Spec do CreateNode é uma string. Se você usar o SDK Python, chame json.dumps(flow_spec, ensure_ascii=False) para serializar o FlowSpec antes de passar o valor para CreateNodeRequest.spec. Não passe um dicionário Python diretamente.

Especifique o identificador do grupo de recursos em spec.nodes[].runtimeResource.resourceGroup. Use o identificador do grupo de recursos em vez do nome de exibição.

O Id retornado pelo CreateNode é o ID real do nó gerado pelo servidor e pode diferir de nodes[].id no FlowSpec. Após criar o nó, chame GetNode com o Id retornado. Verifique o tipo de nó, o workflow, a fonte de dados e o grupo de recursos na resposta.

Exemplo 1: Transmitir resultados de consulta MaxCompute SQL

Exemplo 2: Transmitir resultados de saída Python 2

A saída da instrução print do Python 2 é dividida por vírgulas (,) e transmitida aos nós downstream como um array unidimensional.

• Configuração do nó upstream (nó de atribuição - Python 2)

  O código Python 2 é o seguinte:

  print 'Electronics, Clothing, Books';

• Configuração e saída do nó downstream (nó Shell)

  No nó Shell, adicione um parâmetro de entrada chamado types e referencie o parâmetro outputs do nó de atribuição upstream.

  Escreva o seguinte código para ler os dados: # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  O DataWorks analisa diretamente o parâmetro e realiza a substituição estática. A saída é a seguinte: Entire result set: Electronics,Clothing,Books Second element: Clothing

Contexto

Suponha que você seja um engenheiro de desenvolvimento de dados em uma empresa abrangente de internet, responsável por processar dados de três linhas de negócio principais: e-commerce (ecom), finanças (finance) e logística (logistics), com possibilidade de adicionar mais no futuro. Você precisa executar diariamente a mesma lógica de agregação nos logs de comportamento do usuário dessas três linhas para calcular as visualizações diárias de página (PV) por usuário e armazenar os resultados em uma tabela agregada unificada.

• Tabelas de source upstream (camada DWD):

  • dwd_user_behavior_ecom_d: Tabela de comportamento do usuário de e-commerce.

  • dwd_user_behavior_finance_d: Tabela de comportamento do usuário de finanças.

  • dwd_user_behavior_logistics_d: Tabela de comportamento do usuário de logística.

  • dwd_user_behavior_${business_line}_d: Tabelas de comportamento do usuário para possíveis linhas de negócio futuras.

  • Essas tabelas possuem o mesmo esquema e são particionadas por dia (dt).

• Tabela de destino downstream (camada DWS):

  • dws_user_summary_d: Tabela agregada de usuários.

  • Esta tabela possui partição dupla por linha de negócio (biz_line) e dia (dt) para armazenar os resultados agregados de todas as linhas de negócio de maneira unificada.

Criar uma tarefa separada para cada linha de negócio resulta em altos custos de manutenção e propensão a erros. Com um nó for-each, você mantém um único conjunto de lógica de processamento, e o sistema itera automaticamente por todas as linhas de negócio para concluir a computação.

Preparação dos dados

Primeiro, crie as tabelas de exemplo e insira dados de teste (usando a data de negócio 20251010 como exemplo).

1. Associate a compute resource com o workspace.

2. Acesse o Data Studio para desenvolvimento de dados e crie um nó MaxCompute SQL.

3. Crie as tabelas de source (camada DWD): Adicione o seguinte código ao nó MaxCompute SQL e execute-o.

   -- 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. Crie a tabela de destino (camada DWS): Adicione o seguinte código ao nó MaxCompute SQL e execute-o.

   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'
   );

   Importante

   Se o workspace utilizar o modo padrão, implante este nó no ambiente de produção e realize o backfill dos dados.

Implementação do workflow

1. Crie um workflow. Na seção Scheduling Parameters à direita, defina o parâmetro de agendamento bizdate para o dia anterior: $[yyyymmdd-1].

2. No workflow, crie um nó de atribuição chamado get_biz_list e escreva o seguinte código em MaxCompute SQL. Este nó gera a lista de linhas de negócio a serem processadas:

   -- 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. Configure o nó for-each

   • Retorne à página do workflow e crie um nó for-each downstream para o nó de atribuição get_biz_list.

   • Abra a página de configurações do nó for-each. Na seção Scheduling Parameters > Script Parameters sob schedule settings à direita, vincule o parâmetro loopDataArray aos outputs do nó get_biz_list.

   • No corpo do loop do nó for-each, clique em Create Internal Node e crie um nó MaxCompute SQL. Escreva a lógica de processamento dentro do corpo do loop.

     Nota

     • Este script é impulsionado pelo nó for-each e executado uma vez para cada linha de negócio.

     • A variável interna ${dag.foreach.current} é substituída dinamicamente pelo nome da linha de negócio atual em cada iteração. Os valores de iteração esperados são: '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. Adicione um nó de verificação

   Volte ao workflow. Clique em Create Downstream no nó for-each para criar um nó MaxCompute SQL e adicione o seguinte código.

   SELECT * FROM dws_user_summary_d WHERE dt='20251010' ORDER BY biz_line, user_id;

Implantação e resultados

Faça o Deploy do workflow para o ambiente de produção. Acesse a página Auto Triggered Task O&M > Auto Triggered Task no Operation Center. Localize o workflow de destino e execute testes de fumaça com a data de negócio definida como '20251010'.

Após a conclusão da execução, visualize o log de tempo de execução na instância de teste. A saída esperada do nó final é a seguinte:

Vantagens

• Alta escalabilidade: Para adicionar uma nova linha de negócio, basta inserir uma linha de SQL no nó de atribuição, sem modificar a lógica de processamento.

• Facilidade de manutenção: Todas as linhas de negócio compartilham um único conjunto de lógica de processamento. Uma única modificação se aplica a todas.