Realtime Compute for Apache Flink:Instruction CREATE TABLE AS (CTAS) (en cours de retrait)

Synchronisation des données

Synchronisation des modifications de schéma

Tout en synchronisant les données en temps réel, l'instruction CTAS réplique également les modifications de schéma de la table source vers la table de destination. Les modifications de schéma incluent la création initiale de la table et toute altération ultérieure de la table.

• Modifications de schéma prises en charge

  Modification de schéma	Description
  Ajout d'une colonne nullable	La colonne correspondante est automatiquement ajoutée à la fin du schéma de la table de destination et ses données sont synchronisées. La nouvelle colonne est nullable par défaut et sa valeur pour les lignes préexistantes est définie sur NULL.
  Ajout d'une colonne non null	La colonne correspondante est automatiquement ajoutée à la fin du schéma de la table de destination et ses données sont synchronisées.
  Suppression d'une colonne nullable	La colonne n'est pas supprimée de la table de destination. Au lieu de cela, ses données sont automatiquement définies sur NULL.
  Renommage d'une colonne	Cette opération est traitée comme l'ajout d'une nouvelle colonne et la suppression de l'ancienne. La colonne renommée est ajoutée à la fin du schéma de la table de destination et les données de la colonne d'origine sont automatiquement définies sur NULL. Remarque Par exemple, si col_a est renommé en col_b, col_b est ajouté à la fin de la table de destination et les données de col_a sont définies sur NULL.
  Modification du type de données d'une colonne	Si le système de destination prend en charge les modifications de type de données : actuellement, seul Paimon prend en charge la gestion des modifications de type de données. CTAS prend en charge les modifications de type pour les colonnes standard, par exemple de INT à BIGINT. Les modifications de type prises en charge dépendent des règles de la destination. Reportez-vous à la documentation du connecteur de destination spécifique pour plus de détails. Si le système de destination ne prend pas en charge les modifications de type de données : actuellement, seul Hologres prend en charge le mode de normalisation de type pour gérer les modifications de type de données. Dans ce mode, un job CTAS crée une table en aval avec des types de données plus larges. Cette approche tire parti de la compatibilité de la destination avec les modifications de type de données. Pour plus de détails, voir Exemple : Synchronisation des données en mode de normalisation de type. Important Vous devez activer le mode de normalisation de type lors du premier démarrage du job CTAS. Sinon, vous devez supprimer la table de destination et effectuer un redémarrage sans état pour que ce mode prenne effet.

  Important

  CTAS détecte les modifications de schéma en comparant les schémas des enregistrements consécutifs plutôt qu'en identifiant l'opération DDL spécifique.

  • Si une colonne est supprimée puis rajoutée sans aucune modification de données entre-temps, CTAS ne détecte pas de modification de schéma.

  • CTAS détecte une modification de schéma uniquement après l'ajout d'une nouvelle colonne et la survenue de modifications de données. Il synchronise ensuite la modification de schéma vers la table de destination.

• Modifications de schéma non prises en charge

  • Modifications des contraintes, telles qu'une clé primaire ou un index.

  • Suppression de colonnes non null.

  • Passage d'une colonne de NOT NULL à NULLABLE.

  Important

  Si une modification de schéma non prise en charge se produit, vous devez supprimer manuellement la table de destination et redémarrer le job CTAS. Cela recrée la table de destination et resynchronise les données historiques.

CREATE TABLE IF NOT EXISTS <sink_table>
(
  [ <table_constraint> ]
)
[COMMENT table_comment]
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
WITH (
  key1=val1,
  key2=val2, 
  ...
 )
AS TABLE <source_table> [/*+ OPTIONS(key1=val1, key2=val2, ... ) */]
[ADD COLUMN { <column_component> | (<column_component> [, ...])}];

<sink_table>:
  [catalog_name.][db_name.]table_name

<table_constraint>:
  [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED

<source_table>:
  [catalog_name.][db_name.]table_name

<column_component>:
  column_name AS computed_column_expression [COMMENT column_comment] [FIRST | AFTER column_name]

La syntaxe CTAS repose sur l'instruction CREATE TABLE. Le tableau suivant décrit ses paramètres.

Synchronisation d'une table unique

Scénario : Synchronisez la table web_sales depuis MySQL vers Hologres.

Prérequis : Vous avez enregistré les catalogues suivants dans votre espace de travail.

• Un catalogue Hologres nommé holo.

• Un catalogue MySQL nommé mysql.

Exemple de code :

L'instruction CTAS s'utilise généralement avec des catalogues pour la source et la destination. Un catalogue source analyse automatiquement le schéma et les options de la table source, ce qui évite la rédaction manuelle du DDL. Cette approche permet une synchronisation complète et incrémentielle des données de la table source vers la table de destination.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- If no database is specified, the table is synchronized to the web_sales table in the default database.
WITH ('jdbcWriteBatchSize' = '1024')   -- Optional: Specifies parameters for the sink table.
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- Specifies additional parameters for the mysql-cdc source table.

Fusion de tables et de bases de données fragmentées

Scénario : Utilisez l'instruction CTAS pour fusionner plusieurs tables MySQL fragmentées en une seule table Hologres.

Solution : En combinaison avec un catalogue MySQL, utilisez une expression régulière pour faire correspondre les noms de base de données et de table que vous souhaitez synchroniser. Les noms de base de données et de table sont inscrits sous forme de deux colonnes supplémentaires dans la table de destination. Pour garantir l'unicité de la clé primaire, le nom de la base de données, le nom de la table et la clé primaire d'origine sont combinés afin de former une nouvelle clé primaire composite pour la table Hologres.

Exemple de code et résultat de la fusion :

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`  
/*+ OPTIONS('server-id'='8001-8004') */;

ALTER TABLE `user02` ADD COLUMN `age` INT;
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30);

Synchronisation de colonnes calculées

Scénario : Ajoutez des colonnes calculées personnalisées à une table Hologres lors du processus de fusion et de synchronisation de tables MySQL fragmentées.

Exemple de code et résultat de la fusion :

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3  AFTER `id`
);

Plusieurs instructions CTAS dans une seule tâche

Scénario : Synchronisez la table MySQL web_sales et les tables utilisateur fragmentées vers Hologres au sein d'une seule tâche.

Solution : Utilisez la syntaxe STATEMENT SET pour valider plusieurs instructions CTAS en une seule tâche. Cette approche permet de réutiliser un nœud source unique pour lire les données de plusieurs tables métier, ce qui réduit le nombre d'ID de serveur, les connexions à la base de données et la charge de lecture sur la source MySQL CDC.

Exemple de code :

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Synchronize the web_sales table.
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

-- Synchronize the sharded user tables.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Synchronisation d'une source vers plusieurs tables de destination

• Si les tables de destination ne nécessitent pas de colonnes calculées

  USE CATALOG `holo`;
  
  BEGIN STATEMENT SET;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user table in the database1 of the Hologres data warehouse.
  CREATE TABLE IF NOT EXISTS `database1`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user table in the database2 of the Hologres data warehouse.
  CREATE TABLE IF NOT EXISTS `database2`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

• Si les tables de destination nécessitent des colonnes calculées

  -- Create a temporary table user_with_changed_id based on the source table user. It supports defining computed columns, such as computed_id, which is calculated based on the id from the source table.
  CREATE TEMPORARY TABLE `user_with_changed_id` (
    `computed_id` AS `id` + 1000
  ) LIKE `mysql`.`tpcds`.`user`;
  
  -- Create a temporary table user_with_changed_age based on the source table user. It supports defining computed columns, such as computed_age, which is calculated based on the age from the source table.
  CREATE TEMPORARY TABLE `user_with_changed_age` (
    `computed_age` AS `age` + 1
  ) LIKE `mysql`.`tpcds`.`user`;
  
  BEGIN STATEMENT SET;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user_with_changed_id table in the Hologres data warehouse. The table will contain the computed ID in the computed_id column. 
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
  AS TABLE `user_with_changed_id`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user_with_changed_age table in the Hologres data warehouse. The table will contain the computed age in the computed_age column. 
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
  AS TABLE `user_with_changed_age`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

Synchronisation d'une nouvelle table

Scénario : Après le démarrage d'une tâche contenant plusieurs instructions CTAS, vous devez ajouter une nouvelle instruction CTAS pour synchroniser les données d'une table nouvellement ajoutée.

Solution : Activez la fonctionnalité de détection de nouvelles tables dans la tâche SQL, ajoutez la nouvelle instruction CTAS, puis redémarrez la tâche à partir d'un point de sauvegarde. Une fois la nouvelle table capturée, ses données seront synchronisées.

Limitations :

• La fonctionnalité de détection de nouvelles tables est prise en charge dans VVR 8.0.1 et versions ultérieures.

• Lors de la synchronisation depuis une table source CDC, la fonctionnalité de détection de nouvelles tables n'est prise en charge que pour les tâches dont le mode de démarrage de la table source est défini sur initial.

• La configuration de la nouvelle table source dans l'instruction CTAS ajoutée doit être identique aux configurations des tables sources existantes pour garantir la réutilisation de la source.

• Les paramètres de configuration de la tâche, tels que le mode de démarrage, ne peuvent pas être modifiés avant et après l'ajout de la nouvelle instruction CTAS.

Procédure :

1. Lorsque vous devez ajouter une nouvelle instruction CTAS, accédez à la page Deployments, arrêtez la tâche et sélectionnez Stop With Savepoint.

2. Dans la tâche SQL, activez la détection de nouvelles tables, ajoutez la nouvelle instruction CTAS, puis Deploy à nouveau la tâche.

   1. Ajoutez l'instruction suivante à la tâche SQL pour activer la détection de nouvelles tables.

      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

   2. Ajoutez la nouvelle instruction CTAS à la tâche SQL. Le code complet final est le suivant.

      -- Enable new table detection.
      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';
      
      USE CATALOG holo;
      
      BEGIN STATEMENT SET;
      
      -- Synchronize the web_sales table.
      CREATE TABLE IF NOT EXISTS web_sales
      AS TABLE mysql.tpcds.web_sales
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- Synchronize the sharded user tables.
      CREATE TABLE IF NOT EXISTS user
      AS TABLE mysql.`wp.*`.`user[0-9]+`
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- Synchronize the product table. (New table)
      CREATE TABLE IF NOT EXISTS product
      AS TABLE mysql.tpcds.product
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      END;

   3. Cliquez sur Deploy.

3. Restaurez la tâche à partir du point de sauvegarde.

   1. Sur la page Deployments, cliquez sur le nom de la tâche cible, accédez à l'onglet State et cliquez sur History.

   2. Dans la liste Savepoints, recherchez le point de sauvegarde créé lors de l'arrêt de la tâche.

   3. Dans la colonne Actions du point de sauvegarde cible, sélectionnez More > Start Job from This Savepoint pour démarrer la tâche. Pour plus d'informations, consultez Démarrer une tâche.

Synchronisation vers une table partitionnée Hologres

Scénario : Utilisez une instruction CTAS pour synchroniser une table source MySQL vers une table partitionnée Hologres.

Règle de partitionnement Hologres : Dans Hologres, si une clé primaire est définie pour la table de destination, les colonnes de partition doivent être incluses dans la clé primaire.

Exemple de code :

L'instruction DDL de la table source MySQL est la suivante :

CREATE TABLE orders (
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    city VARCHAR(100) NOT NULL
    order_date DATE,
    purchaser INTEGER,
    PRIMARY KEY(order_id, product_id)
);

L'approche diffère selon que la clé primaire de la table source inclut ou non la clé de partition.

• Si la clé primaire source inclut la colonne de partition, vous pouvez synchroniser directement avec une instruction CTAS.

  Hologres vérifie automatiquement que la colonne de partition fait partie de la clé primaire.

  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
  PARTITIONED BY (product_id)
  AS TABLE `mysql`.`tpcds`.`orders`;

• Si la clé primaire source n'inclut pas la colonne de partition, redéclarez la clé primaire de la table de destination dans l'instruction CTAS.

  Si une colonne de partition (par exemple, city) ne fait pas partie de la clé primaire de la table source, la tâche échoue. Vous devez redéclarer la clé primaire de la table de destination dans l'instruction CTAS pour garantir que la colonne de partition en fait partie.

  -- You can use the following SQL to specify the primary key of the Hologres partitioned table as order_id, product_id, and city.
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`(
      CONSTRAINT `PK_order_id_city` PRIMARY KEY (`order_id`,`product_id`,`city`) NOT ENFORCED
  )
  PARTITIONED BY (city)
  AS TABLE `mysql`.`tpcds`.`orders`;

Synchronisation des données en mode de normalisation des types

Scénario : Lorsque vous utilisez une instruction CTAS pour synchroniser des données vers une table Hologres, vous devez prendre en charge les scénarios nécessitant l'ajustement de la précision des types de données existants (par exemple, passer de VARCHAR(10) à VARCHAR(20)) ou le changement du type de données (par exemple, passer de SMALLINT à INT).

Solution : Utilisez le mode de normalisation des types de Hologres pour synchroniser les données. Ce mode doit être activé au premier démarrage de la tâche CTAS. S'il n'est pas activé au démarrage, vous devrez supprimer la table en aval et effectuer un redémarrage sans état de la tâche pour que la modification soit prise en compte.

Règles de normalisation des types :

Lorsqu'un type de données en amont est modifié, si le nouveau type et le type d'origine se normalisent vers le même type cible, la tâche s'exécute normalement. Dans le cas contraire, la modification est considérée comme incompatible et la tâche CTAS génère une exception. Les règles spécifiques sont les suivantes :

• Les types TINYINT, SMALLINT, INT et BIGINT sont normalisés vers BIGINT.

• Les types CHAR, VARCHAR et STRING sont normalisés vers STRING.

• Les types FLOAT et DOUBLE sont normalisés vers DOUBLE.

• Les autres types de données sont créés selon les règles de mappage des types d'origine. Pour plus d'informations, consultez la section Mappage des types.

Exemple de code :

CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders` 
WITH (
'connector' = 'hologres', 
'enableTypeNormalization' = 'true' -- Enable type normalization mode.
) AS TABLE `mysql`.`tpcds`.`orders`;

Synchronisation de MongoDB vers Hologres

Limites :

• Nécessite Realtime Compute for Apache Flink VVR 8.0.6 ou version ultérieure et MongoDB 6.0 ou version ultérieure.

• Dans les indices SQL, définissez scan.incremental.snapshot.enabled et scan.full-changelog sur true.

• La fonctionnalité d'images avant et après doit être activée dans la base de données MongoDB. Pour savoir comment l'activer, consultez la documentation Document Preimages.

• La synchronisation de plusieurs collections MongoDB dans une seule tâche impose les exigences suivantes :

  • La configuration MongoDB doit être identique pour chaque table, y compris les paramètres hosts, scheme, username, password et connectionOptions.

  • Le paramètre scan.startup.mode doit être identique pour chaque table.

Exemple de code :

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table1`
AS TABLE `mongodb`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table2`
AS TABLE `mongodb`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

END;