Realtime Compute for Apache Flink:Instrução CREATE DATABASE AS (CDAS)

Sincronização de dados

Evolução de schema

Durante a sincronização de um banco de dados via CDAS, alterações de schema — como adição de colunas — são propagadas automaticamente para o sink. O comportamento e a política são os mesmos do CTAS. Consulte Evolução de schema.

Limitações de sintaxe

• Não há suporte para depuração de rascunhos SQL que contenham uma instrução CDAS.

• MiniBatch não é suportado.

  Importante

  Remova todas as configurações do MiniBatch antes de criar um rascunho SQL com uma instrução CTAS ou CDAS: 1. Acesse O&M > Configurations. 2. Selecione a aba Deployment Defaults. 3. Na seção Other Configuration, verifique se as configurações do MiniBatch foram removidas. Se encontrar o erro "Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax" ao criar ou iniciar um deployment, consulte Como corrigir este erro?

Conectores suportados

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 um banco de dados

Sincronize todas as tabelas do banco de dados MySQL tpcds para o Hologres.

Pré-requisitos:

• Um catálogo Hologres chamado holo deve estar criado no workspace.

• Um catálogo MySQL chamado mysql deve estar criado no workspace.

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.

As opções definidas na cláusula WITH aplicam-se apenas ao job atual e controlam o comportamento de escrita. Elas não são persistidas no catálogo Hologres. Para opções suportadas, consulte Hologres .

Consolidar e sincronizar shards de banco de dados

Mescle todas as tabelas com nomes idênticos em vários shards de banco de dados MySQL em tabelas únicas no Hologres.

Cenário: Uma instância MySQL possui shards chamados order_db01 até order_db99. Cada shard contém tabelas como order e order_detail. Todos os dados dos shards — incluindo alterações de schema — precisam ser enviados ao Hologres.

Solução: Use uma expressão regular no nome do banco de dados para corresponder a todos os shards. Os nomes do banco de dados e da tabela são adicionados a cada tabela de sink como dois campos adicionais. A chave primária do Hologres inclui o nome do banco de dados, o nome da tabela e as colunas de chave primária da tabela de origem para garantir unicidade. Não é necessário criar tabelas de destino antecipadamente.

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

Tabelas com nomes idênticos entre shards são mescladas em uma única tabela Hologres.

Sincronizar novas tabelas

Após o início de um job CDAS em execução, ative a detecção de novas tabelas e reinicie a partir de um savepoint para capturar tabelas recém-adicionadas.

A detecção de novas tabelas requer VVR 8.0.6 ou posterior. O modo de inicialização da tabela de origem deve ser initial .

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

2. Na caixa de diálogo, expanda More Strategies, selecione Stop With Savepoint e clique em OK.

3. No rascunho SQL, adicione a seguinte instrução:

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

4. Clique em Deploy.

5. Recupere o job a partir do savepoint:

   1. Na página Deployments, clique no nome do deployment.

   2. Na página de detalhes do deployment, clique na aba State e depois na subaba History.

   3. Na lista Savepoints, localize o savepoint criado quando você parou o job.

   4. Escolha More > Start job from this savepoint na coluna Actions. Consulte Iniciar um job.

Executar múltiplas instruções CDAS e CTAS como um único job

Sincronize vários bancos de dados em um único job para maximizar a reutilização da source.

Cenário: Sincronizar tpcds, tpch e os shards user_db01–user_db99 para o Hologres em um único job.

Solução: Envolva todas as instruções em um bloco STATEMENT SET. O Flink reutiliza um único vértice de source em todas as instruções, reduzindo o número de server IDs, conexões de banco de dados e a carga geral de leitura — especialmente para fontes 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;

Sincronizar múltiplos bancos de dados MySQL para o Kafka

Sincronize tabelas de vários bancos de dados MySQL para o Kafka evitando conflitos de nomes de tópicos.

Cenário: Tabelas com o mesmo nome existem tanto em tpcds quanto em tpch. Enviar ambas para o mesmo tópico Kafka sobrescreveria os dados.

Solução: Utilize a opção cdas.topic.pattern para gerar nomes de tópicos exclusivos por banco de dados. O placeholder {table-name} é substituído em tempo de execução pelo nome real da tabela. Por exemplo, 'cdas.topic.pattern'='tpcds-{table-name}' direciona a table1 do tpcds para o tópico 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;

A introdução do Kafka como camada intermediária entre o MySQL e o Flink também reduz a carga no MySQL. Consulte Sincronizar banco de dados MySQL para Kafka com Flink CDC.