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

Pré-requisitos

É necessário ter uma instância dedicada do Hologres com um banco de dados. Para mais informações, consulte Create a database.

Limitações

Catálogos não podem ser modificados. Para aplicar alterações, exclua o catálogo existente e crie um novo.
É obrigatório usar uma instância dedicada do Hologres. O Realtime Compute for Apache Flink não oferece suporte a instâncias de cluster compartilhado, pois só consegue acessar tabelas internas do Hologres.

Criar um catálogo do Hologres

Importante

A configuração de um catálogo não pode ser alterada após a criação. Para fazer mudanças, exclua o catálogo existente e crie um novo.

UI

Para definir parâmetros compatíveis com o conector do Hologres, crie o catálogo usando SQL.

Acesse a página Data Management.
1. Faça login no Realtime Compute for Apache Flink console e clique em Console na coluna Actions do workspace desejado.
2. Clique em Catalogs.
Clique em Create Catalog, selecione Hologres e clique em Next.

Configure os parâmetros.

Parâmetro	Descrição	Obrigatório	Observações
catalogname	O nome do catálogo do Hologres.	Sim	Somente letras minúsculas (a-z) e dígitos (0-9) são aceitos. Letras maiúsculas, hifens (-), underscores (_) e outros caracteres especiais não são permitidos.
endpoint	O endpoint de rede da instância do Hologres.	Sim	Se a instância do Hologres e o workspace do Flink estiverem no mesmo VPC, acesse o Hologres console. Na página de detalhes da instância desejada, obtenha as informações de rede de Specified VPC na seção Network Information. Para outros tipos de rede, consulte Obtain the endpoint of a Hologres instance.
username	O nome de usuário de uma conta personalizada, no formato `BASIC$<user_name>`. O AccessKey ID de uma conta Alibaba Cloud ou de um usuário RAM.	Sim	O usuário configurado deve ter permissões para acessar o banco de dados do Hologres correspondente. Para mais informações, consulte Hologres permission model e user management. Para saber como obter um AccessKey ID e um AccessKey Secret, consulte Obtain an AccessKey pair. Importante Para evitar o vazamento de sua AK ou senha, recomendamos usar variáveis de projeto para inserir esses valores. Para mais informações, consulte Project variables. Um catálogo criado com uma conta personalizada exibe apenas os bancos de dados aos quais a conta tem permissão. Um catálogo criado com um AccessKey exibe todos os bancos de dados da instância.
password	A senha da conta personalizada. O AccessKey Secret da conta Alibaba Cloud ou do usuário RAM.	Sim
dbname	O nome do banco de dados do Hologres.	Sim	O banco de dados já deve existir na instância do Hologres de destino. Caso contrário, a criação do catálogo falhará.

Clique em OK.

Após a criação, o novo catálogo ficará disponível em Catalogs.

SQL

No editor da página Data Query, insira o comando para criar um catálogo do Hologres.

Sintaxe:

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

Exemplos:

Exemplo simples

Exemplo de consumo em tempo real

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'
);

O Binlog deve estar ativado para suportar o consumo de dados de binlog.

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', -- When you create the catalog, you can set WITH parameters for source, dimension, and sink tables. These parameters are then applied by default to tables accessed through this catalog.
  'cdcmode' = 'true',
  'connectionpoolname' = 'the_conn_pool',
  'table_property.binlog.level' = 'replica', -- You can also pass persisted Hologres table properties when creating the catalog. Then, binlog is enabled by default for new tables.
  'table_property.binlog.ttl' = '259200'
);

A tabela a seguir descreve os parâmetros.

Parâmetro	Descrição	Obrigatório	Observações
catalogname	O nome do catálogo do Hologres.	Sim	Somente letras minúsculas (a-z) e dígitos (0-9) são aceitos. Letras maiúsculas, hifens (-), underscores (_) e outros caracteres especiais não são permitidos.
type	O tipo do catálogo.	Sim	O valor deve ser `hologres`.
endpoint	O endpoint da instância do Hologres.	Sim	Se a instância do Hologres e o workspace do Flink estiverem no mesmo VPC, acesse o Hologres management console e obtenha as informações de rede de specified VPC na seção Network Information da página de detalhes da instância desejada. Para outros tipos de rede, consulte Obtain the endpoint of a Hologres instance.
username	O AccessKey ID da sua conta Alibaba Cloud ou do usuário RAM.	Sim	Para mais informações, consulte Obtain an AccessKey pair. Nota Para evitar o vazamento das informações de AK, recomendamos usar uma variável para especificar o valor do parâmetro password. Para mais informações, consulte Project variables. O usuário correspondente ao AccessKey especificado deve ter permissões para acessar o banco de dados do Hologres. Para mais informações sobre permissões de banco de dados, consulte Hologres permission model.
password	O AccessKey Secret da sua conta Alibaba Cloud ou do usuário RAM.	Sim
dbname	O nome do banco de dados do Hologres.	Sim	O banco de dados já deve existir na instância do Hologres de destino. Caso contrário, a criação do catálogo falhará.
ignore-non-persisted-options	Define o comportamento ao criar uma tabela com opções não persistidas por meio do catálogo.	Não	Valores válidos: `true` (padrão): a tabela é criada e todas as opções não persistidas são ignoradas. `false`: um erro é reportado e a criação da tabela falha. Nota Uma opção de tabela persistida é aquela cujo valor, definido na instrução DDL, é armazenado no catálogo do Hologres e pode ser recuperado posteriormente. Atualmente, somente as opções `endpoint`, `username`, `password` e `dbname` são persistidas.
catalog.table.metadata-columns	Ao usar este catálogo para especificar uma tabela source, é possível adicionar colunas metadata da tabela source de binlog do Hologres ao esquema da tabela conforme necessário. Separe múltiplas colunas de metadados com ponto e vírgula (;). Exemplo: `hg_binlog_event_type;hg_binlog_timestamp_us`.	Não	Por padrão, nenhuma coluna de metadados é adicionada. Ao configurar este parâmetro, as colunas de metadados especificadas são incluídas no esquema de tabela retornado. Essas colunas se aplicam apenas a tabelas source de binlog do Hologres; portanto, tabelas deste catálogo podem ser usadas apenas como tabelas source, não como tabelas sink ou de dimensão. Há suporte para seis tipos de colunas de metadados. Para mais informações, consulte Hologres binlog fields. Nota Este parâmetro é compatível somente com Ververica Runtime (VVR) 8.0.11 e versões posteriores.
Outros parâmetros compatíveis com o conector do Hologres	É possível especificar outros parâmetros, incluindo WITH options, ao criar o catálogo. Esses parâmetros passam a ser as configurações padrão para todas as tabelas acessadas por meio deste catálogo.	Não	Para usar este recurso, defina o parâmetro `ignore-non-persisted-options` como `true`.

Após inserir o comando de criação do catálogo, clique em Run no canto superior direito.

Catálogo do Hologres

Após configurar um catálogo do Hologres, siga estas etapas para visualizar seus metadados.

Acesse a página Catalogs.
1. Faça login no Realtime Compute for Apache Flink console.
2. Na coluna Actions do workspace desejado, clique em Console.
3. No painel de navegação à esquerda, clique em Catalogs.
Na página Catalog List, localize seu catálogo pelo Name e pelo Type.

Clique em View para ver os bancos de dados e tabelas do catálogo. Se o schema for público, o nome da tabela aparecerá sem prefixo de schema.

Usar um Hologres Catalog

Observações:

Se o schema for público, ao inserir ${schema_name.table_name}, você pode omitir o prefixo do nome do schema e inserir apenas o nome da tabela como ${table_name}.
As tabelas em um Hologres Catalog podem consumir dados do tipo UPDATE. Para essas tabelas, a propriedade ignoredelete tem valor padrão false e a propriedade mutatetype tem valor padrão insertorupdate. Para mais informações sobre essas propriedades, consulte Merge data into a wide table and update data partially.

Criar uma tabela Hologres

Este exemplo cria uma tabela chamada holotable no banco de dados holodb do catalog holocatalog.

Nota

Ao criar uma tabela em um service de dados Hologres registrado, o parâmetro connector é obrigatório na cláusula WITH, e seu valor deve ser hologres. Os demais parâmetros, como endpoint, podem ser omitidos.
Não é possível adicionar ou modificar diretamente os parâmetros WITH suportados em uma tabela Hologres. Para adicionar ou modificar esses parâmetros, utilize SQL hints em uma instrução INSERT.

UI

Acesse a página Catalogs.
1. Faça login no Realtime Compute for Apache Flink console.
2. Na coluna Actions do workspace de destino, clique em Console.
3. Clique em Catalogs.
Acesse o banco de dados de destino.
1. Na coluna Actions do catalog de destino, clique em View.
2. Na coluna Actions do banco de dados de destino, clique em View.

Crie a tabela Hologres.

Clique em Create Table.
Na aba Built-in, selecione o conector Hologres e clique em Next.

Insira a instrução CREATE TABLE e configure os parâmetros. Veja o exemplo a seguir.

Syntax

Example

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

SQL

No editor da página Data Query, insira a instrução CREATE TABLE.

É possível criar uma tabela Hologres de uma das seguintes formas:

`USE CATALOG`

Utilize uma instrução USE para referenciar o Hologres Catalog.

Syntax

Example

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'
);

DDL

Referencie o Hologres Catalog na instrução DDL.

Syntax

Example

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'
);

Também é possível definir propriedades de tabela física na instrução DDL.

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'
);

Após inserir a instrução, clique em Run no canto superior direito para criar a tabela.

Ao criar uma tabela Hologres, é possível especificar propriedades de tabela na cláusula WITH. Algumas propriedades não podem ser modificadas posteriormente. Defini-las corretamente melhora a organização dos dados e o desempenho das consultas.

Propriedades de tabela física: as propriedades suportadas são equivalentes às do Hologres, exceto pelo fato de que é necessário adicionar o prefixo table_property.. Para mais informações sobre os parâmetros, consulte Overview of table creation e Subscribe to Hologres binlogs.

Parâmetro	Descrição	Exemplo	Modificável
table_property.orientation	Define o formato de armazenamento da tabela.	'table_property.orientation' = 'row,column'	Não
table_property.table_group	Define o grupo de tabelas.	'table_property.table_group' = 'table_group_xxx'
table_property.distribution_key	Define a chave de distribuição.	'table_property.distribution_key' = 'a,b'
table_property.clustering_key	Define a chave de clustering.	'table_property.clustering_key' = 'a,b:desc'
table_property.event_time_column (formerly table_property.segment_key)	Define a chave de segmento.	'table_property.event_time_column' = 'c,d' (or 'table_property.segment_key' = 'c,d')
table_property.bitmap_columns	Define o índice bitmap.	'table_property.bitmap_columns' = 'a:on,b:off'	Sim
table_property.dictionary_encoding_columns	Define a codificação por dicionário.	'table_property.dictionary_encoding_columns' = 'a:on,b:off,c:auto'
table_property.time_to_live_in_seconds	Define o tempo de vida (TTL) dos dados da tabela.	'table_property.time_to_live_in_seconds' = '864000'
table_property.binlog.level	Ativa ou desativa o binlog.	'table_property.binlog.level' = 'replica'
table_property.binlog.ttl	Define o TTL do binlog.	'table_property.binlog.ttl' = '86400'

Parâmetro de modo leniente: para ativar o modo leniente ao criar uma tabela com um Hologres catalog, configure o parâmetro enableTypeNormalization. Em um job de ingestão de dados do Flink CDC, ative o modo leniente definindo sink.type-normalize-strategy.

Item	Descrição
Caso de uso	Em cenários de sincronização de dados, pode ser necessário ajustar a precisão do tipo de dado de um campo existente (por exemplo, de VARCHAR(10) para VARCHAR(20)) ou alterar o tipo de dado (por exemplo, de SMALLINT para INT).
Observações	O modo leniente deve ser ativado na primeira execução do job. Se não for ativado na primeira execução, será necessário excluir a tabela downstream e reiniciar o job sem estado para que a alteração entre em vigor. Após ativar o modo leniente, se o tipo de dado de uma coluna upstream for alterado, o job continuará sendo executado desde que os tipos novo e original compartilhem o mesmo tipo normalizado. Caso contrário, os tipos serão considerados incompatíveis e o job lançará uma exceção.
Valores	`false` (padrão): cria uma tabela física Hologres com base no mapeamento de tipos padrão. `true`: ativa o modo leniente. Ao criar uma tabela física Hologres, utiliza-se um tipo de dado de maior precisão após a normalização de tipos. As regras de normalização vigentes são as seguintes: TINYINT, SMALLINT, INT e BIGINT são normalizados para BIGINT. CHAR, VARCHAR e STRING são normalizados para STRING. FLOAT e DOUBLE são normalizados para DOUBLE. Os demais tipos de dados são criados com base nas regras de mapeamento de tipos originais. Para mais informações, consulte Type mapping.

Modificar uma tabela Hologres

Um Hologres Catalog suporta as seguintes operações de modificação de tabela:

Operação	Sintaxe e Exemplo
Modificar propriedades da tabela	Apenas algumas propriedades de tabela podem ser modificadas. Para mais informações, consulte Create a Hologres table. Sintaxe: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_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	Sintaxe: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` RENAME TO `${catalog_name}`.`${db_name}`.`${schema_name.new_table_name}`; Exemplo: ALTER TABLE `holocatalog`.`holodb`.`holotable` RENAME TO `holocatalog`.`holodb`.`new_holotable`;
Adicionar uma coluna	Sintaxe: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_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	Sintaxe: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` RENAME <old_column_name> TO <new_column_name>; Exemplo: ALTER TABLE `holocatalog`.`holodb`.`holotable` RENAME address TO new_address;
Modificar o comentário de uma coluna	Sintaxe: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_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 dados em tabelas Hologres

Leia dados de uma tabela Hologres e grave-os em uma tabela de destino.

Por padrão, o Flink lê tabelas source do Hologres no modo batch e não processa novos dados em tempo real. Para transmitir dados do Hologres em tempo real, utilize um dos seguintes métodos:

Configure o Hologres Catalog na criação: ao criar o catalog via SQL, é possível ativar o binlog para consumir dados de binlog. Para mais informações, consulte Real-time consumption example. Em seguida, leia os dados do Hologres. O código a seguir apresenta um exemplo.

Syntax	Example
INSERT INTO ${other_sink_table} SELECT ... FROM `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`;	INSERT INTO sink_table SELECT id, name FROM `holocatalog`.`holodb`.`holotable`;

Alterne para o modo stream usando um table hint. Adicione o hint /*+ OPTIONS('binlog'='true') */ à instrução SQL. O código a seguir apresenta um exemplo.
```
INSERT INTO sinktable
SELECT id, name
  FROM `holocatalog`.`holodb`.`holotable`
  /*+ OPTIONS ('binlog' = 'true') */;
```

Grave dados de uma tabela source em uma tabela Hologres.

Syntax	Example
INSERT INTO `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` SELECT ... FROM ${other_source_table}	INSERT INTO `holocatalog`.`holodb`.`holotable` SELECT id, name FROM source_table;

Destino de ingestão de dados do Flink CDC

Exemplo

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  tables: tpcds.web_sales
  server-id: <server id>
  
sink:
  type: hologres
  using.built-in-catalog: hologres_catalog

Os jobs de ingestão de dados do Flink CDC permitem definir propriedades de tabela física no destino. Essas propriedades são aplicadas quando a tabela de destino é criada. Para mais informações sobre as propriedades de tabela suportadas, consulte Create a Hologres table.

Durante a sincronização de dados a partir de uma source, o Hologres Catalog pode reescrever o schema da tabela de destino para garantir que os dados sejam gravados corretamente no Hologres. Isso ocorre nos seguintes casos:

O schema da source utiliza uma coluna do tipo DECIMAL como chave primária.

Como o Hologres não suporta DECIMAL como chave primária, o tipo da coluna é reescrito para BIGINT por padrão. Se essa alteração não atender aos seus requisitos, utilize o módulo de transformação para converter a coluna referenciada para o tipo STRING e recrie a chave primária.
O schema da source contém uma coluna do tipo TIME, TIMESTAMP ou TIMESTAMP_LTZ com precisão superior a 6.

O Hologres suporta precisão de até 6 para tipos de tempo. Para garantir que os dados sejam gravados corretamente, o Flink descarta implicitamente os dígitos que excedem a precisão máxima suportada pelo Hologres.

Excluir um Hologres catalog

Importante

Excluir um Hologres catalog não afeta os jobs em execução, mas impacta jobs não publicados ou jobs que precisem ser pausados e retomados. Prossiga com cautela.

UI

Acesse a página Catalogs.
1. Faça login no Realtime Compute for Apache Flink console.
2. Na coluna Actions do workspace de destino, clique em Console.
3. No painel de navegação à esquerda, clique em Catalogs.
Na página Catalog List, localize o catalog de destino e clique em Delete na coluna Actions.
No diálogo de confirmação, clique em Delete.
Confirme que o catalog de destino não aparece mais na seção Catalogs à esquerda.

Flink SQL

No editor de texto da página Data Query, insira o seguinte comando.
```
DROP CATALOG ${catalog_name}
```
Em que ${catalog_name} é o nome do Hologres Catalog a ser excluído, conforme exibido no Realtime Compute for Apache Flink console.
Selecione o comando, clique com o botão direito e escolha Run.
Confirme que o catalog de destino não aparece mais na seção Catalogs à esquerda.

Perguntas frequentes

Consumo em tempo real de dados do Hologres: What to do when a job reading from a Hologres source table using a Hologres Catalog completes automatically instead of running continuously?
Problemas de conectividade de rede: How do I find the endpoint of a Hologres instance?
Limitações de armazenamento de source e sink: Why do I receive the error "CREATE TABLE ... AS TABLE ... statement requires target catalog ... implements org.apache.flink.table.catalog.CatalogTableProvider interface."?

Documentos relacionados

Parâmetros WITH do Hologres: WITH parameters
Casos de uso do Hologres Catalog:
- Real-time log ingestion
- Building a Hologres real-time data warehouse