Data Transmission Service:Uso de ETL para analizar pedidos en tiempo real

Casos de uso

Para cumplir con los requisitos empresariales de procesamiento de datos en tiempo real, la función de ETL ofrece capacidades de extracción, transformación y carga de datos en streaming. Integra de manera eficiente grandes volúmenes de datos en tiempo real, admite operaciones de arrastrar y soltar, y utiliza un enfoque de desarrollo low-code. Esto permite a las empresas analizar decisiones de negocio, agilizar la generación de informes y procesar datos en tiempo real. Durante la transformación digital, los casos de uso más comunes del procesamiento de datos en tiempo real incluyen:

• Consolidación en tiempo real de datos multirregión o heterogéneos: almacene datos de varias regiones o de fuentes heterogéneas en una sola base de datos en tiempo real. De esta forma, facilita la gestión centralizada y eficiente, así como la toma de decisiones.

• Generación acelerada de informes: construya un sistema de informes en tiempo real que mejore significativamente la eficiencia de generación. Este sistema admite más escenarios de análisis en tiempo real y satisface las altas exigencias de eficiencia durante la transformación digital.

• Computación en tiempo real: limpie y procese datos de negocio en streaming de forma inmediata para generar valores de características y etiquetas. Esta capacidad es compatible con modelos de negocio en línea, como perfiles de usuario, control de riesgos y sistemas de recomendación, además de paneles en tiempo real y otros escenarios de computación en streaming.

Contexto

Este ejemplo demuestra cómo usar la función de ETL en streaming para unir datos de transacciones en tiempo real (como ID de pedido, ID de cliente, código de producto, monto de la transacción y hora de la transacción) con datos de dimensiones de negocio (como código de producto, precio unitario y nombre del producto). Los datos que cumplen condiciones de filtro específicas, por ejemplo, transacciones con un monto mayor a 3,000, se cargan en un almacén de datos en tiempo real. Esto habilita el análisis multidimensional de las transacciones por dimensiones como producto y cliente. También puede usar herramientas de visualización para crear paneles dinámicos que generen información basada en datos.

Procedimiento

Preparativos

Antes de configurar la tarea de ETL, almacene los datos de transacciones en tiempo real en una tabla de streaming y los datos de dimensiones de negocio en una tabla de dimensiones dentro de la base de datos de origen de ApsaraDB RDS for MySQL.

A continuación, cree una tabla en la base de datos de destino de ApsaraDB RDS for MySQL según sus requisitos de negocio.

create table test_orders(
   order_id bigint not null COMMENT 'Order ID',
     user_id bigint not null comment 'User ID',
     product_id bigint not null comment 'Product ID',
   total_price decimal(15,2) not null COMMENT 'Total order value',
   order_date TIMESTAMP  not null COMMENT 'Order date',
   PRIMARY KEY (order_id))

CREATE table product (
      product_id bigint not null comment 'Product ID',
      product_name varchar(20) comment 'Product name',
      product_price decimal(15,2) not null comment 'Product unit price')

create table test_orders(
 order_id bigint not null COMMENT 'Order ID',
 user_id bigint not null comment 'User ID',
 product_id bigint not null comment 'Product ID',
 total_price decimal(15,2) not null COMMENT 'Total order value',
 order_date TIMESTAMP  not null
 COMMENT 'Order date',
 product_id_2 bigint not null comment 'Product ID',
 product_name varchar(20) comment 'Product name',
 product_price decimal(15,2) not null comment 'Product unit price',
 PRIMARY KEY (order_id))

Paso 1: Configurar la base de datos de origen

1. Vaya a la página de lista de tareas de ETL.

   1. Inicie sesión en la consola de Data Transmission Service (DTS).

   2. En el panel de navegación izquierdo, haga clic en ETL.

2. En la esquina superior izquierda, haga clic en . En el cuadro de diálogo Create Data Flow, introduzca un nombre para la tarea de ETL en el campo Data Flow Name y establezca Development Method en DAG.

3. Haga clic en Confirm.

4. Configure la tabla de streaming y la tabla de dimensiones.

   1. Configuración de la tabla de streaming

      1. Desde el panel izquierdo, arrastre el nodo Input/Dimension Table (MySQL) hacia el lienzo de la derecha.

      2. Haga clic en el nodo Input/Dimension Table MySQL-1 del lienzo y configure la base de datos de origen.

      3. En la pestaña Node Configuration, establezca los siguientes parámetros.

         Parámetro	Descripción
         Data Source Name	Para facilitar su identificación, se recomienda usar un nombre descriptivo y no único.
         Region	Seleccione la región de la base de datos de origen. Nota Las tareas de ETL solo están disponibles en las siguientes regiones: China (Hangzhou), China (Shanghai), China (Qingdao), China (Beijing), China (Zhangjiakou), China (Shenzhen), China (Guangzhou) y China (Hong Kong).
         Instances	Seleccione el nombre de la instancia que aloja la base de datos de origen. También puede hacer clic en Create Instance para crear una nueva instancia. Para obtener más información, consulte Databases supported by DMS.
         Node Type	Seleccione el tipo de tabla de origen. En este ejemplo, seleccione Stream Table. Stream Table: tabla que cambia en tiempo real. Se puede unir con una tabla de dimensiones para enriquecer los datos. Dimension Tables: tabla que se actualiza con poca frecuencia (no en tiempo real). Por lo general, se usa para unirla con datos en tiempo real y crear una tabla ancha destinada al análisis.
         Convert Format	Al procesar datos, ETL convierte un flujo en una tabla dinámica y ejecuta consultas continuas sobre ella. Las operaciones INSERT, UPDATE y DELETE modifican constantemente la tabla dinámica, lo que genera una nueva tabla dinámica. Cuando se escriben datos en la base de datos de destino, esta nueva tabla dinámica se vuelve a convertir en un flujo. Debe especificar un conversion format para codificar los cambios: Upsert Stream: los datos de la tabla dinámica se pueden modificar mediante operaciones INSERT, UPDATE y DELETE. Al convertirse en un flujo, las operaciones INSERT y UPDATE se codifican como mensajes upsert, mientras que las operaciones DELETE se codifican como mensajes delete. Nota Este formato de codificación requiere que la tabla dinámica tenga una clave única, la cual puede ser una clave compuesta. Append-Only Stream: los datos de la tabla dinámica solo se pueden modificar mediante operaciones INSERT. Al convertirse en un flujo, únicamente se envían los datos insertados.
         Select Databases and Tables	Seleccione las bases de datos y tablas de origen que desea transformar.

      4. Una vez configurado el nodo, se muestra la pestaña Output Fields. En la columna Column Name, seleccione los campos obligatorios.

      5. En este ejemplo se utiliza una tabla de streaming. Haga clic en la pestaña Time Attribute y establezca los siguientes parámetros.

         Parámetro	Descripción
         Select Event Time Watermark	Seleccione un campo de tiempo de la tabla de streaming. Una tabla de streaming suele tener un campo de tiempo que indica cuándo se generaron los datos, como una marca de tiempo específica del negocio, por ejemplo, ordertime.
         Latency of Event Time Watermark	Introduzca la latencia máxima de datos que puede tolerar. Debido a las condiciones de la red, es posible que los datos no lleguen al servicio de ETL en el orden en que se generaron, lo que provoca retrasos. ETL no puede esperar indefinidamente por datos que llegan tarde. Por lo tanto, debe especificar un periodo de latencia para manejar los datos desordenados. Por ejemplo, si llegaron los datos de las 10:00 pero no los de las 9:59, ETL espera solo hasta las 10:00 más la latencia especificada. Si los datos de las 9:59 llegan después de ese periodo, ETL los descarta.
         Processing Time	El tiempo de procesamiento es la hora local del servidor cuando ETL procesa los datos. Debe definir una columna para almacenar este valor. Se usa principalmente para cálculos de operadores; por ejemplo, una unión temporal utiliza el tiempo de procesamiento para buscar la versión más reciente de una tabla normal.

      Nota

      La configuración finaliza cuando desaparece el ícono de signo de exclamación del lado derecho del nodo de la base de datos de origen.

   2. Configuración de la tabla de dimensiones

      1. Desde el panel izquierdo, arrastre el nodo Input/Dimension Table (MySQL) hacia el lienzo de la derecha.

      2. Haga clic en el nodo Input/Dimension Table MySQL-2 del lienzo y configure la base de datos de origen.

      3. En la pestaña Node Configuration, establezca los siguientes parámetros.

         Parámetro	Descripción
         Data Source Name	Para facilitar su identificación, se recomienda usar un nombre descriptivo y no único.
         Region	Seleccione la región de la base de datos de origen.
         Instances	Seleccione el nombre de la instancia que aloja la base de datos de origen. También puede hacer clic en Create Instance para crear una nueva instancia. Para obtener más información, consulte Databases supported by DMS.
         Node Type	Seleccione el tipo de tabla de origen. En este ejemplo, seleccione Dimension Tables.
         Select Databases and Tables	Seleccione las bases de datos y tablas de origen que desea transformar.

      4. Una vez configurado el nodo, se muestra la pestaña Output Fields. En la columna Column Name, seleccione los campos obligatorios.

      Nota

      La configuración finaliza cuando desaparece el ícono de signo de exclamación del lado derecho del nodo de la base de datos de origen.

Paso 2: Configurar el nodo Table Join

1. En el panel izquierdo, arrastre el nodo Table Join hacia un área vacía del lienzo.

2. Coloque el puntero sobre los nodos configurados de la tabla de streaming y la tabla de dimensiones. Haga clic en el círculo de cada nodo y arrastre una línea de conexión hasta el nodo Table Join-1.

3. Haga clic en el nodo Table Join-1 del lienzo para configurarlo.

   1. En la pestaña Node Settings, establezca los siguientes parámetros.

      Sección	Parámetro	Descripción
      Conversion Name	Enter Transformation Name	Para facilitar su identificación, se recomienda usar un nombre descriptivo y no único.
      JOIN Settings	Left Table in JOIN Clause	Seleccione la tabla que se colocará en el lado izquierdo del operador JOIN. Esta tabla actúa como tabla principal. En este ejemplo, seleccione la tabla de streaming.
      	Temporal Join Time Attribute (Regular joins apply if not selected)	Seleccione el atributo de tiempo de la tabla de streaming que se usará para unirla con la tabla temporal. Si no selecciona ningún atributo, DTS ejecuta una unión normal. En este ejemplo, seleccione Based on Processing Time. Nota Una tabla temporal, también conocida como tabla dinámica, es una vista parametrizada de una tabla que registra los cambios de datos a lo largo del tiempo. Las tablas temporales incluyen tablas versionadas, que muestran versiones históricas de los datos, y tablas normales, que solo muestran la versión más reciente. Una unión temporal requiere que la tabla de streaming tenga un atributo de tiempo y que la tabla derecha tenga una clave principal. Si la tabla derecha es una tabla de dimensiones, la Join condition debe incluir la clave principal de dicha tabla. Based on Event Time Watermark: usa el tiempo de evento de la tabla de streaming para unirla con la versión correspondiente en la tabla versionada. Based on Processing Time: usa el tiempo de procesamiento de la tabla de streaming para unirla con la versión más reciente de la tabla normal.
      	Select JOIN Operation	Seleccione una operación de unión. En este ejemplo, seleccione Inner Join. Inner Join: devuelve los registros que tienen valores coincidentes en ambas tablas. Left Join: devuelve todos los registros de la tabla izquierda y los registros coincidentes de la tabla derecha. Right Join: devuelve todos los registros de la tabla derecha y los registros coincidentes de la tabla izquierda.
      JOIN Condition	+ Add condition	Haga clic en + Add Condition para seleccionar los campos de la condición de unión. Nota El campo a la izquierda del signo igual (=) pertenece a la tabla izquierda de la unión, y el campo a la derecha pertenece a la tabla derecha.

4. Después de configurar la condición de unión, haga clic en la pestaña Output Fields. En la columna Column Name, seleccione los campos obligatorios.

Paso 3: Configurar el nodo Table Record Filter

1. En el panel izquierdo, arrastre el nodo Table Record Filter hacia un área vacía del lienzo.

2. Coloque el puntero sobre el nodo Table Join-1 configurado. Haga clic en el círculo y arrastre una línea de conexión desde el nodo Table Join-1 hasta el nodo Table Record Filter-1.

3. Haga clic en el nodo Table Record Filter-1 del lienzo para configurarlo.

   1. En la sección Conversion Name, introduzca un nombre para la transformación.

      Nota

      Para facilitar su identificación, se recomienda usar un nombre descriptivo y no único.

   2. En la sección WHERE Condition, configure la condición WHERE mediante uno de los siguientes métodos:

      • Introduzca directamente la condición WHERE. Por ejemplo, introduzca total_price > 3000.00 para filtrar los registros donde el valor de total_price de la tabla unida sea mayor a 3000.00.

      • Haga clic en las opciones de las secciones Input Fields u Operator para construir la condición WHERE.

Paso 4: Configurar la base de datos de destino

1. En el lado izquierdo de la página, arrastre el nodo output MySQL hacia el área vacía del lienzo en el lado derecho de la página.

2. Coloque el puntero sobre el nodo Table Record Filter-1 configurado. Haga clic en el círculo y arrastre una línea de conexión desde el nodo Table Record Filter-1 hasta el nodo Output MySQL-1.

3. Haga clic en el nodo Output MySQL-1 del lienzo para configurar la base de datos de destino.

   1. En la pestaña Node Configuration, establezca los siguientes parámetros.

      Parámetro	Descripción
      Data Source Name	Para facilitar su identificación, se recomienda usar un nombre descriptivo y no único.
      Region	Seleccione la región de la base de datos de destino. Nota Las tareas de ETL solo están disponibles en las siguientes regiones: China (Hangzhou), China (Shanghai), China (Qingdao), China (Beijing), China (Zhangjiakou), China (Shenzhen), China (Guangzhou) y China (Hong Kong).
      Instances	Seleccione el nombre de la instancia que aloja la base de datos de destino. También puede hacer clic en Create Instance para crear una nueva instancia. Para obtener más información, consulte Databases supported by DMS.
      Table Mapping	Seleccione la tabla de destino donde se almacenarán los datos transformados. En la sección Select Destination Table, haga clic en la tabla de destino.

4. En la columna Column Name, seleccione los parámetros obligatorios.

Paso 5: Ejecutar la comprobación previa e iniciar la tarea

1. Una vez completada la configuración, haga clic en Generate Flink SQL Validation. ETL genera y valida el código Flink SQL.

2. Cuando finalice la validación, haga clic en View ETL Validation Details para consultar la instrucción Flink SQL generada y los resultados de validación en un cuadro de diálogo. Después de confirmar que la información es correcta, haga clic en Close.

   Nota

   Si la validación falla, resuelva el problema con base en las causas que se muestran en los resultados.

3. Haga clic en Next: Save Task Settings and Precheck. La tarea de ETL solo puede iniciarse después de aprobar la comprobación previa. Si la comprobación falla, haga clic en View Details junto al elemento con errores, corrija el problema según la información proporcionada y vuelva a ejecutar la comprobación previa.

4. Cuando termine la comprobación previa, haga clic en Next: Purchase Instance en la parte inferior de la página.

5. En la página Purchase, seleccione una Instance Class y especifique las Compute Units (CUs). Luego, lea y seleccione Data Transmission Service (Pay-as-you-go) Service Terms y Service Terms for Public Preview.

   Nota

   Durante la versión preliminar pública, cada usuario puede crear y usar dos instancias de ETL sin costo.

6. Haga clic en Buy and Start para iniciar la tarea de ETL.

Resultados de la tarea

En este ejemplo, después de que se inicia la tarea de ETL (por ejemplo, el 1 de agosto), cuando un registro actualizado en la tabla de transacciones en tiempo real test_orders cumple la condición de filtro (total_price > 3000.00), dicho registro se sincroniza con la tabla de destino test_orders_new.