Gerenciar catálogos Paimon | Realtime Compute for Apache Flink - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Os catálogos Paimon conectam o Flink aos metadados do data lake armazenados no OSS ou DLF. Saiba como criar e gerenciar catálogos, bancos de dados e tabelas.

Antes de começar

Requisitos de versão
- VVR 8.0.5 ou posterior: Necessário para criar catálogos e tabelas Paimon.
- VVR 11.1 ou posterior: Necessário para usar o DLF como metastore.

Requisitos de armazenamento

O Paimon armazena dados e metadados no OSS. Configure seu bucket do OSS da seguinte forma:
- Classe de armazenamento: Use Standard. Classe de armazenamento.
- Configuração do bucket: Crie um bucket dedicado na mesma região do seu workspace do Flink. Não reutilize o bucket padrão do sistema para evitar conflitos de dados.
Permissões de acesso Garanta que o par de AccessKey tenha permissões de leitura e gravação no seu bucket do OSS ou diretório DLF.
Compatibilidade de formato

Os catálogos Paimon suportam apenas o formato Paimon. Outros formatos, como Lance e Iceberg, não são compatíveis.
Atualização do console Após criar ou excluir um catálogo, banco de dados ou tabela, clique em refresh para atualizar o console.

Mapeamento de versões do Paimon

Versão do Apache Paimon	VVR
1.3.1	11.5, 11.6
1,3	11,4
1,2	11,2, 11,3
1,1	11.1
1.0	8.0.11
0.9	8.0.7, 8.0.8, 8.0.9, 8.0.10

Criar um catálogo

Escolha um tipo de metastore com base no seu cenário:

Catálogo Filesystem: Armazena metadados no OSS. Ideal para implantações simples.
Catálogo DLF: Armazena metadados no Data Lake Formation (DLF). Recomendado para cenários empresariais com gerenciamento unificado de metadados.

Criar um catálogo Filesystem

Opção 1: Usar o console (Recomendado)

Acesse a página Catalogs.
1. Faça login no Portal de Gerenciamento. Localize o workspace desejado e clique em Console na coluna Actions.
2. No painel de navegação à esquerda, clique em Catalogs.
Clique em Create Catalog, selecione Apache Paimon e clique em Next.
Defina metastore como filesystem e configure as demais definições do catálogo.

Opção 2: Usar SQL

Execute o seguinte comando no editor SQL em Development > Scripts:

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

Parâmetros gerais

Item de configuração	Descrição	Obrigatório	Observações
`my-catalog`	Nome do catálogo Paimon.	Sim	Insira um nome personalizado.
`type`	Tipo do catálogo.	Sim	Defina como `paimon`.
`metastore`	Tipo do metastore.	Sim	Valores válidos: `filesystem`: Armazena metadados no OSS. `dlf`: Armazena metadados no DLF.

Parâmetros do OSS

Item de configuração	Descrição	Obrigatório	Observações
warehouse	Diretório do data warehouse no OSS.	Sim	O formato é `oss://<bucket>/<object>`, onde: `bucket`: nome do seu bucket do OSS. `object`: caminho onde seus dados estão armazenados. Visualize os nomes do bucket e do objeto no console do OSS.
fs.oss.endpoint	Endpoint do serviço OSS.	Sim	Se o Flink e o DLF estiverem na mesma região, use o endpoint de VPC. Caso contrário, use o endpoint de rede pública. Este parâmetro é obrigatório se o bucket do OSS especificado para warehouse não estiver na mesma região do workspace do Flink ou se você usar um bucket do OSS pertencente a outra conta Alibaba Cloud. Regiões e Endpoints e Criar um AccessKey.
fs.oss.accessKeyId	AccessKey ID da conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.	Sim
fs.oss.accessKeySecret	AccessKey secret da conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.	Sim

Criar um catálogo DLF

DLF

Crie um catálogo no console do DLF na mesma região do seu workspace do Flink. Uso rápido do DLF.

Registre o catálogo Paimon no Console de Desenvolvimento.

Nota

Esta operação mapeia seu catálogo DLF apenas como metadados. Criar ou excluir este catálogo no Realtime Compute for Apache Flink não afeta os dados subjacentes no DLF.

Faça login no console do Realtime Compute.
Na lista de workspaces totalmente gerenciados, clique no nome do seu workspace para abrir o console correspondente.

Registre o catálogo usando a interface ou executando comandos SQL.

UI

No painel de navegação à esquerda, clique em Catalogs.
Na página Catalog List, clique em Create Catalog.
Selecione Apache Paimon e clique em Next.
Defina metastore como dlf. Em catalog name, selecione o catálogo DLF desejado. Clique em OK para criar o catálogo.

Comandos SQL

No editor de Scripts, insira e execute a seguinte instrução SQL para criar o catálogo.

CREATE CATALOG `flink_catalog_name` 
WITH (
  'type' = 'paimon',
  'metastore' = 'rest',
  'token.provider' = 'dlf',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com',
  'warehouse' = 'dlf_test'
);

A tabela a seguir descreve os parâmetros.

Parâmetro	Descrição	Obrigatório	Exemplo
type	Tipo do catálogo. Este parâmetro é fixo como `paimon`.	Sim	paimon
metastore	Tipo do metastore. Este parâmetro é fixo como `rest`.	Sim	rest
token.provider	Provedor de token. Este parâmetro é fixo como `dlf`.	Sim	dlf
uri	URI para acessar o DLF REST Catalog Server. O formato é `http://[region-id]-vpc.dlf.aliyuncs.com`. Para mais informações sobre IDs de região, consulte Endpoints.	Sim	http://cn-hangzhou-vpc.dlf.aliyuncs.com
warehouse	Nome do catálogo DLF subjacente a ser usado como warehouse.	Sim	dlf_test

DLF-Legacy

Crie um catálogo no console do DLF na mesma região do seu workspace do Flink. Introdução.

Registre o catálogo Paimon no Console de Desenvolvimento.

Opção 1: Usar o console (Recomendado)

Acesse a página Catalogs.
1. Faça login no Portal de Gerenciamento e clique em Console na coluna Actions do workspace desejado.
2. Clique em Catalogs.
Clique em Create Catalog. Selecione Apache Paimon e clique em Next.
Defina metastore como DLF e selecione o catálogo DLF V1.0 a ser associado na lista suspensa catalog name. Em seguida, configure as demais definições do catálogo.

Opção 2: Usar SQL

Execute o seguinte comando no editor SQL em Development > Scripts:

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'dlf',
  'warehouse' = '<warehouse>',
  'dlf.catalog.id' = '<dlf.catalog.id>',
  'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
  'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
  'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
  'dlf.catalog.region' = '<dlf.catalog.region>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

Descrição dos parâmetros:

Geral

Item de configuração	Descrição	Obrigatório	Observações
my-catalog	Nome do catálogo Paimon.	Sim	Insira um nome personalizado.
type	Tipo do catálogo.	Sim	O valor é fixo como `paimon`.
metastore	Tipo do metastore.	Sim	O valor é fixo como `dlf`.

OSS

Item de configuração	Descrição	Obrigatório	Observações
warehouse	Diretório do data warehouse no OSS.	Sim	O formato é `oss://<bucket>/<object>`, onde: `bucket`: nome do seu bucket do OSS. `object`: caminho onde seus dados estão armazenados. Visualize os nomes do bucket e do objeto no console do OSS.
fs.oss.endpoint	Endpoint do serviço OSS.	Sim	Se o Flink e o DLF estiverem na mesma região, use o endpoint de VPC. Caso contrário, use o endpoint de rede pública. Para armazenar tabelas Paimon no OSS-HDFS, defina o valor do parâmetro `fs.oss.endpoint` como `cn-<region>.oss-dls.aliyuncs.com`, por exemplo, `cn-hangzhou.oss-dls.aliyuncs.com`.
fs.oss.accessKeyId	AccessKey ID da conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.	Sim	Para mais informações, consulte Regiões e endpoints e Criar um AccessKey.
fs.oss.accessKeySecret	AccessKey secret da conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.	Sim

DLF

Item de configuração	Descrição	Obrigatório	Observações
dlf.catalog.id	ID do catálogo de dados DLF.	Sim	Visualize o ID do catálogo de dados no console do Data Lake Formation.
dlf.catalog.accessKeyId	AccessKey ID necessário para acessar o serviço DLF.	Sim	Para mais informações, consulte Criar um AccessKey.
dlf.catalog.accessKeySecret	AccessKey secret necessário para acessar o serviço DLF.	Sim	Para mais informações, consulte Criar um AccessKey.
dlf.catalog.endpoint	Endpoint do serviço DLF.	Sim	Regiões e endpoints. Nota Se o Flink e o DLF estiverem na mesma região, use o endpoint de VPC. Caso contrário, use o endpoint de rede pública.
dlf.catalog.region	Região onde o DLF está localizado.	Sim	Regiões e endpoints. Nota Certifique-se de que a região seja a mesma especificada para dlf.catalog.endpoint.

Gerenciar bancos de dados

Execute os seguintes comandos no editor SQL em Development > Scripts.

Criar um banco de dados

Um banco de dados default é criado automaticamente. Para criar mais bancos de dados:

-- Replace my-catalog with the name of your Paimon catalog.
USE CATALOG `my-catalog`;

-- Replace my_db with a custom database name in English.
CREATE DATABASE `my_db`;

Excluir um banco de dados

Importante

O banco de dados default não pode ser excluído de um catálogo DLF. No entanto, é possível excluí-lo de um catálogo Filesystem.

-- Replace my-catalog with the name of your Paimon catalog.
USE CATALOG `my-catalog`;

-- Replace my_db with the name of the database that you want to delete.
DROP DATABASE `my_db`; -- Deletes a database only if it contains no tables.
DROP DATABASE `my_db` CASCADE; -- Deletes the database and all tables in it.

Gerenciar tabelas

Criar uma tabela

Nota

Após registrar um catálogo Paimon, referencie suas tabelas diretamente nos jobs do Flink sem criar definições DDL. Para referenciar uma tabela:

Use o nome totalmente qualificado: ${Paimon-catalog-name}.${Paimon-db-name}.${Paimon-table-name}
Ou defina o catálogo e o banco de dados atuais usando USE CATALOG ${Paimon-catalog-name} e USE ${Paimon-db-name}. Em seguida, referencie as tabelas usando apenas seus nomes: ${Paimon-table-name}

Criar uma tabela usando a instrução CREATE TABLE

Execute o seguinte comando no editor SQL em Development > Scripts.

O exemplo a seguir cria uma tabela particionada no banco de dados my_db do my-catalog.

-- Replace my-catalog with the name of your Paimon catalog.
-- Replace my_db with the name of the database that you want to use.
-- You can also replace my_tbl with a custom name in English.
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket' = '4'
);

Para mais informações, consulte Conector Apache Paimon e Tabelas de chave primária e somente anexação.

Criar uma tabela usando a instrução CREATE TABLE AS (CTAS) ou CREATE DATABASE AS (CDAS)****

CTAS e CDAS sincronizam automaticamente dados e alterações de schema, replicando tabelas de fontes como MySQL e Kafka para o Paimon.

Nota

Habilitar bucketing dinâmico definindo 'bucket' = '-1' não é compatível com Tabelas de chave primária ou Tabelas de chave primária e somente anexação.
CTAS e CDAS permitem configurar propriedades downstream replicáveis. Conector Paimon.

Replicar uma tabela específica

O exemplo a seguir cria automaticamente uma tabela Paimon my-catalog com base no schema da tabela mysql.tpcds.web_sales e replica os dados.
```
CREATE TABLE IF NOT EXISTS `<catalog name>`.`<db name>`.`<table name>`
WITH (
  'bucket' = '4',
  'changelog-producer' = 'input'
) AS TABLE mysql.tpcds.web_sales;
```
Replicar um banco de dados inteiro

O exemplo a seguir cria automaticamente tabelas Paimon no banco de dados my-catalog com base no schema de cada tabela no banco de dados mysql.tpcds e replica os dados.
```
CREATE DATABASE IF NOT EXISTS `<catalog name>`.`<db name>`
WITH (
  'changelog-producer' = 'input'
) AS DATABASE mysql.tpcds INCLUDING ALL TABLES;
```
Replicar alterações de tipo de coluna

Tabelas Paimon criadas usando instruções CTAS ou CDAS suportam alterações específicas de tipo de coluna.
- Replicação de tipo padrão
  
  Por padrão, as tabelas Paimon preservam os tipos exatos de coluna da origem. No entanto, certas promoções de tipo são suportadas sem configuração explícita:
  - Tipos inteiros: TINYINT, SMALLINT, INT e BIGINT podem ser promovidos para um tipo inteiro de precisão igual ou superior. (TINYINT < SMALLINT < INT < BIGINT).
  - Tipos de ponto flutuante: FLOAT e DOUBLE podem ser promovidos para um tipo de ponto flutuante de precisão igual ou superior. (FLOAT < DOUBLE).
  - Tipos de string: CHAR, VARCHAR e STRING podem ser promovidos para um tipo de string de precisão igual ou superior.
- Normalização de tipo
  
  Para permitir alterações de tipo mais flexíveis e evitar falhas de job devido a evoluções de schema compatíveis, defina 'enableTypeNormalization' = 'true' na cláusula WITH da sua instrução CTAS/CDAS. Neste modo, o Flink normaliza os tipos de dados durante a evolução do schema. Um job falhará apenas se os tipos antigo e novo não puderem ser normalizados para o mesmo tipo subjacente. Regras:
  - TINYINT, SMALLINT, INT, BIGINT são normalizados para BIGINT.
  - FLOAT, DOUBLE são normalizados para DOUBLE.
  - CHAR, VARCHAR, STRING são normalizados para STRING.
  - Outros tipos de dados não são normalizados.
- Impacto da normalização:
  
  Quando a normalização de tipo está ativa, os tipos de dados armazenados na tabela Paimon serão do tipo normalizado. Por exemplo, se as colunas de origem c1 (SMALLINT) e c2 (INT) forem ambas normalizadas para BIGINT, tanto c1 quanto c2 serão armazenadas como BIGINT na tabela Paimon.
- Exemplos:
  - Sucesso: Se SMALLINT na origem mudar para INT, ambos os tipos normalizam para BIGINT. Esta é uma alteração compatível e o job continua normalmente.
  - Falha: Se FLOAT na origem mudar para BIGINT, seus tipos normalizados (DOUBLE e BIGINT) são incompatíveis, o que causará uma exceção.

Modificar um schema de tabela

Execute os seguintes comandos no editor SQL em Development > Scripts.

Operação	Código de exemplo
Adicionar ou modificar propriedades da tabela	`ALTER TABLE my_table SET ( 'write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true' );`
Modificar temporariamente propriedades da tabela	As propriedades da tabela modificadas temporariamente têm efeito apenas para o job SQL atual. Ao gravar na tabela my_table, defina temporariamente `write-buffer-size` como 256 MB e `write-buffer-spillable` como true. `INSERT INTO my_table /+ OPTIONS('write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true') / SELECT ...;` Ao consumir dados da tabela my_table, defina temporariamente `scan.mode` como latest e `scan.parallelism` como 10. `SELECT * FROM t /+ OPTIONS('scan.mode' = 'latest', 'scan.parallelism' = '10') /;`
Renomear uma tabela	`ALTER TABLE my_table RENAME TO my_table_new;` Importante As operações de renomeação do OSS não são atômicas. Se você armazenar arquivos de tabela Paimon no OSS, use o OSS-HDFS para operações atômicas de arquivo.
Adicionar uma nova coluna	Adicione a coluna `c1` do tipo INT e a coluna `c2` do tipo STRING ao final da tabela `my_table`. `ALTER TABLE my_table ADD (c1 INT, c2 STRING);` Adicione a coluna `c2` do tipo STRING após a coluna `c1` na tabela `my_table`. `ALTER TABLE my_table ADD c2 STRING AFTER c1;` Adicione a coluna `c1` do tipo INT no início da tabela `my_table`. `ALTER TABLE my_table ADD c1 INT FIRST;`
Renomear uma coluna	Renomeie a coluna c0 para c1 em `my_table`. `ALTER TABLE my_table RENAME c0 TO c1;`
Remover uma coluna	Remova as colunas `c1` e `c2` de `my_table`. `ALTER TABLE my_table DROP (c1, c2);`
Remover uma partição	Remova as partições `dt=20240108,hh=06` e `dt=20240109,hh=07` de `my_table`. ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');
Modificar um comentário de coluna	Altere o comentário da coluna `buy_count` em `my_table` para this is buy count. `ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';`
Modificar ordem das colunas	Mova `col_a` do tipo DOUBLE para o início de `my_table`. `ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;` Mova `col_a` do tipo DOUBLE para depois de `col_b` em `my_table`. `ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;`
Modificar um tipo de coluna	Altere o tipo de `col_a` em `my_table` para DOUBLE. `ALTER TABLE my_table MODIFY col_a DOUBLE;` Conversões de tipo de coluna compatíveis (〇 = compatível, vazio = não compatível):

Excluir uma tabela

Execute o seguinte comando no editor SQL em Development > Scripts.

-- Replace my-catalog with the name of your Paimon catalog.
-- Replace my_db with the name of the database that you want to use.
-- Replace my_tbl with the name of the Paimon catalog table that you created.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

Se a mensagem The following statement has been executed successfully! for retornada, a tabela Paimon foi excluída.

Visualizar ou excluir um catálogo

No Portal de Gerenciamento, clique em Console na coluna Actions do workspace desejado.
Na página Catalogs, visualize ou exclua um catálogo Apache Paimon.
- Na página Catalogs, visualize o nome e o tipo do catálogo. Para visualizar os bancos de dados e tabelas em um catálogo, clique em View na coluna Actions.
- Na página Catalogs, clique em Delete na coluna Actions do catálogo a ser excluído.
  
  Nota
  Excluir um catálogo remove apenas seu registro no Flink. Os arquivos de dados subjacentes não são afetados. Registre o catálogo novamente para restaurar o acesso.
  
  Alternativamente, no editor SQL em Development > Scripts, insira DROP CATALOG <catalog name>;, selecione o código e clique em Run.
Referências

Antes de começar

Criar um catálogo

Criar um catálogo Filesystem

Opção 1: Usar o console (Recomendado)

Opção 2: Usar SQL

Criar um catálogo DLF

DLF

UI

Comandos SQL

DLF-Legacy

Opção 1: Usar o console (Recomendado)

Opção 2: Usar SQL

Gerenciar bancos de dados

Gerenciar tabelas

Criar uma tabela

Modificar um schema de tabela

Excluir uma tabela

Visualizar ou excluir um catálogo

Referências