DataWorks:Nodo de asignación

Crear y usar un nodo de asignación en la consola

El siguiente ejemplo muestra el procedimiento general al transmitir el resultado de un nodo de asignación a un nodo Shell. En la práctica, cualquier tipo de nodo puede funcionar como nodo descendente.

1. Configurar el nodo de asignación ascendente

   En el flujo de trabajo de destino, cree y edite un nodo de asignación. Seleccione MaxCompute SQL, Python 2 o Shell, y escriba el código que genere el resultado que desea transmitir al nodo descendente.

   print '10,20,30,40'

2. Configurar el nodo Shell descendente

   Cree un nodo Shell. En la página de edición del nodo Shell, haga referencia al resultado ascendente:

   1. En el panel Scheduling Settings del lado derecho, seleccione la pestaña Node Context Parameters.

   2. En la sección Input Parameters, haga clic en Add parameters.

   3. En el cuadro de diálogo que aparece, establezca el parámetro de salida del nodo ascendente como el parámetro outputs del nodo de asignación configurado en el paso anterior. Luego, especifique un Parameter Name personalizado para el parámetro de entrada del nodo actual (por ejemplo, param).

      Nota

      Después de la configuración, el nodo descendente establece automáticamente una dependencia con el nodo de asignación ascendente.

   4. Una vez configurados los parámetros, utilice el valor transmitido desde el nodo ascendente en el código del nodo Shell descendente mediante el formato ${param}.

3. Configurar un nodo de Data Integration descendente

   En la práctica, un nodo de sincronización sin conexión de Data Integration también puede actuar como nodo descendente de un nodo de asignación:

   1. Cree un nodo de sincronización sin conexión de Data Integration y conéctelo como nodo descendente del nodo de asignación.

   2. En la página de configuración de programación del nodo de sincronización sin conexión, haga clic en la pestaña Node Context Parameters.

   3. Haga clic en Add parameters para agregar un parámetro de entrada y establezca su origen de valor como el parámetro outputs del nodo de asignación ascendente.

   4. Haga referencia al parámetro en el código del nodo: encierre el valor entre comillas dobles ("") si el parámetro es de tipo cadena. No agregue comillas si el parámetro no es de tipo cadena.

4. Comprobar el resultado

   1. Regrese al flujo de trabajo y haga clic en Deploy en la barra de herramientas. Seleccione el despliegue completo.

   2. Vaya a la página Node O&M > Auto Triggered Task O&M > Auto Triggered Task en Operation Center y ejecute smoke testing.

   3. En la instancia de prueba, compruebe si el resultado final cumple con sus expectativas.

Crear un nodo de asignación mediante OpenAPI

También puede crear un nodo de asignación al llamar a la operación CreateNode de DataWorks openapi.

Para crear un nodo en un flujo de trabajo existente, llame a ListWorkflowDefinitions para consultar el flujo de trabajo. Después, utilice el Id de nivel superior del objeto del flujo de trabajo en la respuesta como ContainerId. No use Script.Id.

Configure los siguientes parámetros de solicitud cuando llame a CreateNode:

• ProjectId: el ID del workspace de DataWorks.

• Scene: establezca este parámetro en DATAWORKS_PROJECT cuando cree un nodo en un proyecto de DataWorks.

• ContainerId: el Id de nivel superior del flujo de trabajo de destino. Especifique este parámetro solo cuando desee crear el nodo en un flujo de trabajo existente.

• Spec: la cadena JSON serializada de FlowSpec.

El siguiente ejemplo de FlowSpec crea un nodo de asignación de MaxCompute SQL. Asegúrese de que el lenguaje del script, el origen de datos y el contenido del script sean coherentes.

{
  "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>"
      }
    ]
  }
}

El parámetro Spec de CreateNode es una cadena. Si utiliza el SDK de Python, llame a json.dumps(flow_spec, ensure_ascii=False) para serializar FlowSpec antes de pasar el valor a CreateNodeRequest.spec. No pase un diccionario de Python directamente.

Especifique el identificador del resource group en spec.nodes[].runtimeResource.resourceGroup. Utilice el identificador del resource group en lugar del nombre para mostrar.

El Id que devuelve CreateNode es el ID real del nodo generado por el servidor y puede diferir de nodes[].id en FlowSpec. Una vez creado el nodo, llame a GetNode con el Id devuelto. Compruebe el tipo de nodo, el flujo de trabajo, el origen de datos y el resource group en la respuesta.

Ejemplo 1: transmitir resultados de consultas de MaxCompute SQL

Ejemplo 2: transmitir resultados de salida de Python 2

La salida de la instrucción print de Python 2 se divide por comas (,) y se transmite a los nodos descendentes como una matriz unidimensional.

• Configuración del nodo ascendente (nodo de asignación - Python 2)

  El código de Python 2 es el siguiente:

  print 'Electronics, Clothing, Books';

• Configuración y salida del nodo descendente (nodo Shell)

  En el nodo Shell, agregue un parámetro de entrada llamado types y haga referencia al parámetro outputs del nodo de asignación ascendente.

  Escriba el siguiente código para leer los datos: # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  DataWorks analiza el parámetro directamente y realiza una sustitución estática. La salida es la siguiente: Entire result set: Electronics,Clothing,Books Second element: Clothing

Contexto

Suponga que usted es ingeniero de desarrollo de datos en una empresa integral de internet y se encarga de procesar datos de tres líneas de negocio principales: comercio electrónico (ecom), finanzas (finance) y logística (logistics), con la posibilidad de agregar más en el futuro. Necesita ejecutar la misma lógica de agregación todos los días sobre los registros de comportamiento de usuarios de estas tres líneas de negocio para calcular las vistas de página diarias (PV) por usuario y almacenar los resultados en una tabla agregada unificada.

• Tablas de origen ascendentes (capa DWD):

  • dwd_user_behavior_ecom_d: tabla de comportamiento de usuarios de comercio electrónico.

  • dwd_user_behavior_finance_d: tabla de comportamiento de usuarios de finanzas.

  • dwd_user_behavior_logistics_d: tabla de comportamiento de usuarios de logística.

  • dwd_user_behavior_${business_line}_d: tablas de comportamiento de usuarios para posibles líneas de negocio futuras.

  • Estas tablas tienen el mismo esquema y están particionadas por día (dt).

• Tabla de destino descendente (capa DWS):

  • dws_user_summary_d: tabla agregada de usuarios.

  • Esta tabla tiene doble partición por línea de negocio (biz_line) y por día (dt) para almacenar de manera unificada los resultados agregados de todas las líneas de negocio.

Crear una tarea independiente para cada línea de negocio genera altos costos de mantenimiento y es propenso a errores. Con un nodo for-each, usted mantiene un único conjunto de lógica de procesamiento y el sistema itera automáticamente por todas las líneas de negocio para completar el cálculo.

Preparación de datos

Primero, cree las tablas de ejemplo e inserte datos de prueba (se toma como ejemplo la fecha de negocio 20251010).

1. Associate a compute resource con el workspace.

2. Vaya a Data Studio para el desarrollo de datos y cree un nodo de MaxCompute SQL.

3. Cree las tablas de origen (capa DWD): agregue el siguiente código al nodo de MaxCompute SQL y ejecútelo.

   -- 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. Cree la tabla de destino (capa DWS): agregue el siguiente código al nodo de MaxCompute SQL y ejecútelo.

   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

   Si el workspace utiliza el modo estándar, debe desplegar este nodo en el entorno de producción y rellenar los datos.

Implementación del flujo de trabajo

1. Cree un flujo de trabajo. En la sección Scheduling Parameters del lado derecho, establezca el parámetro de programación bizdate en el día anterior: $[yyyymmdd-1].

2. En el flujo de trabajo, cree un nodo de asignación llamado get_biz_list y escriba el siguiente código en MaxCompute SQL. Este nodo genera la lista de líneas de negocio que se van a procesar:

   -- 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. Configurar el nodo for-each

   • Regrese a la página del flujo de trabajo y cree un nodo for-each descendente para el nodo de asignación get_biz_list.

   • Abra la página de configuración del nodo for-each. En la sección Scheduling Parameters > Script Parameters dentro de schedule settings del lado derecho, vincule el parámetro loopDataArray a los outputs del nodo get_biz_list.

   • En el cuerpo del bucle del nodo for-each, haga clic en Create Internal Node y cree un nodo de MaxCompute SQL. Escriba la lógica de procesamiento dentro del cuerpo del bucle.

     Nota

     • Este script lo controla el nodo for-each y se ejecuta una vez por cada línea de negocio.

     • La variable integrada ${dag.foreach.current} se reemplaza dinámicamente con el nombre de la línea de negocio actual en cada iteración. Los valores de iteración esperados son: '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. Agregar un nodo de verificación

   Regrese al flujo de trabajo. Haga clic en Create Downstream en el nodo for-each para crear un nodo de MaxCompute SQL y agregue el siguiente código.

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

Despliegue y resultados

Despliegue el flujo de trabajo en el entorno de producción. Vaya a la página Auto Triggered Task O&M > Auto Triggered Task en Operation Center. Busque el flujo de trabajo de destino y ejecute una prueba de humo con la fecha de negocio establecida en '20251010'.

Cuando finalice la ejecución, consulte el registro de tiempo de ejecución en la instancia de prueba. La salida esperada del nodo final es la siguiente:

Ventajas

• Alta escalabilidad: para agregar una nueva línea de negocio, basta con añadir una línea de SQL en el nodo de asignación, sin modificar la lógica de procesamiento.

• Fácil mantenimiento: todas las líneas de negocio comparten un único conjunto de lógica de procesamiento. Una sola modificación se aplica a todas.