Como gerenciar catálogos do Hologres - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Operações suportadas

Operação	UI	Flink SQL
Criar um catálogo	Sim	Sim
Visualizar um catálogo	Sim	—
Criar uma tabela em um catálogo	Sim	Sim
Modificar uma tabela em um catálogo	—	Sim
Ler e gravar em tabelas do catálogo	—	Sim
Usar um catálogo como destino CTAS	—	Sim
Usar um catálogo como destino CDAS	—	Sim
Excluir um catálogo	Sim	Sim

Pré-requisitos

Antes de começar, verifique se você tem:

Uma instância dedicada do Hologres (instâncias de cluster compartilhado não são suportadas)
Um banco de dados criado na instância. Consulte Criar um banco de dados

Limitações

Não é possível modificar catálogos após a criação. Para alterar a configuração de um catálogo, exclua-o e crie um novo.
Somente instâncias dedicadas do Hologres são suportadas. O Realtime Compute for Apache Flink lê e grava apenas em tabelas internas do Hologres; portanto, instâncias de cluster compartilhado não são compatíveis.

Criar um catálogo do Hologres

Importante

Não é possível modificar as configurações do catálogo após a criação. Para alterá-las, remova o catálogo existente e crie um novo.

UI

Acesse a página Catalogs.
1. Faça login no console do Realtime Compute for Apache Flink. Na coluna Actions do seu workspace, clique em Console.
2. No painel de navegação à esquerda, clique em Catalogs.
Clique em Create Catalog, selecione Hologres e clique em Next.

Configure os parâmetros do catálogo.

Parâmetro	Obrigatório	Descrição
catalog name	Sim	Apenas letras minúsculas (a–z) e dígitos (0–9). Letras maiúsculas, hifens (-), sublinhados (_) e outros caracteres especiais não são suportados.
endpoint	Sim	Para conexões na mesma VPC: acesse o console do Hologres, abra a instância de destino e copie o endpoint da Dedicated VPC na seção Network Information. Para outros tipos de rede, consulte Conectividade de rede.
username	Sim	AccessKey ID de uma conta Alibaba Cloud ou usuário RAM, ou conta personalizada no formato `BASIC$<user_name>`. A conta deve ter permissão para acessar o banco de dados do Hologres. Consulte Modelos de permissão do Hologres e Gerenciamento de usuários. Um catálogo criado com conta personalizada exibe apenas os bancos de dados acessíveis a essa conta; um catálogo criado com par de AccessKeys mostra todos os bancos de dados da instância.
password	Sim	AccessKey secret ou senha da conta personalizada. Para evitar vazamento de credenciais, armazene-as como variáveis de projeto em vez de inseri-las diretamente.
dbname	Sim	Nome de um banco de dados existente no Hologres. Se o banco de dados não existir, a criação do catálogo falhará.

Clique em OK. O catálogo aparecerá em Catalogs no painel de navegação à esquerda.

Flink SQL

No editor da página Data Query, execute a instrução CREATE CATALOG. Sintaxe:

Parâmetros:

Parâmetro	Obrigatório	Padrão	Descrição
`catalogname`	Sim	—	Nome do catálogo. Apenas letras minúsculas (a–z) e dígitos (0–9).
`type`	Sim	—	Valor fixo: `hologres`.
`endpoint`	Sim	—	Endpoint VPC da instância do Hologres. Para conexões na mesma VPC, acesse o console do Hologres e copie o endpoint VPC em Network Information. Para outros tipos de rede, consulte Conectividade de rede.
`username`	Sim	—	AccessKey ID da conta Alibaba Cloud. Para evitar vazamentos, armazene-o como variável de projeto. A conta deve ter permissão para acessar o banco de dados do Hologres. Consulte Modelo de permissões do Hologres.
`password`	Sim	—	AccessKey secret.
`dbname`	Sim	—	Nome de um banco de dados existente no Hologres.
`ignore-non-persisted-options`	Não	`true`	Define se opções não persistentes devem ser ignoradas ao criar uma tabela com tais opções. `true`: a tabela é criada e as opções não persistentes são ignoradas. `false`: a criação da tabela falha. Apenas `endpoint`, `username`, `password` e `dbname` são persistidos.
`catalog.table.metadata-columns`	Não	Nenhum	Colunas de metadados a adicionar da tabela de source de binary logging do Hologres. Separe múltiplas colunas com ponto e vírgula (`;`), por exemplo, `hg_binlog_event_type;hg_binlog_timestamp_us`. Seis tipos de metacolunas são suportados. Tabelas com este parâmetro funcionam apenas como tabelas de source, não como tabelas de destino (sink) ou dimensão. Suportado no Ververica Runtime (VVR) 8.0.11 e posterior. Consulte Campos de binary logging do Hologres.
Outros parâmetros do conector Hologres	Não	—	Parâmetros WITH suportados pelo conector Hologres. Quando definidos no nível do catálogo, aplicam-se a todas as tabelas por padrão. Requer `ignore-non-persisted-options` = `true`. Consulte Opções do conector.

CREATE CATALOG <catalogname> WITH (
  'type' = 'hologres',
  'endpoint' = '<endpoint>',
  'username' = '<AccessKey ID>',
  'password' = '<AccessKey secret>',
  'dbname' = '<dbname>'
);

Exemplos: Exemplo simples:

CREATE CATALOG holocatalog WITH (
  'type' = 'hologres',
  'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80',
  'username' = 'LTAI********************',
  'password' = '${secret_values.ak_holo}',
  'dbname' = 'holo_test'
);

Exemplo de consumo em tempo real (o binary logging deve estar habilitado na tabela do Hologres):

CREATE CATALOG holocatalog WITH (
  'type' = 'hologres',
  'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80',
  'username' = 'LTAI********************',
  'password' = '${secret_values.ak_holo}',
  'dbname' = 'holo_test',
  'binlog' = 'true',               -- Enable binary logging consumption
  'cdcmode' = 'true',
  'connectionpoolname' = 'the_conn_pool',
  'table_property.binlog.level' = 'replica',  -- Enable binary logging on tables created through this catalog
  'table_property.binlog.ttl' = '259200'
);

Clique em Run no canto superior direito. Verifique se o catálogo aparece em Catalogs no painel de navegação à esquerda.

Visualizar um catálogo do Hologres

Faça login no console do Realtime Compute for Apache Flink. Na coluna Actions do seu workspace, clique em Console e, em seguida, clique em Catalogs no painel de navegação à esquerda.
Na página Catalog List, clique em View na coluna Actions do catálogo desejado para navegar por seus bancos de dados e tabelas. Se o schema for public, o prefixo do schema será omitido do nome da tabela.

Usar um catálogo do Hologres

Notas de uso:

Se o schema for public, referencie a tabela como ${table_name} em vez de ${schema_name.table_name}.
Tabelas em um catálogo do Hologres suportam consumo de dados atualizados. Por padrão, ignoredelete é false e mutatetype é insertorupdate. Consulte Mesclar dados em uma tabela ampla.

Criar uma tabela do Hologres

Este exemplo cria uma tabela chamada holotable no banco de dados holodb dentro do catálogo holocatalog.

Antes de criar uma tabela:

O parâmetro connector é obrigatório na cláusula WITH e seu valor deve ser hologres. Outros parâmetros, como endpoint, podem ser omitidos.
Não é possível adicionar ou modificar diretamente parâmetros WITH na definição de uma tabela do Hologres. Use SQL hints nas instruções INSERT.

UI

Faça login no console do Realtime Compute for Apache Flink. Na coluna Actions do seu workspace, clique em Console e, em seguida, clique em Catalogs.
Na coluna Actions do catálogo de destino, clique em View. Em seguida, clique novamente em View para o banco de dados desejado.

Clique em Create Table.

Na aba Built-in Connector, selecione o conector Hologres e clique em Next.

Insira a instrução CREATE TABLE.

Sintaxe	Exemplo
`CREATE TABLE \`${catalog_name}`.\`${db_name}`.\`${table_name} `(...) WITH ('connector' = 'hologres');`	`CREATE TABLE \` holocatalog `.\` holo_test `.\` product `(id INT, name STRING) WITH ('connector' = 'hologres');`

Clique em Confirm.

Flink SQL

No editor da página Scripts, insira uma instrução CREATE TABLE usando uma das seguintes abordagens.

Opção 1: Usar `USE CATALOG` (recomendado)

Mude para o catálogo primeiro e depois crie a tabela sem um nome totalmente qualificado.

Sintaxe

Exemplo

USE CATALOG ${catalog_name};

CREATE TABLE \${db_name}\.\${schema_name.table_name}\(...) WITH ('connector' = 'hologres');

USE CATALOG holocatalog;

CREATE TABLE \ holodb\.\ holotable\ (id INT, name STRING) WITH ('connector' = 'hologres');

Opção 2: Usar um nome totalmente qualificado no DDL

Referencie o catálogo diretamente na instrução CREATE TABLE.

Sintaxe	Exemplo
`CREATE TABLE \`${catalog_name}\`.\`${db_name}\`.\`${schema_name.table_name}\`(...) WITH ('connector' = 'hologres');`	`CREATE TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `(id INT, name STRING) WITH ('connector' = 'hologres');`

Definir propriedades físicas da tabela

Adicione parâmetros table_property.* à cláusula WITH para controlar como o Hologres armazena a tabela. Essas propriedades são definidas durante a criação da tabela e algumas não podem ser alteradas posteriormente.

CREATE TABLE `holocatalog`.`holodb`.`holotable` (
  id INT,
  name STRING
) WITH (
  'connector' = 'hologres',
  'table_property.orientation' = 'column',
  'table_property.distribution_key' = 'a',
  'table_property.clustering_key' = 'b:desc',
  'table_property.bitmap_columns' = 'a,b',
  'table_property.segment_key' = 'c',
  'table_property.time_to_live_in_seconds' = '86400',
  'table_property.binlog.level' = 'replica',
  'table_property.binlog.ttl' = '86400'
);

Propriedades suportadas (todas usam o prefixo table_property.):

Propriedade	Descrição	Exemplo	Modificável
`table_property.orientation`	Formato de armazenamento	`'row,column'`	Não
`table_property.table_group`	Grupo de tabelas	`'table_group_xxx'`	—
`table_property.distribution_key`	Chave de distribuição	`'a,b'`	—
`table_property.clustering_key`	Chave de clustering	`'a,b:desc'`	—
`table_property.event_time_column` (anteriormente `table_property.segment_key`)	Chave de segmento	`'c,d'`	—
`table_property.bitmap_columns`	Índice bitmap	`'a:on,b:off'`	Sim
`table_property.dictionary_encoding_columns`	Codificação de dicionário	`'a:on,b:off,c:auto'`	Sim
`table_property.time_to_live_in_seconds`	TTL (tempo de vida) dos dados da tabela	`'864000'`	Sim
`table_property.binlog.level`	Habilitar binary logging	`'replica'`	Sim
`table_property.binlog.ttl`	TTL dos logs binários	`'86400'`	Sim

Para mais informações, consulte Visão geral da criação de tabelas e Assinar logs binários do Hologres.

Habilitar modo flexível (enableTypeNormalization)

O modo flexível normaliza os tipos de dados do Flink para tipos mais amplos compatíveis com o Hologres, útil em cenários de Create Table As Select (CTAS) onde a precisão dos tipos da source pode variar.

Valor	Comportamento
`false` (padrão)	Cria a tabela do Hologres usando o mapeamento de tipos exato.
`true`	Normaliza os tipos: TINYINT/SMALLINT/INT/BIGINT → BIGINT; CHAR/VARCHAR/STRING → STRING; FLOAT/DOUBLE → DOUBLE. Outros tipos seguem o mapeamento de tipos padrão.

Importante

Ative o modo flexível antes da primeira execução do job CTAS. Se você ativá-lo após a primeira execução, exclua a tabela descendente e faça uma reinicialização sem estado (stateless restart) para que a alteração tenha efeito.

Modificar uma tabela do Hologres

Operação	Sintaxe e exemplo
Modificar propriedades da tabela	`ALTER TABLE \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ `SET ('table_property.binlog.level' = 'replica', 'table_property.binlog.ttl' = '64700');` Exemplo: `ALTER TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `SET ('table_property.binlog.level' = 'replica', 'table_property.binlog.ttl' = '64700');`
Renomear uma tabela	`ALTER TABLE \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ `RENAME TO \`${catalog_name}\`.\`${db_name}\`.\`${new_table_name}\`;` Exemplo: `ALTER TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `RENAME TO \` holocatalog\`.\` holodb\`.\` new_holotable\`;`
Adicionar uma coluna	`ALTER TABLE \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ `ADD <column_name> <column_datatype> COMMENT '<column_comment>';` Exemplo: `ALTER TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `ADD address STRING COMMENT 'Address information';`
Renomear uma coluna	`ALTER TABLE \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ RENAME TO <new_column_name>; Example: `ALTER TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `RENAME address TO new_address;`
Modificar o comentário de uma coluna	`ALTER TABLE \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ `MODIFY <column_name> <original_column_type> COMMENT '<new_column_comment>';` Exemplo: `ALTER TABLE \` holocatalog\`.\` holodb\`.\` holotable\ `MODIFY new_address STRING COMMENT 'New address information';`

Ler e gravar em tabelas do Hologres

Ler dados do Hologres

Por padrão, o Flink lê tabelas de source do Hologres em modo batch e não consome novos dados escritos em tempo real. Para leitura em tempo real, use um dos métodos abaixo:

Configurar binary logging na criação do catálogo (recomendado para configurações persistentes): Habilite o binary logging ao criar o catálogo usando CREATE CATALOG. Consulte o exemplo de consumo em tempo real acima. Depois, consulte a tabela normalmente:

Sintaxe	Exemplo
`INSERT INTO ${other_sink_table} SELECT ... FROM \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\`;`	`INSERT INTO sink_table SELECT id, name FROM \` holocatalog\`.\` holodb\`.\` holotable\`;`

Usar um SQL hint (para substituições por consulta): Adicione o hint /*+ OPTIONS('binlog'='true') */ à consulta.

INSERT INTO sinktable
SELECT id, name
FROM `holocatalog`.`holodb`.`holotable`
/*+ OPTIONS ('binlog' = 'true') */;

Gravar dados no Hologres

Sintaxe	Exemplo
`INSERT INTO \`${catalog_name}\`.\`${db_name}\`.\`${table_name}\ `SELECT ... FROM ${other_source_table};`	`INSERT INTO \` holocatalog\`.\` holodb\`.\` holotable\ `SELECT id, name FROM source_table;`

Usar como destino CTAS

Importante

Durante a sincronização CTAS, o catálogo do Hologres pode reescrever automaticamente o schema de destino para garantir que os dados possam ser gravados no Hologres: uma chave primária DECIMAL é reescrita para BIGINT por padrão; colunas TIME, TIMESTAMP ou TIMESTAMP_LTZ com precisão maior que 6 são implicitamente truncadas para precisão 6. Se a reescrita DECIMAL não atender aos seus requisitos, use a sintaxe CTAS para converter o tipo para STRING e restabelecer a chave primária.

O CTAS (Create Table As Select) cria uma tabela do Hologres e a popula a partir de uma tabela de source em uma única instrução. Defina as propriedades físicas da tabela na cláusula WITH — elas são aplicadas quando a tabela de destino é criada.

CREATE TABLE IF NOT EXISTS `${catalog_name}`.`${db_name}`.`${table_name}`
WITH (
  'connector' = 'hologres'
) AS TABLE ${other_source_table};

Exemplo:

CREATE TABLE IF NOT EXISTS `holocatalog`.`holodb`.`holotable`
WITH (
  'connector' = 'hologres'
) AS TABLE source_table;

Usar como destino CDAS

Os parâmetros WITH do CDAS aplicam-se globalmente — não é possível definir propriedades diferentes para cada tabela de destino. Para definir propriedades por tabela, crie as tabelas de destino manualmente antes de iniciar o job CDAS. Consulte Criar uma tabela do Hologres para ver as propriedades físicas de tabela suportadas.

O CDAS (Create Database As Database) sincroniza um banco de dados de source inteiro com o Hologres.

CREATE DATABASE IF NOT EXISTS `${catalog_name}`.`${db_name}`
WITH (
  'sink.parallelism' = '5' -- Set the parallelism for each sink table
) AS DATABASE ${other_source_database};

Exemplo:

CREATE DATABASE IF NOT EXISTS `holocatalog`.`holodb`
WITH (
  'sink.parallelism' = '5'
) AS DATABASE source_database;

Os parâmetros WITH no CDAS aplicam-se a todas as tabelas de destino downstream. Para detalhes sobre os parâmetros suportados, consulte Tabela de destino do Hologres.

Parâmetros adicionais:

Parâmetro	Obrigatório	Padrão	Descrição
`schemaname`	Não	`public`	Schema no banco de dados de destino do Hologres para onde os dados serão sincronizados.

Excluir um catálogo do Hologres

Importante

A exclusão de um catálogo não afeta jobs em execução, mas impacta jobs ainda não publicados e aqueles que precisam ser pausados e retomados. Proceda com cautela.

UI

Faça login no console do Realtime Compute for Apache Flink. Na coluna Actions do seu workspace, clique em Console e, em seguida, clique em Catalogs no painel de navegação à esquerda.
Na página Catalog List, localize o catálogo e clique em Delete na coluna Actions.
Na caixa de diálogo de confirmação, clique em Delete.
Verifique se o catálogo não aparece mais na seção Catalogs à esquerda.

Flink SQL

No editor da página Data Query, execute o seguinte comando:
```
DROP CATALOG ${catalog_name}
```
Substitua ${catalog_name} pelo nome do catálogo conforme exibido no console do Realtime Compute for Apache Flink.
Clique com o botão direito no comando e selecione Run.
Confirme se o catálogo não aparece mais na seção Metadata à esquerda.

FAQ

Erros de consumo em tempo real: Consulte FAQ sobre erros de catálogo.
Problemas de conectividade de rede: Consulte Conectividade de rede.
Erro CTAS sobre CatalogTableProvider: Consulte O que fazer se a mensagem de erro "CREATE TABLE ... AS TABLE ... statement requires target catalog ... implements org.apache.flink.table.catalog.CatalogTableProvider interface." aparecer?

Próximos passos

Explore as opções de conector para o Hologres: Opções do conector
Crie pipelines ponta a ponta com catálogos do Hologres:
- Ingerir dados de log em data warehouses em tempo real
- Construir um data warehouse em tempo real com Flink e Hologres

Operações suportadas

Pré-requisitos

Limitações

Criar um catálogo do Hologres

UI

Flink SQL

Visualizar um catálogo do Hologres

Usar um catálogo do Hologres

Criar uma tabela do Hologres

UI

Flink SQL

Opção 1: Usar USE CATALOG (recomendado)

Opção 2: Usar um nome totalmente qualificado no DDL

Modificar uma tabela do Hologres

Ler e gravar em tabelas do Hologres

Usar como destino CTAS

Usar como destino CDAS

Excluir um catálogo do Hologres

UI

Flink SQL

FAQ

Próximos passos

Opção 1: Usar `USE CATALOG` (recomendado)