E-MapReduce:Synchroniser des données de MySQL vers StarRocks à l'aide de CTAS et CDAS

Contexte

L'instruction CTAS (CREATE TABLE AS) crée automatiquement une table StarRocks dont le schéma est identique à celui de la table source MySQL. Elle synchronise également les données et réplique les modifications de schéma de la source vers la destination en temps réel, ce qui simplifie la création des tables cibles et garantit la cohérence des schémas.

Une instruction CTAS synchronise les données en temps réel et propage les modifications de schéma de la source vers la table de destination.

Prérequis

• Activez Realtime Compute for Apache Flink (Fully Managed) et créez un cluster Flink. Pour plus d'informations, consultez Activate Realtime Compute for Apache Flink (Fully Managed) et Getting started with a Flink SQL deployment.

• Créez un cluster StarRocks. Pour plus d'informations, consultez Create a StarRocks cluster.

• Créez une instance ApsaraDB RDS for MySQL. Pour plus d'informations, consultez Create an ApsaraDB RDS for MySQL instance.

Limites

• Le cluster Flink, le cluster StarRocks et l'instance ApsaraDB RDS for MySQL doivent se trouver dans le même VPC.

• L'instance ApsaraDB RDS for MySQL doit être en version 5.7 ou ultérieure.

• L'accès Internet doit être activé sur le cluster StarRocks.

• La version de Flink utilisée dans votre cluster doit être 1.15-vvr-6.0.3 ou ultérieure.

Étape 1 : Préparer les données de test

1. Créez une base de données et un compte de test. Pour plus d'informations, consultez Create databases and accounts for an ApsaraDB RDS for MySQL instance.

   Après avoir créé la base de données et le compte, accordez les autorisations de lecture et d'écriture au compte de test.

   Remarque

   Dans cette rubrique, la base de données se nomme test_cdc et le compte test.

2. Connectez-vous à l'instance MySQL à l'aide du compte de test. Pour plus d'informations, consultez Use DMS to log on to an ApsaraDB RDS for MySQL instance.

3. Exécutez les commandes suivantes dans MySQL pour créer une table de données.

   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. Connectez-vous au cluster StarRocks via SSH. Pour plus d'informations, consultez Log on to a cluster.

5. Exécutez la commande suivante pour vous connecter au cluster StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

6. Exécutez les commandes suivantes pour créer un utilisateur et lui attribuer des autorisations.

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

Étape 2 : Créer des catalogs

• Catalog MySQL

  • Exemple

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

  • Paramètres

    Parameter	Description
    type	Type du catalog. Définissez la valeur sur mysql.
    hostname	Endpoint interne de l'instance ApsaraDB RDS for MySQL. Vous pouvez copier cet endpoint depuis la page Database Connection de l'instance dans la console ApsaraDB RDS. Exemple : rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com.
    port	Numéro de port du service de base de données MySQL. La valeur par défaut est 3306.
    username	Nom d'utilisateur pour accéder au service de base de données MySQL. Utilisez le nom d'utilisateur du compte créé lors de l'Étape 1 : Préparer les données de test. Cet exemple utilise test.
    password	Mot de passe pour accéder au service de base de données MySQL. Utilisez le mot de passe du compte créé lors de l'Étape 1 : Préparer les données de test.
    default-database	Nom de la base de données MySQL par défaut. Utilisez le nom de la base de données indiqué dans l'Étape 1 : Préparer les données de test. Cet exemple utilise test_cdc.

• Catalog StarRocks

  • Exemple

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

  • Paramètres

    Parameter	Description
    type	Type du catalog. Définissez la valeur sur starrocks.
    endpoint	Adresse IP et port du Frontend (FE) StarRocks.
    username	Nom d'utilisateur pour accéder au cluster StarRocks. Utilisez le nom d'utilisateur du compte créé lors de l'Étape 1 : Préparer les données de test. Cet exemple utilise test.
    password	Mot de passe pour le service de base de données StarRocks. Utilisez le mot de passe du compte créé lors de l'Étape 1 : Préparer les données de test.
    dbname	Nom de la base de données StarRocks. Utilisez le nom de la base de données indiqué dans l'Étape 1 : Préparer les données de test. Cet exemple utilise test_cdc.

Étape 3 : Créer et publier un déploiement

1. Sur la page Draft Editor de la console Realtime Compute for Apache Flink, rédigez une instruction CTAS.

   Voici trois exemples d'instructions CTAS.

   • Sémantique At-least-once : Utilisez l'option sink.buffer-flush.interval-ms pour configurer l'intervalle d'écriture des données dans StarRocks. Cette option permet de réduire la latence et l'utilisation de la mémoire.

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

   • Sémantique Exactly-once : Vous devez configurer un intervalle de checkpoint. Cela empêche la perte et la duplication des données en cas de défaillance, mais la visibilité des données dépend de l'intervalle de checkpoint. Pour plus d'informations, consultez 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'  )*/;
                                     

   • Mode simple : La définition des champs n'est pas requise lors de la création de la table, car le schéma est copié directement depuis MySQL. Toutefois, ce mode ne permet pas de créer de partitions. Pour utiliser des partitions, optez pour le mode 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'  )*/;
                                     

   Tableau 1. Paramètres WITH

   Parameter	Required	Description
   starrocks.create.table.properties	Yes	Définitions suffixées de l'instruction StarRocks CREATE TABLE, à l'exclusion des définitions de colonnes. Cela inclut notamment engine, key et buckets.
   database-name	Yes	Nom de la base de données StarRocks. Cet exemple utilise test_cdc.
   jdbc-url	Yes	Permet d'exécuter des opérations de requête dans StarRocks. Par exemple, jdbc:mysql://172.16..:9030. La partie 172.16.. correspond à l'adresse IP interne du cluster StarRocks.
   load-url	Yes	Adresse IP et port HTTP du frontend (FE) StarRocks au format Internal IP address of StarRocks cluster:Port. Cette rubrique prend le port 8030 comme exemple. Sélectionnez un port en fonction de la version de votre cluster : 18030 : Pour EMR V5.9.0 ou ultérieur et EMR V3.43.0 ou ultérieur. 8030 : Pour EMR V5.8.0 ou antérieur et EMR V3.42.0 ou antérieur. Remarque Pour plus d'informations sur les ports, consultez UI and ports.
   sink.semantic	No	Définissez cette valeur sur exactly-once pour garantir la sémantique de cohérence des données. La valeur par défaut est at-least-once.
   starrocks.create.table.mode	No	Valeurs prises en charge : normal (par défaut) : Vous devez fournir des configurations complètes, telles que engine, key et buckets, dans l'option starrocks.create.table.properties. simple : Le moteur est défini sur olap et le type de clé sur primary key par défaut. La clé primaire est héritée de la table MySQL. Par défaut, la table est distribuée par hachage de toutes les colonnes de clé primaire, sans partitions. Vous devez spécifier buckets dans l'option starrocks.create.table.properties. D'autres configurations comme properties sont facultatives.

   Remarque

   • Le paramètre sink.use.new-api a été supprimé dans les versions Flink 1.15-vvr-6.0.5 et ultérieures. Si vous utilisez une version antérieure à 1.15-vvr-6.0.5, ajoutez 'sink.use.new-api'='false', aux paramètres WITH.

   • Pour plus d'informations sur les autres configurations, consultez Continuously load data from Apache Flink.

   Tableau 2. Paramètres OPTIONS

   Parameter	Description
   connector	Type de connecteur. Définissez la valeur sur mysql-cdc.
   hostname	Endpoint interne de l'instance ApsaraDB RDS for MySQL. Vous pouvez copier cet endpoint depuis la page Database Connection de l'instance dans la console ApsaraDB RDS. Par exemple, rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com.
   port	Numéro de port du service de base de données MySQL. La valeur par défaut est 3306.
   username	Nom d'utilisateur pour accéder au service de base de données MySQL. Utilisez le nom d'utilisateur du compte créé lors de l'Étape 1 : Préparer les données de test. Cet exemple utilise test.
   password	Mot de passe pour accéder au service de base de données MySQL. Utilisez le mot de passe du compte créé lors de l'Étape 1 : Préparer les données de test.
   table-name	Nom de la table dans StarRocks. Utilisez le nom de la table indiqué dans l'Étape 1 : Préparer les données de test. Cet exemple utilise runoob_tbl.
   database-name	Nom de la base de données MySQL par défaut. Utilisez le nom de la base de données indiqué dans l'Étape 1 : Préparer les données de test. Cet exemple utilise test_cdc.

2. Dans les Advanced settings de la page Draft Editor, sélectionnez la version Flink 1.15-vvr-6.0.3 ou ultérieure.

3. Cliquez sur online.

4. Sur la page Deployments, localisez le déploiement cible et cliquez sur START dans la colonne Actions.

Étape 4 : Démonstration de scénarios

Interrogation des données

1. Connectez-vous au cluster StarRocks via SSH. Pour plus d'informations, consultez Log on to a cluster.

2. Exécutez la commande suivante pour vous connecter au cluster StarRocks.

   mysql -h127.0.0.1 -P 9030 -uroot

3. Dans le CLI StarRocks, exécutez les commandes suivantes pour consulter les données de la table.

   use test_cdc;
   select * from runoob_tbl1;

   Le résultat indique que les données de la table MySQL ont bien été synchronisées vers StarRocks.

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

Interrogation des données insérées

1. Dans la console SQL de l'instance ApsaraDB RDS for MySQL, exécutez la commande suivante pour insérer des données.

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

2. Dans le CLI StarRocks, exécutez la commande suivante pour consulter les données de la table.

   select * from runoob_tbl1;

   Le résultat confirme que les données ont été insérées avec succès.

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

Synchronisation des mises à jour de données

1. Dans la console SQL de l'instance ApsaraDB RDS for MySQL, exécutez la commande suivante pour mettre à jour des données spécifiques.

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

2. Dans le CLI StarRocks, exécutez la commande suivante pour consulter les données de la table.

   select * from runoob_tbl1;

   Le résultat montre que la mise à jour des données a été synchronisée.

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

Synchronisation des suppressions de données

1. Dans la console SQL de l'instance ApsaraDB RDS for MySQL, exécutez la commande suivante pour supprimer des données spécifiques.

   DELETE FROM runoob_tbl WHERE runoob_id = 1

2. Dans le CLI StarRocks, exécutez la commande suivante pour consulter les données de la table.

   select * from runoob_tbl1;

   Le résultat indique que la suppression des données a été synchronisée.

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

Ajout d'une colonne nullable

1. Dans la console SQL de l'instance ApsaraDB RDS for MySQL, exécutez la commande suivante pour ajouter une colonne nullable.

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

2. Exécutez la commande suivante pour insérer des données.

   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. Dans le CLI StarRocks, exécutez la commande suivante pour consulter les données de la table.

   select * from runoob_tbl1;

   Le résultat confirme que la modification du schéma a réussi.

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

Présentation de CDAS

Une instruction CDAS constitue un raccourci syntaxique pour CTAS. Elle permet de synchroniser une base de données MySQL entière vers StarRocks via un seul déploiement Flink. Vous pouvez également utiliser la syntaxe « including table » pour ne synchroniser qu'un sous-ensemble de tables.

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