Manage MySQL catalogs - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Bancos de dados compatíveis

O serviço é compatível com ApsaraDB RDS for MySQL, PolarDB for MySQL e bancos de dados MySQL autogerenciados. Somente as versões MySQL 5.7 e MySQL 8.0.x têm suporte.

Limitações

A instância MySQL e o Realtime Compute for Apache Flink devem estar na mesma virtual private cloud (VPC). Para conexões entre VPCs ou pela Internet, consulte FAQ sobre conectividade de rede.
Não é possível modificar a configuração do catálogo após a criação. Para alterá-la, exclua o catálogo e crie um novo.
É permitido consultar apenas tabelas em bancos de dados existentes. O Realtime Compute for Apache Flink não oferece suporte à criação de bancos de dados ou tabelas.
Tabelas source de CDC do MySQL aceitam leitura apenas no modo streaming, não no modo batch.
Views não podem ser usadas como tabelas em um catálogo MySQL criado com Ververica Runtime (VVR) 8.0.7 ou posterior.
O catálogo MySQL não identifica tabelas criadas com sintaxe específica do PolarDB (por exemplo, PARTITION BY KEY(...) PARTITIONS 16, UNIQUE KEY ... em uma instrução CREATE TABLE).

Observações de uso

Ative o log binário no ApsaraDB RDS for MySQL, PolarDB for MySQL ou no banco de dados MySQL autogerenciado antes de usar uma tabela como tabela source de CDC do MySQL. Para mais informações, consulte Configurar um banco de dados MySQL.
Defina um server-id exclusivo por implantação ao ler de uma tabela source de CDC do MySQL. Ao executar várias implantações simultaneamente, use um intervalo de valores que contenha pelo menos a mesma quantidade de valores do paralelismo da implantação.
Armazene a senha do banco de dados como variável em vez de especificá-la em texto simples. Para mais informações, consulte Criar uma variável.
O campo de comentário não aparece nos detalhes do esquema da tabela.
Excluir um catálogo não remove as tabelas subjacentes no MySQL. As implantações em execução continuam sem impacto, mas reimplantar ou reiniciar uma implantação que referencie um catálogo excluído gera um erro de "tabela não encontrada".

Criar um catálogo MySQL

Crie um catálogo MySQL pelo console (recomendado) ou execute uma instrução SQL CREATE CATALOG.

Console (recomendado)

Acesse a página Catalogs.
1. Faça login no console do Realtime Compute for Apache Flink. Localize o workspace desejado e clique em Console na coluna Actions.
2. No painel de navegação à esquerda, clique em Data Management.
Na página Catalog List, clique em Create Catalog. Na aba Built-in Catalog, clique em MySQL e, em seguida, clique em Next.

Na etapa Configure Catalog, preencha os parâmetros.

Importante

Não é possível modificar a configuração do catálogo após a criação. Para alterá-la, exclua o catálogo e crie um novo.

Parâmetro	Descrição	Obrigatório
`catalogname`	Nome do catálogo MySQL.	Sim
`hostname`	Endereço IP ou hostname do banco de dados MySQL. Se a instância MySQL estiver em uma VPC diferente ou precisar de acesso pela Internet, estabeleça as conexões de rede primeiro. Consulte FAQ sobre conectividade de rede.	Sim
`port`	Número da porta do banco de dados MySQL. Padrão: `3306`.	Não
`default-database`	Nome do banco de dados MySQL padrão.	Sim
`username`	Nome de usuário do banco de dados MySQL.	Sim
`password`	Senha do banco de dados MySQL. Armazene-a como variável para evitar exposição em texto simples. Consulte Criar uma variável.	Sim

Configure catalog parameters

Clique em Confirm. O novo catálogo aparece no painel Catalogs, no lado esquerdo da página Catalog List.

Visualizar e excluir um catálogo MySQL

Console (recomendado)

Na página Catalogs, clique no nome do catálogo para visualizar seu Name e Type na seção Catalog List.

Visualizar bancos de dados e tabelas: Clique em View na coluna Actions.
Excluir o catálogo: Clique em Delete na coluna Actions.

Usar um catálogo MySQL

Ler de uma tabela source de CDC

Use SQL hints para definir o server-id de cada implantação. Quando várias implantações leem da mesma tabela source simultaneamente, use um intervalo que contenha pelo menos a mesma quantidade de valores do paralelismo da implantação.

INSERT INTO `<othersinktable>`
SELECT ...
FROM `<mysqlcatalog>`.`<dbname>`.`<tablename>` /*+ OPTIONS('server-id' = '6000-6008') */;

Ler de tabelas fragmentadas

Consulte tabelas fragmentadas como uma única tabela lógica usando expressões regulares nos segmentos de nome do banco de dados e da tabela. O resultado da consulta inclui dois campos de sistema, _db_name (STRING) e _table_name (STRING), que formam a nova chave primária conjunta com as chaves primárias originais.

Por exemplo, tabelas de user01 a user99 distribuídas em shards de banco de dados de db01 a db10:

SELECT ... FROM `db.*`.`user.*` /*+ OPTIONS('server-id'='6000-6018') */;

A chave primária conjunta dessa tabela lógica é (_db_name, _table_name, id), onde id é a chave primária original.

Para mesclar e sincronizar dados em várias tabelas em um banco de dados fragmentado, consulte Consolidar e sincronizar shards de tabela e banco de dados.

Sincronizar dados com CTAS e CDAS

Use CREATE TABLE AS (CTAS) para sincronizar uma única tabela — incluindo alterações de esquema — ou para mesclar dados de tabelas fragmentadas. Use CREATE DATABASE AS (CDAS) para sincronizar um banco de dados inteiro, incluindo alterações contínuas de esquema.

Nota

CTAS e CDAS exigem armazenamentos de dados upstream e downstream compatíveis. Se o conector downstream não implementar CatalogTableProvider, o sistema retorna um erro.

-- Single-table sync: propagates schema changes and data changes in real time
CREATE TABLE IF NOT EXISTS `<targetcatalog>`.`<targetdbname>`.`<targettablename>`
WITH (...)
AS TABLE `<mysqlcatalog>`.`<dbname>`.`<tablename>`
/*+ OPTIONS('server-id'='6000-6018') */;

-- Full-database sync: propagates schema changes and data changes in real time
CREATE DATABASE `<targetcatalog>`.`<targetdbname>` WITH (...)
AS DATABASE `<mysqlcatalog>`.`<dbname>` INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='6000-6018') */;

O exemplo a seguir sincroniza uma tabela MySQL com o Hologres:

USE CATALOG holocatalog; -- Set the target catalog

CREATE TABLE IF NOT EXISTS holotable   -- Target table name; defaults to the catalog's default database if no database is specified
WITH ('jdbcWriteBatchSize' = '1024')   -- Optional sink connector options
AS TABLE mysqlcatalog.dbmysql.mysqltable
/*+ OPTIONS('server-id'='8001-8004') */; -- Additional options for the MySQL CDC source table

Para um tutorial completo, consulte Usar um catálogo Hologres.

Para mais informações, consulte CREATE TABLE AS (CTAS) para sincronização de tabela única e tabelas fragmentadas, e CREATE DATABASE AS (CDAS) para sincronização de banco de dados completo.

Ler de uma tabela de dimensão

INSERT INTO `<othersinktable>`
SELECT ...
FROM `<othersourcetable>` AS e
JOIN `<mysqlcatalog>`.`<dbname>`.`<tablename>` FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

Gravar em uma tabela MySQL

INSERT INTO `<mysqlcatalog>`.`<dbname>`.`<tablename>`
SELECT ...
FROM `<othersourcetable>`;