E-MapReduce:Sincronizar dados do MySQL para o StarRocks usando CTAS e CDAS

Contexto

A instrução CTAS (CREATE TABLE AS) cria automaticamente uma tabela no StarRocks com esquema idêntico ao da tabela de origem no MySQL. Ela também sincroniza dados e replica alterações de esquema da origem para o destino em tempo real, simplificando a criação de tabelas de destino e a manutenção da consistência do esquema.

Uma instrução CTAS sincroniza dados em tempo real e propaga alterações de esquema da origem para a tabela de destino.

Pré-requisitos

• Ative o Realtime Compute for Apache Flink (Fully Managed) e crie um cluster Flink. Para mais informações, consulte Activate Realtime Compute for Apache Flink (Fully Managed) e Getting started with a Flink SQL deployment.

• Crie um cluster StarRocks. Para mais informações, consulte Create a StarRocks cluster.

• Crie uma instância ApsaraDB RDS for MySQL. Para mais informações, consulte Create an ApsaraDB RDS for MySQL instance.

Limitações

• O cluster Flink, o cluster StarRocks e a instância ApsaraDB RDS for MySQL devem estar na mesma VPC.

• A instância ApsaraDB RDS for MySQL deve ser versão 5,7 ou superior.

• O acesso à internet deve estar habilitado no cluster StarRocks.

• A versão do Flink no seu cluster deve ser 1.15-vvr-6.0.3 ou superior.

Etapa 1: Preparar dados de teste

1. Crie um banco de dados e uma conta de teste. Para mais informações, consulte Create databases and accounts for an ApsaraDB RDS for MySQL instance.

   Após criar o banco de dados e a conta, conceda permissões de leitura e gravação à conta de teste.

   Nota

   Neste tópico, o banco de dados chama-se test_cdc e a conta chama-se test.

2. Use a conta de teste para conectar-se à instância MySQL. Para mais informações, consulte Use DMS to log on to an ApsaraDB RDS for MySQL instance.

3. Execute os comandos abaixo no MySQL para criar uma tabela de dados.

   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. Faça login no cluster StarRocks via SSH. Para mais informações, consulte Log on to a cluster.

5. Execute o comando a seguir para conectar-se ao cluster StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

6. Execute os comandos abaixo para criar um usuário e conceder permissões.

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

Etapa 2: Criar catalogs

• Catalog MySQL

  • Exemplo

    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	Descrição
    type	Tipo do catalog. Defina o valor como mysql.
    hostname	Endpoint interno da instância ApsaraDB RDS for MySQL. Copie o endpoint interno na página Database Connection da instância no console ApsaraDB RDS. Exemplo: rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com.
    port	Número da porta do service de banco de dados MySQL. O valor padrão é 3306.
    username	Nome de usuário para acessar o service de banco de dados MySQL. Utilize o nome de usuário da conta criada em Step 1: Prepare test data. Este exemplo usa test.
    password	Senha para acessar o service de banco de dados MySQL. Utilize a senha da conta criada em Step 1: Prepare test data.
    default-database	Nome do banco de dados MySQL padrão. Utilize o nome do banco de dados de Step 1: Prepare test data. Este exemplo usa test_cdc.

• Catalog StarRocks

  • Exemplo

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

  • Parâmetros

    Parâmetro	Descrição
    type	Tipo do catalog. Defina o valor como starrocks.
    endpoint	Endereço IP e porta do Frontend (FE) do StarRocks.
    username	Nome de usuário para acessar o cluster StarRocks. Utilize o nome de usuário da conta criada em Step 1: Prepare test data. Este exemplo usa test.
    password	Senha para o service de banco de dados StarRocks. Utilize a senha da conta criada em Step 1: Prepare test data.
    dbname	Nome do banco de dados StarRocks. Utilize o nome do banco de dados de Step 1: Prepare test data. Este exemplo usa test_cdc.

Etapa 3: Criar e publicar um deployment

1. Na página Draft Editor do console Realtime Compute for Apache Flink, escreva uma instrução CTAS.

   Seguem três exemplos de instruções CTAS.

   • Semântica at-least-once: utilize a opção sink.buffer-flush.interval-ms para configurar o intervalo de gravação de dados no StarRocks. Essa opção reduz latência e uso de memória.

     /*
           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: configure um intervalo de checkpoint. Isso evita perda e duplicação de dados durante falhas, mas a visibilidade dos dados depende do intervalo de checkpoint. Para mais informações, 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 simples: não é necessário definir campos ao criar a tabela. O esquema é copiado do MySQL. No entanto, não é possível criar partições. Para utilizar partições, use o 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'  )*/;
                                     

   Tabela 1. Parâmetros WITH

   Parâmetro	Obrigatório	Descrição
   starrocks.create.table.properties	Sim	Definições de sufixo na instrução CREATE TABLE do StarRocks, excluindo definições de colunas. Exemplos incluem engine, key e buckets.
   database-name	Sim	Nome do banco de dados StarRocks. Este exemplo usa test_cdc.
   jdbc-url	Sim	Utilizado para executar operações de consulta no StarRocks. Por exemplo, jdbc:mysql://172.16..:9030. A parte 172.16.. é o endereço IP interno do cluster StarRocks.
   load-url	Sim	Endereço IP e porta HTTP do frontend (FE) do StarRocks no formato Endereço IP interno do cluster StarRocks:Porta. Este tópico usa a porta 8030 como exemplo. Selecione uma porta com base na versão do seu cluster: 18030: Para EMR V5.9.0 ou superior e EMR V3.43.0 ou superior. 8030: Para EMR V5.8.0 ou anterior e EMR V3.42.0 ou anterior. Nota Para mais informações sobre portas, consulte UI and ports.
   sink.semantic	Não	Defina como exactly-once para garantir semântica de consistência de dados. O padrão é at-least-once.
   starrocks.create.table.mode	Não	Valores suportados: normal (padrão): forneça configurações completas, como engine, key e buckets, na opção starrocks.create.table.properties. simple: o engine é definido como olap e o tipo de chave como primary key por padrão. A chave primária é herdada da tabela MySQL. Por padrão, a tabela é distribuída por hash de todas as colunas de chave primária, sem partições. Especifique buckets na opção starrocks.create.table.properties. Outras configurações como properties são opcionais.

   Nota

   • O parâmetro sink.use.new-api foi removido nas versões 1.15-vvr-6.0.5 e posteriores do Flink. Caso utilize uma versão anterior a 1.15-vvr-6.0.5, adicione 'sink.use.new-api'='false', aos parâmetros WITH.

   • Para informações sobre outras configurações, consulte Carregar dados continuamente do Apache Flink.

   Tabela 2. Parâmetros OPTIONS

   Parâmetro	Descrição
   connector	Tipo do conector. Defina o valor como mysql-cdc.
   hostname	Endpoint interno da instância ApsaraDB RDS for MySQL. Copie o endpoint interno na página Database Connection da instância no console ApsaraDB RDS. Exemplo: rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com.
   port	Número da porta do service de banco de dados MySQL. O valor padrão é 3306.
   username	Nome de usuário para acessar o service de banco de dados MySQL. Utilize o nome de usuário da conta criada em Step 1: Prepare test data. Este exemplo usa test.
   password	Senha para acessar o service de banco de dados MySQL. Utilize a senha da conta criada em Step 1: Prepare test data.
   table-name	Nome da tabela no StarRocks. Utilize o nome da tabela de Step 1: Prepare test data. Este exemplo usa runoob_tbl.
   database-name	Nome do banco de dados MySQL padrão. Utilize o nome do banco de dados de Step 1: Prepare test data. Este exemplo usa test_cdc.

2. Nas Advanced settings da página Draft Editor, selecione a versão 1.15-vvr-6.0.3 ou superior do Flink.

3. Clique em online.

4. Na página Deployments, localize o deployment desejado e clique em START na coluna Actions.

Etapa 4: Demonstração de cenários

Consultar dados

1. Faça login no cluster StarRocks via SSH. Para mais informações, consulte Log on to a cluster.

2. Execute o comando a seguir para conectar-se ao cluster StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

3. No CLI do StarRocks, execute os comandos abaixo para visualizar os dados da tabela.

   use test_cdc;
   select * from runoob_tbl1;

   A saída mostra que os dados da tabela MySQL foram sincronizados com o StarRocks.

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

Consultar dados inseridos

1. No console SQL da instância ApsaraDB RDS for MySQL, execute o comando abaixo para inserir dados.

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

2. No CLI do StarRocks, execute o comando a seguir para visualizar os dados da tabela.

   select * from runoob_tbl1;

   A saída confirma que os dados foram inseridos com sucesso.

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

Sincronizar atualizações de dados

1. No console SQL da instância ApsaraDB RDS for MySQL, execute o comando abaixo para atualizar dados específicos.

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

2. No CLI do StarRocks, execute o comando a seguir para visualizar os dados da tabela.

   select * from runoob_tbl1;

   A saída indica que a atualização dos dados foi sincronizada.

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

Sincronizar exclusões de dados

1. No console SQL da instância ApsaraDB RDS for MySQL, execute o comando abaixo para excluir dados específicos.

   DELETE FROM runoob_tbl WHERE runoob_id = 1

2. No CLI do StarRocks, execute o comando a seguir para visualizar os dados da tabela.

   select * from runoob_tbl1;

   A saída demonstra que a exclusão dos dados foi sincronizada.

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

Adicionar uma coluna anulável

1. No console SQL da instância ApsaraDB RDS for MySQL, execute o comando abaixo para adicionar uma coluna anulável.

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

2. Execute o comando a seguir para inserir dados.

   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. No CLI do StarRocks, execute o comando a seguir para visualizar os dados da tabela.

   select * from runoob_tbl1;

   A saída confirma que a alteração de esquema foi bem-sucedida.

   +-----------+--------------+---------------+-----------------+---------+----------+
   | 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 |
   +-----------+--------------+---------------+-----------------+---------+----------+ 

Introdução ao CDAS

Uma instrução CDAS funciona como um atalho sintático para o CTAS. Ela sincroniza um banco de dados MySQL inteiro com o StarRocks como um único deployment Flink. Também é possível usar a sintaxe including table para sincronizar apenas um subconjunto de tabelas.

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' )*/;