DataWorks:Nó de atribuição

{
  "version": "1.1.0",
  "kind": "Node",
  "spec": {
    "nodes": [
      {
        "recurrence": "Normal",
        "script": {
          "runtime": {
            "command": "CONTROLLER_ASSIGNMENT"
          },
          "content": "print '10,20,30'"
        },
        "runtimeResource": {
          "resourceGroup": "S_res_group_XXX_XXXX"
        },
        "name": "assignment_node_demo"
      }
    ]
  }
}

Exemplo 1: Passar resultados de consulta MaxCompute SQL

Exemplo 2: Passar resultados de saída Python 2

A saída da instrução print do Python 2 é dividida por vírgulas (,) e passada para os 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 e responda pelo processamento de dados de três linhas de negócio principais: e-commerce (ecom), finanças (finance) e logística (logistics), com a possibilidade de adicionar mais linhas de negócio no futuro. Execute a mesma lógica de agregação nos logs de comportamento do usuário dessas três linhas de negócio diariamente 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 origem 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. Ao usar um nó for-each, mantenha apenas um único conjunto de lógica de processamento, e o sistema itera automaticamente por todas as linhas de negócio para concluir o cálculo.

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. Associe um recurso de computação ao workspace.

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

3. Crie as tabelas de origem (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 usar o modo padrão, implante este nó no ambiente de produção e faça 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 dentro de 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

   Retorne 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 realize 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: Ao 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 o mesmo conjunto de lógica de processamento. Uma única modificação se aplica a todas as linhas de negócio.