Realtime Compute for Apache Flink:CREATE DATABASE AS (CDAS) statement (Retiring)

Sincronización de datos

Evolución de esquema

Cuando CDAS sincroniza una base de datos, los cambios de esquema (como la adición de columnas) se propagan automáticamente al destino. El comportamiento y la directiva son los mismos que para CTAS. Consulte Schema evolution.

Limitaciones de sintaxis

• No se admite la depuración de un borrador SQL que contenga una instrucción CDAS.

• MiniBatch no es compatible.

  Importante

  Elimine todas las configuraciones de MiniBatch antes de crear un borrador SQL con una instrucción CTAS o CDAS: 1. Vaya a O&M > Configurations. 2. Seleccione la pestaña Deployment Defaults. 3. En la sección Other Configuration, compruebe que se hayan eliminado las configuraciones de MiniBatch. Si encuentra el error "Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax" al crear o iniciar un despliegue, consulte ¿Cómo soluciono este error?

Conectores compatibles

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

Parámetros

Sincronizar una base de datos

Sincronice todas las tablas de la base de datos MySQL tpcds hacia Hologres.

Requisitos previos:

• Se creó un catálogo de Hologres llamado holo en el espacio de trabajo.

• Se creó un catálogo de MySQL llamado mysql en el espacio de trabajo.

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.

Las opciones establecidas en la cláusula WITH se aplican únicamente al trabajo actual y controlan el comportamiento de escritura. No se conservan en el catálogo de Hologres. Para consultar las opciones admitidas, vea Hologres .

Consolidar y sincronizar shards de base de datos

Combine todas las tablas con nombres idénticos en varios shards de base de datos MySQL en tablas únicas de Hologres.

Escenario: Una instancia de MySQL tiene shards denominados desde order_db01 hasta order_db99. Cada shard contiene tablas como order y order_detail. Todos los datos de los shards, incluidos los cambios de esquema, deben almacenarse en Hologres.

Solución: Utilice una expresión regular en el nombre de la base de datos para coincidir con todos los shards. Los nombres de la base de datos y de la tabla se agregan a cada tabla de destino como dos campos adicionales. La clave principal de Hologres incluye el nombre de la base de datos, el nombre de la tabla y las columnas de clave principal de la tabla de origen para garantizar la unicidad. No es necesario crear las tablas de destino con anticipación.

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).

Las tablas con nombres idénticos en distintos shards se combinan en una sola tabla de Hologres.

Sincronizar nuevas tablas

Una vez iniciado un trabajo CDAS en ejecución, habilite la detección de nuevas tablas y reinicie desde un punto de guardado para capturar las tablas recién agregadas.

La detección de nuevas tablas requiere VVR 8.0.6 o posterior. El modo de inicio de la tabla de origen debe ser initial. Durante la fase de instantánea completa, no agregue ni elimine tablas después de crear un punto de guardado y luego reiniciar el trabajo desde dicho punto. De lo contrario, el trabajo no podrá leer los datos correctamente.

1. En la página Deployments, busque el despliegue objetivo y haga clic en Cancel en la columna Actions.

2. En el cuadro de diálogo, expanda More Strategies, seleccione Stop With Savepoint y haga clic en OK.

3. En el borrador SQL, agregue la siguiente instrucción:

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

4. Haga clic en Deploy.

5. Recupere el trabajo desde el punto de guardado:

   1. En la página Deployments, haga clic en el nombre del despliegue.

   2. En la página de detalles del despliegue, haga clic en la pestaña State y luego en la subpestaña History.

   3. En la lista Savepoints, busque el punto de guardado creado al detener el trabajo.

   4. Elija More > Start job from this savepoint en la columna Actions. Consulte Start a job.

Ejecutar múltiples instrucciones CDAS y CTAS como un solo trabajo

Sincronice varias bases de datos en un solo trabajo para maximizar la reutilización del origen.

Escenario: Sincronice tpcds, tpch y los shards user_db01–user_db99 hacia Hologres en un solo trabajo.

Solución: Envuelva todas las instrucciones en un bloque STATEMENT SET. Flink reutiliza un único vértice de origen en todas las instrucciones, lo que reduce la cantidad de ID de servidor, conexiones a la base de datos y la carga general de lectura, especialmente para orígenes CDC de MySQL.

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;

Sincronizar múltiples bases de datos MySQL a Kafka

Sincronice tablas de varias bases de datos MySQL a Kafka y evite conflictos de nombres de temas.

Escenario: Existen tablas con el mismo nombre tanto en tpcds como en tpch. Enviar ambas al mismo tema de Kafka sobrescribiría los datos.

Solución: Utilice la opción cdas.topic.pattern para generar nombres de tema únicos por base de datos. El marcador de posición {table-name} se reemplaza en tiempo de ejecución por el nombre real de la tabla. Por ejemplo, 'cdas.topic.pattern'='tpcds-{table-name}' dirige table1 de tpcds al tema 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;

El uso de Kafka como capa intermedia entre MySQL y Flink también reduce la carga en MySQL. Consulte Synchronize MySQL database to Kafka with Flink CDC.