Data Transmission Service:Configurar una tarea ETL con Flink SQL

Contexto

• Antes de configurar una tarea ETL, familiarícese con los siguientes conceptos:

  • Input/Dimension Table es la base de datos de origen de la tarea ETL.

  • Output es la base de datos de destino donde ETL escribe los datos procesados.

• DTS ofrece una función ETL de streaming dentro del proceso de sincronización de datos. Esta función permite agregar componentes de transformación entre las bases de datos de origen y de destino. Dichos componentes aplican un amplio conjunto de transformaciones de datos y escriben los resultados en el destino en tiempo real. Por ejemplo, puede unir dos tablas de flujo para crear una tabla ancha y escribirla en la base de datos de destino. También puede agregar un campo nuevo a una tabla de origen, asignarle valores mediante una función y escribir el campo transformado en la base de datos de destino.

Requisitos previos

• Las tareas ETL solo se pueden crear en las siguientes regiones: China (Hangzhou), China (Shanghai), China (Qingdao), China (Beijing), China (Zhangjiakou), China (Shenzhen), China (Guangzhou) y China (Hong Kong).

• Los tipos de base de datos de origen compatibles incluyen MySQL, PolarDB for MySQL, Oracle, PostgreSQL, DB2 for iSeries (AS/400), DB2 for LUW, PolarDB-X (Formerly DRDS) (PolarDB-X 1.0), PolarDB PostgreSQL, MariaDB, PolarDB for Oracle, SQL Server y PolarDB-X 2.0.

• Los tipos de base de datos de destino compatibles incluyen MySQL, PolarDB for MySQL, Oracle, AnalyticDB for MySQL V3.0, PolarDB PostgreSQL, PostgreSQL, DB2 for LUW, DB2 for iSeries (AS/400), AnalyticDB for PostgreSQL, SQL Server, MariaDB, PolarDB-X (Formerly DRDS) (PolarDB-X 1.0), PolarDB for Oracle y Tablestore.

• La función ETL no admite la migración de esquemas. Debe crear por adelantado los esquemas de tabla correspondientes en la base de datos de destino, de acuerdo con las condiciones de transformación. Por ejemplo, si la tabla A contiene Campo 1, Campo 2 y Campo 3, y la tabla B contiene Campo 2, Campo 3 y Campo 4, y desea unir ambas tablas para generar como salida Campo 2 y Campo 3, debe crear la tabla C en la base de datos de destino con dichos campos.

• La función ETL no admite la sincronización completa de datos. Solo es posible aplicar transformaciones en tiempo real a los datos incrementales.

Notas de uso

• Todas las bases de datos de origen y de destino deben residir en la misma región.

• Todas las Stream Table deben provenir de la misma instancia.

• El nombre de la base de datos y el nombre de la tabla deben ser únicos.

• Actualmente no se admiten las tareas entre cuentas.

Procedimiento

1. Vaya a la página de la lista de tareas 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 ETL en el campo Data Flow Name y establezca Development Method en FlinkSQL.

3. Haga clic en Confirm.

4. En la sección Data Flow Information de la página Streaming ETL, agregue las bases de datos de origen y de destino.

   Parámetro	Descripción
   Region	Seleccione la región del origen de datos.
   Type	Seleccione el tipo de tabla. Al configurar una tabla de origen, seleccione Stream Table si el origen es una tabla de flujo (una tabla que cambia en tiempo real y se puede unir con una tabla de dimensiones para enriquecer los datos). Seleccione Dimension Tables si el origen es una tabla de dimensiones (una tabla que se actualiza con poca frecuencia y suele usarse para crear tablas anchas al unirla con datos en tiempo real). Al configurar una tabla de destino, seleccione Output.
   Database type	Seleccione el tipo de la base de datos de origen o de destino.
   Instance	Introduzca el nombre o el ID para buscar y seleccionar las instancias de origen y de destino. Importante Primero debe agregar las instancias de origen y de destino en Data Management (DMS). Para consultar las instrucciones, consulte Gestión de instancias.
   Database	Seleccione la base de datos de origen o de destino que contiene los datos que se van a transformar.
   Physical table	Seleccione la tabla de origen o de destino que contiene los datos que se van a transformar.
   Alias of physical table	Establezca un alias conciso y legible para la tabla de origen o de destino. Este alias ayuda a ETL a localizar la tabla correcta al ejecutar las sentencias SQL.

5. En la ventana de comandos SQL de la página Streaming ETL, agregue las sentencias SQL para configurar la tarea ETL.

   Este ejemplo utiliza las siguientes sentencias SQL para unir la tabla de flujo test_orders con la tabla de dimensiones product e insertar el resultado en la tabla de destino test_orders_new.

   Importante

   Las sentencias SQL deben separarse con punto y coma (;).

   CREATE TABLE `etltest_test_orders` (
     `order_id` BIGINT,
     `user_id` BIGINT,
     `product_id` BIGINT,
     `total_price` DECIMAL(15,2),
     `order_date` TIMESTAMP(6),
     `dts_etl_schema_db_table` STRING,
     `dts_etl_db_log_time` BIGINT,
     `pt` AS PROCTIME(),
     WATERMARK FOR `order_date` AS `order_date` - INTERVAL '5' SECOND
   ) WITH (
     'streamType'= 'append',
     'alias'= 'test_orders',
     'vertexType'= 'stream'
   );
   CREATE TABLE `etltest_product` (
     `product_id` BIGINT,
     `product_name` STRING,
     `product_price` DECIMAL(15,2)
   ) WITH (
     'alias'= 'product',
     'vertexType'= 'lookup'
   );
   CREATE VIEW `etltest_test_orders_JOIN_etltest_product` AS
   SELECT
     `etltest_test_orders`.`order_id` AS `order_id`,
     `etltest_test_orders`.`user_id` AS `user_id`,
     `etltest_test_orders`.`product_id` AS `product_id`,
     `etltest_test_orders`.`total_price` AS `total_price`,
     `etltest_test_orders`.`order_date` AS `order_date`,
     `etltest_test_orders`.`dts_etl_schema_db_table` AS `dts_etl_schema_db_table`,
     `etltest_test_orders`.`dts_etl_db_log_time` AS `dts_etl_db_log_time`,
     `etltest_product`.`product_id` AS `product_id_0001011101`,
     `etltest_product`.`product_name` AS `product_name`,
     `etltest_product`.`product_price` AS `product_price`
   FROM `etltest_test_orders` LEFT JOIN `etltest_product` FOR SYSTEM_TIME AS OF `etltest_test_orders`.`pt` ON etltest_test_orders.product_id = etltest_product.product_id
   ;
   CREATE TABLE `test_orders_new` (
     `order_id` BIGINT,
     `user_id` BIGINT,
     `product_id` BIGINT,
     `total_price` DECIMAL(15,2),
     `order_date` TIMESTAMP(6),
     `product_name` STRING,
     `product_price` DECIMAL(15,2)
   ) WITH (
     'alias'= 'test_orders_new',
     'vertexType'= 'sink'
   );
   INSERT INTO `test_orders_new` (
     `order_id`,
     `user_id`,
     `product_id`,
     `total_price`,
     `order_date`,
     `product_name`,
     `product_price`
   )
   SELECT
     `etltest_test_orders_JOIN_etltest_product`.`order_id`,
     `etltest_test_orders_JOIN_etltest_product`.`user_id`,
     `etltest_test_orders_JOIN_etltest_product`.`product_id`,
     `etltest_test_orders_JOIN_etltest_product`.`total_price`,
     `etltest_test_orders_JOIN_etltest_product`.`order_date`,
     `etltest_test_orders_JOIN_etltest_product`.`product_name`,
     `etltest_test_orders_JOIN_etltest_product`.`product_price`
   FROM `etltest_test_orders_JOIN_etltest_product`;

   Tipo	Descripción
   Información de las tablas de origen y de destino	Utilice la sentencia CREATE TABLE para definir las tablas de origen y de destino. En la cláusula WITH de una sentencia SQL se pueden establecer tres parámetros: streamType, alias y vertexType. Para una tabla de flujo, los tres son obligatorios. En las tablas de dimensiones y de salida, solo se requieren alias y vertexType. streamType: el tipo de flujo. Durante el procesamiento de datos, ETL convierte un flujo en una tabla dinámica. Sobre esta tabla dinámica se ejecutan consultas continuas, y las operaciones INSERT, UPDATE y DELETE la modifican constantemente para generar una nueva tabla dinámica. Cuando esta última se escribe en la base de datos de destino, DTS la vuelve a convertir en un flujo. Este parámetro es obligatorio para codificar los cambios cuando la tabla dinámica se transforma en flujo. Upsert: un flujo de tipo upsert. Los datos de la tabla dinámica se pueden modificar mediante operaciones INSERT, UPDATE y DELETE. Al convertirse en flujo, las operaciones INSERT y UPDATE se codifican como mensajes upsert, mientras que las operaciones DELETE se codifican como mensajes de eliminación. Nota Esta codificación requiere que la tabla dinámica tenga una clave única, la cual puede ser compuesta. append: un flujo de solo anexado. Los datos de la tabla dinámica únicamente se pueden modificar mediante operaciones INSERT. Al convertirse en flujo, solo se emiten las filas insertadas. alias: el valor de Alias of Physical Table que estableció al configurar las bases de datos de origen y de destino en Step 4. vertexType: el tipo de tabla. stream: una tabla de flujo. lookup: una tabla de dimensiones. sink: una tabla de destino.
   Lógica de transformación de datos	Utilice la sentencia CREATE VIEW para definir la lógica de transformación de datos.
   Información de la tabla de destino transformada	Utilice la sentencia INSERT INTO para escribir los datos en la tabla de destino transformada.

6. Después de configurar la base de datos de origen, la base de datos de destino y las sentencias SQL, haga clic en Generate Flink SQL Validation.

   Nota

   • También puede hacer clic en Publish para ejecutar directamente la validación y la comprobación previa.

   • Si la validación de Flink SQL se completa correctamente, haga clic en para consultar los detalles de la validación.

   • Si la validación de Flink SQL termina con errores, haga clic en , corrija la sentencia SQL según el mensaje indicado y vuelva a ejecutar la validación de Flink SQL.

7. Una vez que la validación de Flink SQL finalice correctamente, haga clic en Publish para iniciar la comprobación previa.

8. Cuando la comprobación previa se supere, haga clic en Next: Purchase (Free).

   Nota

   Si la comprobación previa termina con errores, haga clic en View Details junto al elemento fallido, resuelva el problema con base en los detalles proporcionados y vuelva a ejecutar la comprobación previa.

9. En la página Purchase, seleccione una Instance Class. La cantidad de Compute Units (CUs) es fija y corresponde a 2 durante la versión preliminar pública. Lea y active las casillas de verificación de 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 utilizar dos instancias ETL de forma gratuita.

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