Realtime Compute for Apache Flink:Instruction CREATE DATABASE AS (CDAS) (Abandon)

Synchronisation des données

Évolution du schéma

Lorsqu'une base de données est synchronisée via CDAS, les modifications de schéma, telles que l'ajout de colonnes, sont propagées automatiquement vers la cible. Ce comportement suit la même politique que pour CTAS. Consultez Schema evolution.

Limites syntaxiques

• Le débogage d'un brouillon SQL contenant une instruction CDAS n'est pas pris en charge.

• MiniBatch n'est pas pris en charge.

  Important

  Supprimez toutes les configurations MiniBatch avant de créer un brouillon SQL avec une instruction CTAS ou CDAS : 1. Accédez à O&M > Configurations. 2. Sélectionnez l'onglet Deployment Defaults. 3. Dans la section Other Configuration, vérifiez que les configurations MiniBatch ont bien été supprimées. Si vous rencontrez l'erreur « Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax » lors de la création ou du démarrage d'un déploiement, consultez How do I fix this error?

Connecteurs pris en charge

CREATE DATABASE IF NOT EXISTS <target_database>
[COMMENT database_comment]
[WITH (key1=val1, key2=val2, ...)]
AS DATABASE <source_database>
INCLUDING { ALL TABLES | TABLE 'table_name' }
[EXCLUDING TABLE 'table_name']
[/*+ OPTIONS(key1=val1, key2=val2, ... ) */]

<target_database>:
  [catalog_name.]db_name

<source_database>:
  [catalog_name.]db_name

Paramètres

Synchroniser une base de données

Synchronisez toutes les tables de la base de données MySQL tpcds vers Hologres.

Prérequis :

• Un catalogue Hologres nommé holo est créé dans l'espace de travail.

• Un catalogue MySQL nommé mysql est créé dans l'espace de travail.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds          -- Create holo_tpcds in Hologres.
WITH ('sink.parallelism' = '4')                   -- Set sink parallelism (default: 4 for Hologres).
AS DATABASE mysql.tpcds INCLUDING ALL TABLES      -- Sync all tables from mysql.tpcds.
/*+ OPTIONS('server-id'='8001-8004') */;          -- Configure the MySQL CDC source options.

Les options définies dans la clause WITH s'appliquent uniquement au job en cours et contrôlent le comportement d'écriture. Elles ne sont pas persistées dans le catalogue Hologres. Pour connaître les options prises en charge, consultez Hologres .

Consolider et synchroniser des shards de base de données

Fusionnez toutes les tables portant des noms identiques dans plusieurs shards de base de données MySQL en tables Hologres uniques.

Scénario : Une instance MySQL contient des shards nommés de order_db01 à order_db99. Chaque shard renferme des tables telles que order et order_detail. L'ensemble des données des shards, y compris les modifications de schéma, doit être transféré vers Hologres.

Solution : Utilisez une expression régulière sur le nom de la base de données pour cibler tous les shards. Les noms de la base de données et de la table sont ajoutés à chaque table cible sous forme de deux champs supplémentaires. La clé primaire Hologres combine le nom de la base de données, le nom de la table et les colonnes de clé primaire de la table source afin de garantir l'unicité. La création préalable des tables cibles n'est pas nécessaire.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order              -- Create holo_order in Hologres.
WITH('sink.parallelism'='4')                          -- Set sink parallelism (optional).
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES  -- Match all order_db shards.
/*+ OPTIONS('server-id'='8001-8004') */;              -- Configure MySQL CDC source options (optional).

Les tables portant des noms identiques dans les différents shards sont fusionnées en une seule table Hologres.

Synchroniser de nouvelles tables

Une fois qu'un job CDAS est en cours d'exécution, activez la détection des nouvelles tables et redémarrez à partir d'un savepoint pour intégrer les tables récemment ajoutées.

La détection des nouvelles tables nécessite VVR 8.0.6 ou ultérieur. Le mode de démarrage de la table source doit être initial. Pendant la phase de snapshot complet, n'ajoutez ni ne supprimez de tables après la création d'un savepoint suivi d'un redémarrage du job à partir de ce savepoint. Sinon, le job ne pourra pas lire les données correctement.

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

2. Dans la boîte de dialogue, développez More Strategies, sélectionnez Stop With Savepoint, puis cliquez sur OK.

3. Dans le brouillon SQL, ajoutez l'instruction suivante :

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

4. Cliquez sur Deploy.

5. Récupérez le job à partir du savepoint :

   1. Sur la page Deployments, cliquez sur le nom du déploiement.

   2. Sur la page de détails du déploiement, cliquez sur l'onglet State, puis sur le sous-onglet History.

   3. Dans la liste Savepoints, repérez le savepoint créé lors de l'arrêt du job.

   4. Choisissez More > Start job from this savepoint dans la colonne Actions. Consultez Start a job.

Exécuter plusieurs instructions CDAS et CTAS dans un seul job

Synchronisez plusieurs bases de données au sein d'un même job afin d'optimiser la réutilisation des sources.

Scénario : Synchroniser tpcds, tpch et les shards user_db01–user_db99 vers Hologres dans un seul job.

Solution : Regroupez toutes les instructions dans un bloc STATEMENT SET. Flink réutilise alors un vertex source unique pour l'ensemble des instructions, ce qui réduit le nombre d'ID serveur, de connexions à la base de données et la charge globale de lecture, particulièrement pour les sources MySQL CDC.

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Sync user tables from all user_db shards.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync tpcds database.
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync tpch database.
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Synchroniser plusieurs bases de données MySQL vers Kafka

Synchronisez des tables issues de plusieurs bases de données MySQL vers Kafka tout en évitant les conflits de noms de topics.

Scénario : Des tables portant le même nom existent à la fois dans tpcds et tpch. Les envoyer vers le même topic Kafka entraînerait un écrasement des données.

Solution : Utilisez l'option cdas.topic.pattern pour générer des noms de topics uniques par base de données. Le placeholder {table-name} est remplacé au moment de l'exécution par le nom réel de la table. Par exemple, 'cdas.topic.pattern'='tpcds-{table-name}' dirige table1 de tpcds vers le topic tpcds-table1.

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- Sync tpcds database. Topic names follow the pattern "tpcds-<table_name>".
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync tpch database. Topic names follow the pattern "tpch-<table_name>".
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

L'introduction de Kafka comme couche intermédiaire entre MySQL et Flink permet également de réduire la charge sur MySQL. Consultez Synchronize MySQL database to Kafka with Flink CDC.