Uso de un trabajo de Flink de DLI para sincronizar datos de MySQL con un clúster de DWS en tiempo real

Preparativos

• Ha creado un ID de Huawei y habilitado los servicios de Huawei Cloud. La cuenta no puede estar en mora o congelada.
• Ha creado una VPC y una subred. Para obtener más información, consulte Creación de una VPC.

Paso 1: Preparar datos de MySQL

1. Inicie sesión en la consola de RDS y compre una instancia de base de datos de RDS. Configure los parámetros clave que se enumeran en Tabla 1 y conserve los valores predeterminados de los demás parámetros. Para obtener más información, consulte Documentación de RDS.

   Tabla 1 Parámetros de RDS

   Parámetro	Valor
   Billing Mode	Pago por uso
   Region	CN-Hong Kong
   DB Instance Name	rds-demo
   DB Engine	MySQL
   DB Engine Version	5.7 o versión posterior
   Database Port	3306

2. Conéctese a la instancia de RDS y cree una instancia denominada mys_data.

   1	CREATE DATABASE mys_data;

3. Cambie a la nueva base de datos mys_data y ejecute el siguiente comando para crear la tabla mys_orders:

   1 2 3 4 5 6 7 8

   CREATE TABLE mys_data.mys_order ( order_id VARCHAR(12), order_channel VARCHAR(32), order_time DATETIME, cust_code VARCHAR(6), pay_amount DOUBLE, real_pay DOUBLE, PRIMARY KEY (order_id) );

4. Inserte datos en la tabla.

   1 2	INSERT INTO mys_data.mys_order VALUES ('202306270001', 'webShop', TIMESTAMP('2023-06-27 10:00:00'), 'CUST1', 1000, 1000); INSERT INTO mys_data.mys_order VALUES ('202306270002', 'webShop', TIMESTAMP('2023-06-27 11:00:00'), 'CUST2', 5000, 5000);

5. Verifique si los datos se han insertado.

   1	SELECT * FROM mys_data.mys_order;

Paso 2: Crear un clúster de DWS

1. Creación de un clúster. Para garantizar la conectividad de red, seleccione la misma región y la misma VPC que las de la instancia de RDS. En esta práctica, seleccione CN-Hong Kong. La VPC debe ser la misma que la creada para RDS.
2. Inicie sesión en la consola de DWS. Seleccione Dedicated Clusters > Clusters. Busque el clúster de destino y haga clic en Log In en la columna Operation. La información de inicio de sesión es la siguiente:
   • Nombre del origen de datos: dws-demo
   • Clúster: el clúster de DWS creado.
   • Base de datos: gaussdb
   • Nombre de usuario: dbadmin
   • Contraseña: contraseña establecida cuando se crea el clúster de DWS

3. Seleccione Remember Password, haga clic en Test Connection y espere hasta que la conexión sea exitosa.
4. Copie las siguientes sentencias SQL. En la ventana SQL, haga clic en Execute SQL para crear un esquema denominado dws_data.

   1	CREATE SCHEMA dws_data;

5. Cree la tabla dws_order en el nuevo esquema.

   1 2 3 4 5 6 7

   CREATE TABLE dws_data.dws_order ( order_id VARCHAR(12), order_channel VARCHAR(32), order_time TIMESTAMP, cust_code VARCHAR(6), pay_amount DOUBLE PRECISION, real_pay DOUBLE PRECISION );

6. Query data. La tabla actual está vacía.

   1	SELECT * FROM dws_data.dws_order;

Paso 3: Crear un grupo de recursos elásticos de DLI y una cola

1. Inicie sesión en la Consola de DLI de Huawei Cloud.
2. En el panel de navegación de la izquierda, seleccione Resources > Resource Pool.
3. Haga clic en Buy Resource Pool en la esquina superior derecha, establezca los siguientes parámetros y conserve los valores predeterminados para otros parámetros que no se describen en la tabla.

   Tabla 2 Parámetros

   Parámetro	Valor
   Billing Mode	Pago por uso
   Region	CN-Hong Kong
   Name	dli_dws
   Specifications	Estándar
   CIDR Block	172.16.0.0/18, que debe estar en un segmento de red diferente al de MySQL y DWS. Por ejemplo, si MySQL y DWS están en el segmento de red 192.168.x.x, seleccione 172.16.x.x para DLI.

4. Haga clic en Buy y haga clic en Submit.

   Una vez creado el grupo de recursos, siga con el siguiente paso.

5. En la página de grupo de recursos elásticos, busque la fila que contiene el grupo de recursos creado, haga clic en Add Queue en la columna Operation y configure los siguientes parámetros. Conserve los valores predeterminados para otros parámetros que no se describen en la tabla.

   Tabla 3 Adición de una cola

   Parámetro	Valor
   Name	dli_dws
   Type	Cola de uso general

6. Haga clic en Next y en OK. La cola se ha creado.

Paso 4: Crear una conexión de origen de datos mejorada

1. Actualice los permisos de la delegación de DLI.
   1. Vuelva a la consola de DLI y seleccione Global Configuration > Service Authorization a la izquierda.
   2. Seleccione DLI UserInfo Agency Access, DLI Datasource Connections Agency Access y DLI Notification Agency Access.
   3. Haga clic en Update. Haga clic en OK.
   Figura 1 Actualización de permisos de delegación de DLI
   

2. En el grupo de seguridad de RDS, permita el segmento de red donde se encuentra la cola de DLI.
   1. En el panel de navegación de la izquierda, seleccione Resources > Queue Management y registre el bloque CIDR de dli_dws.
      Figura 2 Bloque CIDR de una cola de DLI
      
   2. Vaya a la consola de RDS, seleccione Instances en el panel de navegación y haga clic en el nombre de la instancia de RDS creada.
   3. Registre el valor de Floating IP Address en el área Connectivity, que se utilizará en la prueba de conectividad posterior.
   4. Haga clic en Manage junto al grupo de seguridad en Connectivity.
      Figura 3 Grupo de seguridad de RDS
      

   5. En la lista de grupos de seguridad que se muestra, haga clic en el nombre del grupo de seguridad para ir a la página de configuración del grupo de seguridad.
   6. Elija Inbound Rules > Add Rule, como se muestra en la siguiente figura. Agregue el segmento de red de la cola de DLI. En este ejemplo, el segmento de red es 172.16.0.0/18. Asegúrese de que el segmento de red sea el mismo que se ingresó en Paso 3: Crear un grupo de recursos elásticos de DLI y una cola.
      Figura 4 Adición de una regla al grupo de seguridad de RDS
      

   7. Haga clic en OK.

3. Vuelva a la consola de DLI, haga clic en Datasource Connections a la izquierda, seleccione Enhanced y haga clic en Create.
4. Configure los siguientes parámetros. Conserve los valores predeterminados para otros parámetros que no se describen en la tabla.

   Tabla 4 Conexión de DLI a RDS

   Parámetro	Valor
   Connection Name	dli_rds
   Resource Pool	Seleccione el grupo de recursos elásticos de DLI creado.
   VPC	Seleccione la VPC donde se encuentra RDS.
   Subnet	Seleccione la subred donde se encuentra RDS.
   Other parameters	Conserve los valores por defecto.

   Figura 5 Creación de una conexión de fuente de datos
   

5. Haga clic en OK. Espere hasta que se cree la conexión de RDS.
6. Pruebe la conectividad entre DLI y RDS.
   1. Elija Resources > Queue Management a la izquierda y elija More > Test Address Connectivity a la derecha de dli_dws.
   2. Ingrese la dirección IP privada de RDS registrada en 2.c y el puerto 3306 en el cuadro de dirección.
   3. Haga clic en Test para verificar que DLI se haya conectado correctamente a RDS.
      Figura 6 Prueba de la conexión entre RDS y DLI
      

7. Si la conexión falla, realice las siguientes operaciones:

   Asegúrese de que el grupo de seguridad al que está asociado RDS permita el acceso desde el bloque CIDR donde se encuentra el grupo de recursos de DLI y de que la delegación se haya actualizado.

   1. Inicie sesión en la consola de DLI y haga clic en Datasource Connections. Busque la conexión creada y haga clic en More en la columna Operation y seleccione Unbind Resource Pool.
   2. Deseleccione el grupo de recursos elásticos existente y haga clic en OK.
   3. Haga clic en More en la columna Operation y seleccione Bind Resource Pool.
   4. Seleccione el grupo de recursos elásticos creado dli_dws y haga clic en OK.
   5. Espere unos 2 minutos y vuelva a probar la conexión.
   6. Si la conexión sigue fallando, rectifique el fallo consultando ¿Qué hago si la conexión de la fuente de datos se crea correctamente, pero la prueba de conectividad de red falla?

8. Pruebe la conectividad entre DLI y DWS.
   1. Inicie sesión en la consola de DWS. En el panel de navegación, seleccione Dedicated Clusters > Clusters. Haga clic en el nombre del clúster para ver los detalles del clúster.
   2. Como se muestra en la siguiente figura, registre la dirección IP privada y el número de puerto del clúster de DWS para su uso futuro.
      Figura 7 Dirección IP interna de DWS
      
   3. Haga clic en el nombre del grupo de seguridad.
      Figura 8 Grupo de seguridad de DWS
      

   4. Elija Inbound Rules > Add Rule, como se muestra en la siguiente figura. Agregue el segmento de red de la cola de DLI. En este ejemplo, el segmento de red es 172.16.0.0/18. Asegúrese de que el segmento de red sea el mismo que se ingresó en 4.

      Figura 9 Adición de una regla al grupo de seguridad de DWS
      
   5. Haga clic en OK.
   6. Cambie a la consola de DLI, elija Resources > Queue Management a la izquierda y haga clic en More > Test Address Connectivity a la derecha de dli_dws.
   7. En el cuadro de dirección, ingrese la dirección IP privada y el número de puerto del clúster de DWS.
   8. Haga clic en Test para verificar que DLI se haya conectado correctamente a DWS.
      Figura 10 Prueba de conectividad de DWS
      

Paso 5: Preparar la herramienta dws-connector-flink para interconectar DWS con Flink

dws-connector-flink es una herramienta para interconectarse con Flink basada en las API JDBC de DWS. Durante la configuración del trabajo de DLI, esta herramienta y sus dependencias se almacenan en el directorio de carga de clases de Flink para importar mejor los trabajos de Flink a DWS.

1. Vaya a https://mvnrepository.com/artifact/com.huaweicloud.dws usando un navegador.
2. En la lista de software, seleccione Flink 1.15. En esta práctica, se selecciona DWS Connector Flink SQL 1 15.

   

3. Seleccione la rama más reciente. La rama real está sujeta a la nueva rama publicada en el sitio web oficial.

   

4. Haga clic en el ícono jar para descargar el archivo.

   

5. Cree un bucket de OBS. En esta práctica, configure el nombre de bucket como obs-flink-dws y cargue el archivo dws-connector-flink-sql-1.15-2.12_2.0.0.r4.jar en el bucket de OBS. Asegúrese de que el bucket se encuentre en la misma región que DLI. En esta práctica, se utiliza la región CN-Hong Kong.

Paso 6: Crear un trabajo de Flink de DLI

1. Cree una política de delegación de OBS.
   1. Pase el cursor sobre el nombre de la cuenta en el extremo superior derecho de la consola y haga clic en Identity and Access Management.
   2. En el panel de navegación de la izquierda, seleccione Agencies y, a continuación, haga clic en Create Agency en el extremo superior derecho.
      • Agency Name: dli_ac_obs
      • Agency Type: Cloud service
      • Cloud Service: Data Lake Insight (DLI)
      • Validity Period: Unlimited
      Figura 11 Creación de una delegación de OBS
      

   3. Haga clic en OK y, a continuación, haga clic en Authorize.
   4. En la página que se muestra, haga clic en Create Policy.
   5. Configure la información de la política. Ingrese un nombre de política, por ejemplo, dli_ac_obs, y seleccione JSON.
   6. En el área Policy Content, pegue una política personalizada. Reemplace el nombre de bucket de OBS con el nombre de bucket real creado en 5.

       {
          "Version": "1.1",
          "Statement": [
              {
                  "Effect": "Allow",
                  "Action": [
                      "obs:object:GetObject",
                      "obs:object:DeleteObjectVersion",
                      "obs:bucket:GetBucketLocation",
                      "obs:bucket:GetLifecycleConfiguration",
                      "obs:object:AbortMultipartUpload",
                      "obs:object:DeleteObject",
                      "obs:bucket:GetBucketLogging",
                      "obs:bucket:HeadBucket",
                      "obs:object:PutObject",
                      "obs:object:GetObjectVersionAcl",
                      "obs:bucket:GetBucketAcl",
                      "obs:bucket:GetBucketVersioning",
                      "obs:bucket:GetBucketStoragePolicy",
                      "obs:bucket:ListBucketMultipartUploads",
                      "obs:object:ListMultipartUploadParts",
                      "obs:bucket:ListBucketVersions",
                      "obs:bucket:ListBucket",
                      "obs:object:GetObjectVersion",
                      "obs:object:GetObjectAcl",
                      "obs:bucket:GetBucketPolicy",
                      "obs:bucket:GetBucketStorage"
                  ],
                  "Resource": [
                      "OBS:*:*:object:*",
                      "OBS:*:*:bucket:OBS bucket name"
                  ]
              },
              {
                  "Effect": "Allow",
                  "Action": [
                      "obs:bucket:ListAllMyBuckets"
                  ]
              }
          ]
      }

   7. Haga clic en Next.
   8. Seleccione la política personalizada creada.
   9. Haga clic en Next. Seleccione All resources.
   10. Haga clic en OK.

       La autorización tarda entre 15 y 30 minutos en surtir efecto.

2. Vuelva a la consola de DLI, seleccione Job Management > Flink Jobs a la izquierda y haga clic en Create Job en la esquina superior derecha.
3. Establezca Type en Flink OpenSource SQL y Name en rds-dws.
   Figura 12 Creación de un trabajo
   

4. Haga clic en OK. Se mostrará la página para editar el trabajo.
5. Establezca los siguientes parámetros a la derecha de la página. Conserve los valores predeterminados para otros parámetros que no se describen en la tabla.

   Tabla 5 Parámetros de trabajo de Flink

   Parámetro	Valor
   Queue	dli_dws creado en 4.
   Flink Version	1.15 o versión posterior. (La versión real está sujeta a la GUI).
   UDF Jar	Archivo JAR en el bucket de OBS creado en 5.
   Agency	Delegación creada en 1.
   OBS Bucket	Bucket creado en 5.
   Save Job Log	Marque la casilla.
   Enable Checkpointing	Marque la casilla.
   Other parameters	Conserve el valor predeterminado.

   

6. Copie el siguiente código SQL en la ventana de código SQL de la izquierda.

   Para obtener la dirección IP interna de la base de datos RDS, consulte 2.c. Para obtener la dirección IP interna del clúster de DWS, consulte 8.b. Cambie la contraseña del usuario root de la base de datos RDS y la contraseña del usuario dbadmin de DWS.

   A continuación se describen los parámetros comunes del código de trabajo de Flink SQL:

   • connector: tipo de conector del origen de datos. Para MySQL, establezca este parámetro en mysql-cdc. Para DWS, establezca este parámetro en gaussdb. Para obtener más información, consulte Conectores.
   • driver: nombre del controlador JDBC de DWS. El valor se puede fijar en com.huawei.gauss200.jdbc.Driver.
   • write.mode: modo de importación de datos. El valor puede ser copy, insert o upsert.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

   CREATE TABLE mys_order ( order_id STRING, order_channel STRING, order_time TIMESTAMP, cust_code STRING, pay_amount DOUBLE, real_pay DOUBLE, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'Private IP address of the RDS DB instance', 'port' = '3306', 'username' = 'root', 'password' = 'Password of user root of the RDS DB instance', 'database-name' = 'mys_data', 'table-name' = 'mys_order' ); CREATE TABLE dws_order ( order_id STRING, order_channel STRING, order_time TIMESTAMP, cust_code STRING, pay_amount DOUBLE, real_pay DOUBLE, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'gaussdb', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://DWS cluster private IP address:8000/gaussdb', 'table-name' = 'dws_data.dws_order', 'username' = 'dbadmin', 'password' = 'Password of DWS user dbadmin', 'write.mode' = 'insert' ); INSERT INTO dws_order SELECT * FROM mys_order;

7. Haga clic en Format y en Save.
   

   Haga clic en Format para formatear el código SQL. De lo contrario, se podría introducir caracteres nulos nuevos al copiar y pegar el código, lo que provocaría fallas en la ejecución del trabajo.

   Figura 13 Parámetros de trabajo de Flink
   

8. Vuelva a la página de inicio de la consola de DLI y seleccione Job Management > Flink Jobs a la izquierda.
9. Haga clic en Start a la derecha del nombre del trabajo rds-dws y en Start Now.

   Espere aproximadamente 1 minuto y actualice la página. Si el estado es Running, el trabajo se ejecutará.

   Figura 14 Ejecución exitosa
   

Paso 7: Verificar la sincronización de datos

1. Vuelva a la ventana SQL de la base de datos de DWS. Si se agota el tiempo de espera de la conexión, realice las siguientes operaciones para volver a iniciar sesión:
   1. Vaya a la consola de DWS.
   2. En el panel de navegación de la izquierda, seleccione Dedicated Clusters > Clusters y haga clic en Log In a la derecha de dws-demo.

2. Verifique si dos filas de datos de la tabla MySQL se han sincronizado con DWS.

   1	SELECT * FROM dws_data.dws_order;

   Figura 15 Resultado de la consulta
   

3. Cambie a la página RDS for MySQL y ejecute las siguientes sentencias para insertar tres nuevos registros de datos:

   1 2 3

   INSERT INTO mys_data.mys_order VALUES ('202403090003', 'webShop', TIMESTAMP('2024-03-09 13:00:00'), 'CUST1', 2000, 2000); INSERT INTO mys_data.mys_order VALUES ('202403090004', 'webShop', TIMESTAMP('2024-03-09 14:00:00'), 'CUST2', 3000, 3000); INSERT INTO mys_data.mys_order VALUES ('202403100004', 'webShop', TIMESTAMP('2024-03-10 10:00:00'), 'CUST3', 6000, 6000);

   Figura 16 Nuevos datos de MySQL
   

4. Vuelva a la ventana SQL de DWS y ejecute de nuevo la siguiente sentencia SQL. El resultado devuelto muestra que los datos de MySQL se han sincronizado con DWS en tiempo real.

   1	SELECT * FROM dws_data.dws_order;

   Figura 17 Sincronización de datos en tiempo real
   

Más información

Al almacenar información de autenticación para una fuente de datos directamente en el script de trabajo para el desarrollo de Flink entre fuentes, se puede exponer la contraseña. Para mejorar la seguridad, utilice la función de autenticación de la fuente de datos de DLI en lugar de especificar directamente los nombres de usuario y las contraseñas de MySQL y DWS en los scripts de trabajo.