E-MapReduce:Sincronización de datos de MySQL a StarRocks mediante CTAS y CDAS

Contexto

La instrucción CTAS (CREATE TABLE AS) crea automáticamente una tabla en StarRocks con el mismo esquema que la tabla de origen en MySQL. Además, sincroniza los datos y replica los cambios de esquema del origen al destino en tiempo real. Esto simplifica la creación de las tablas de destino y facilita mantener la consistencia del esquema.

Una instrucción CTAS sincroniza los datos en tiempo real y propaga los cambios de esquema de la tabla de origen a la de destino.

Requisitos previos

• Activó Realtime Compute for Apache Flink (Fully Managed) y creó un clúster de Flink. Para obtener más información, consulte Activate Realtime Compute for Apache Flink (Fully Managed) y Getting started with a Flink SQL deployment.

• Creó un clúster de StarRocks. Para obtener más información, consulte Create a StarRocks cluster.

• Creó una instancia de ApsaraDB RDS for MySQL. Para obtener más información, consulte Create an ApsaraDB RDS for MySQL instance.

Limitaciones

• El clúster de Flink, el clúster de StarRocks y la instancia de ApsaraDB RDS for MySQL deben estar en la misma VPC.

• La instancia de ApsaraDB RDS for MySQL debe ser versión 5.7 o posterior.

• El clúster de StarRocks debe tener habilitado el acceso a Internet.

• La versión de Flink en su clúster debe ser 1.15-vvr-6.0.3 o posterior.

Paso 1: Preparar los datos de prueba

1. Cree una base de datos y una cuenta de prueba. Para obtener más información, consulte Create databases and accounts for an ApsaraDB RDS for MySQL instance.

   Después de crear la base de datos y la cuenta, otorgue permisos de lectura y escritura a la cuenta de prueba.

   Nota

   En este tema, la base de datos se llama test_cdc y la cuenta se llama test.

2. Utilice la cuenta de prueba para conectarse a la instancia de MySQL. Para obtener más información, consulte Use DMS to log on to an ApsaraDB RDS for MySQL instance.

3. Ejecute los siguientes comandos en MySQL para crear una tabla de datos.

   use test_cdc;
   
   CREATE TABLE IF NOT EXISTS `runoob_tbl`(
      `runoob_id` INT UNSIGNED AUTO_INCREMENT,
      `runoob_title` VARCHAR(100) NOT NULL,
      `runoob_author` VARCHAR(40) NOT NULL,
      `submission_date` DATE,
      `add_col` int DEFAULT NULL,
      PRIMARY KEY ( `runoob_id` )
   )ENGINE=InnoDB DEFAULT CHARSET=utf8;
   
   INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2022-06-22 17:13:44',3)

4. Inicie sesión en el clúster de StarRocks mediante SSH. Para obtener más información, consulte Log on to a cluster.

5. Ejecute el siguiente comando para conectarse al clúster de StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

6. Ejecute los siguientes comandos para crear un usuario y otorgarle permisos.

   CREATE DATABASE test_cdc;
   CREATE USER 'test' IDENTIFIED by '123456';
   GRANT CREATE TABLE ON DATABASE test_cdc TO test;

Paso 2: Crear catálogos

• Catálogo de MySQL

  • Ejemplo

    CREATE CATALOG mysql WITH (
      'type' = 'mysql',
      'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
      'port' = '3306',
      'username' = 'emr-test',
      'password' = '123456',
      'default-database' = 'test_cdc'
    );

  • Parámetros

    Parámetro	Descripción
    type	El tipo de catálogo. Establezca el valor en mysql.
    hostname	El endpoint interno de la instancia de ApsaraDB RDS for MySQL. Puede copiarlo desde la página Database Connection de la instancia en la consola de ApsaraDB RDS. Por ejemplo, rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com.
    port	El número de puerto del servicio de base de datos MySQL. El valor predeterminado es 3306.
    username	El nombre de usuario para acceder al servicio de base de datos MySQL. Utilice el nombre de usuario de la cuenta creada en Step 1: Prepare test data. Este ejemplo usa test.
    password	La contraseña para acceder al servicio de base de datos MySQL. Utilice la contraseña de la cuenta creada en Step 1: Prepare test data.
    default-database	El nombre de la base de datos MySQL predeterminada. Utilice el nombre de la base de datos de Step 1: Prepare test data. Este ejemplo usa test_cdc.

• Catálogo de StarRocks

  • Ejemplo

    CREATE CATALOG sr  WITH (
      'type' = 'starrocks',
      'endpoint' = '172.16.**.**:9030',
      'username' = 'test',
      'password' = '123456',
      'dbname' = 'test_cdc'
    );

  • Parámetros

    Parámetro	Descripción
    type	El tipo de catálogo. Establezca el valor en starrocks.
    endpoint	La dirección IP y el puerto del frontend (FE) de StarRocks.
    username	El nombre de usuario para acceder al clúster de StarRocks. Utilice el nombre de usuario de la cuenta creada en Step 1: Prepare test data. Este ejemplo usa test.
    password	La contraseña del servicio de base de datos de StarRocks. Utilice la contraseña de la cuenta creada en Step 1: Prepare test data.
    dbname	El nombre de la base de datos de StarRocks. Utilice el nombre de la base de datos de Step 1: Prepare test data. Este ejemplo usa test_cdc.

Paso 3: Crear y publicar un despliegue

1. En la página Draft Editor de la consola de Realtime Compute for Apache Flink, escriba una instrucción CTAS.

   A continuación se muestran tres ejemplos de instrucciones CTAS.

   • Semántica at-least-once: utilice la opción sink.buffer-flush.interval-ms para configurar el intervalo con el que se escriben los datos en StarRocks. Esta opción reduce la latencia y el uso de memoria.

     /*
           At-least-once semantics
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl_sr with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • Semántica exactly-once: debe configurar un intervalo de checkpoint. Esto evita la pérdida y duplicación de datos durante fallas, aunque la visibilidad de los datos depende de dicho intervalo. Para obtener más información, consulte Checkpointing.

     /*
           Exactly-once semantics.
     */
     set 'execution.checkpointing.interval' = '1 min';
     set 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
     set 'execution.checkpointing.timeout' = '10 min';
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '123456',
     'sink.semantic' = 'exactly-once',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • Modo simple: no necesita definir campos al crear la tabla, ya que el esquema se copia de MySQL. Sin embargo, no es posible crear particiones. Para usar particiones, debe utilizar el modo normal.

     /*
           The two preceding examples use normal mode. This example demonstrates simple mode.
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl1 with (
     'starrocks.create.table.properties'='buckets 8',
     'starrocks.create.table.mode'='simple',
      'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'emr-test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   Tabla 1. Parámetros WITH

   Parámetro	Obligatorio	Descripción
   starrocks.create.table.properties	Sí	Las definiciones de sufijo en la instrucción CREATE TABLE de StarRocks, excluyendo las definiciones de columnas. Algunos ejemplos son engine, key y buckets.
   database-name	Sí	El nombre de la base de datos de StarRocks. Este ejemplo usa test_cdc.
   jdbc-url	Sí	Se utiliza para ejecutar operaciones de consulta en StarRocks. Por ejemplo, jdbc:mysql://172.16..:9030. La parte 172.16.. corresponde a la dirección IP interna del clúster de StarRocks.
   load-url	Sí	La dirección IP y el puerto HTTP del frontend (FE) de StarRocks, con el formato Dirección IP interna del clúster de StarRocks:Puerto. Este tema utiliza el puerto 8030 como ejemplo. Seleccione un puerto según la versión de su clúster: 18030: para EMR V5.9.0 o posterior y EMR V3.43.0 o posterior. 8030: para EMR V5.8.0 o anterior y EMR V3.42.0 o anterior. Nota Para obtener más información sobre los puertos, consulte UI and ports.
   sink.semantic	No	Establézcalo en exactly-once para garantizar la semántica de consistencia de datos. El valor predeterminado es at-least-once.
   starrocks.create.table.mode	No	Valores compatibles: normal (predeterminado): debe proporcionar configuraciones completas, como engine, key y buckets, en la opción starrocks.create.table.properties. simple: el motor se establece en olap y el tipo de clave en primary key de forma predeterminada. La clave primaria se hereda de la tabla de MySQL. De manera predeterminada, la tabla se distribuye mediante un hash de todas las columnas de clave primaria, sin particiones. Debe especificar buckets en la opción starrocks.create.table.properties. Otras configuraciones, como properties, son opcionales.

   Nota

   • El parámetro sink.use.new-api se eliminó en las versiones de Flink 1.15-vvr-6.0.5 y posteriores. Si utiliza una versión anterior a 1.15-vvr-6.0.5, debe agregar 'sink.use.new-api'='false', a los parámetros WITH.

   • Para obtener información sobre otras configuraciones, consulte Carga continua de datos desde Apache Flink.

   Tabla 2. Parámetros OPTIONS

   Parámetro	Descripción
   connector	El tipo de conector. Establezca el valor en mysql-cdc.
   hostname	El endpoint interno de la instancia de ApsaraDB RDS for MySQL. Puede copiarlo desde la página Database Connection de la instancia en la consola de ApsaraDB RDS. Por ejemplo, rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com.
   port	El número de puerto del servicio de base de datos MySQL. El valor predeterminado es 3306.
   username	El nombre de usuario para acceder al servicio de base de datos MySQL. Utilice el nombre de usuario de la cuenta creada en Step 1: Prepare test data. Este ejemplo usa test.
   password	La contraseña para acceder al servicio de base de datos MySQL. Utilice la contraseña de la cuenta creada en Step 1: Prepare test data.
   table-name	El nombre de la tabla en StarRocks. Utilice el nombre de la tabla de Step 1: Prepare test data. Este ejemplo usa runoob_tbl.
   database-name	El nombre de la base de datos MySQL predeterminada. Utilice el nombre de la base de datos de Step 1: Prepare test data. Este ejemplo usa test_cdc.

2. En la sección Advanced de la página Draft Editor, seleccione la versión de Flink 1.15-vvr-6.0.3 o posterior.

3. Haga clic en online.

4. En la página Deployments, busque el despliegue correspondiente y haga clic en START en la columna Actions.

Paso 4: Demostración del escenario

Consulta de datos

1. Inicie sesión en el clúster de StarRocks mediante SSH. Para obtener más información, consulte Log on to a cluster.

2. Ejecute el siguiente comando para conectarse al clúster de StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

3. En la CLI de StarRocks, ejecute los siguientes comandos para consultar los datos de la tabla.

   use test_cdc;
   select * from runoob_tbl1;

   El resultado muestra que los datos de la tabla de MySQL se sincronizaron correctamente con StarRocks.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Consulta de datos insertados

1. En la consola SQL de la instancia de ApsaraDB RDS for MySQL, ejecute el siguiente comando para insertar datos.

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`)  values(1,'second','tom2','2022-06-23',1)

2. En la CLI de StarRocks, ejecute el siguiente comando para consultar los datos de la tabla.

   select * from runoob_tbl1;

   El resultado muestra que los datos se insertaron correctamente.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Sincronización de actualizaciones de datos

1. En la consola SQL de la instancia de ApsaraDB RDS for MySQL, ejecute el siguiente comando para actualizar datos específicos.

   update runoob_tbl set runoob_title= 'new' where runoob_id = 18

2. En la CLI de StarRocks, ejecute el siguiente comando para consultar los datos de la tabla.

   select * from runoob_tbl1;

   El resultado muestra que la actualización de datos se sincronizó correctamente.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | new          | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Sincronización de eliminaciones de datos

1. En la consola SQL de la instancia de ApsaraDB RDS for MySQL, ejecute el siguiente comando para eliminar datos específicos.

   DELETE FROM runoob_tbl WHERE runoob_id = 1

2. En la CLI de StarRocks, ejecute el siguiente comando para consultar los datos de la tabla.

   select * from runoob_tbl1;

   El resultado muestra que la eliminación de datos se sincronizó correctamente.

   +-----------+--------------+---------------+-----------------+---------+ 
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | 
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | new          | tom           | 2022-06-22      |       3 | 
   +-----------+--------------+---------------+-----------------+---------+

Agregar una columna que acepta valores nulos

1. En la consola SQL de la instancia de ApsaraDB RDS for MySQL, ejecute el siguiente comando para agregar una columna que acepta valores nulos.

   alter table `runoob_tbl` add COLUMN `add_col2` INT;

2. Ejecute el siguiente comando para insertar datos.

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`,`add_col2`)  values(1,'second','tom2','2022-06-23',1,2)

3. En la CLI de StarRocks, ejecute el siguiente comando para consultar los datos de la tabla.

   select * from runoob_tbl1;

   El resultado muestra que el cambio de esquema se aplicó correctamente.

   +-----------+--------------+---------------+-----------------+---------+----------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | add_col2 | 
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |        18 | new          | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |         1 | second       | tom2          | 2022-06-23      |       1 |      2   |
   |        18 | first        | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 

Introducción a CDAS

Una instrucción CDAS es una abreviatura sintáctica de CTAS. Permite sincronizar una base de datos completa de MySQL con StarRocks como un único despliegue de Flink. También puede usar la sintaxis including table para sincronizar solo un subconjunto de tablas.

CREATE DATABASE IF NOT EXISTS sr_db with (
  'starrocks.create.table.properties'=' buckets 8',
  'starrocks.create.table.mode'='simple',
  'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
  'load-url'='172.16.**.**:18030',
  'username'='test',
  'password' = '123456',
  'sink.buffer-flush.interval-ms' = '5000' ,
  'sink.properties.row_delimiter' = '\x02',
  'sink.properties.column_separator' = '\x01'
)
as DATABASE mysql.test_cdc including table  'tabl1','tbl2','tbl3'
/*+ OPTIONS (   'connector' = 'mysql-cdc',
'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
'port' = '3306',   
'username' = 'test',
'password' = '123456',
'database-name' = 'test_cdc' )*/;