Conector do PolarDB for PostgreSQL - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Importante

O conector do PolarDB for PostgreSQL (compatível com Oracle) 1.0 está sendo descontinuado e será removido em uma versão futura. Ele não receberá mais atualizações nem manutenção, e suas opções de configuração serão removidas do console. Para obter detalhes, consulte End-of-support (EOS) for the PolarDB for PostgreSQL (Compatible with Oracle) 1.0 connector. Migre suas cargas de trabalho prontamente para evitar interrupções no service.

Tipo suportado: Sink | Modos de execução: Streaming e batch | Tipo de API: SQL | Suporte a atualização e exclusão: Suportado

Use este conector para gravar dados do Flink em um cluster PolarDB for PostgreSQL (Compatible with Oracle) 1.0. O PolarDB for PostgreSQL (compatível com Oracle) é um banco de dados nativo da cloud baseado em arquitetura de separação entre armazenamento e computação, com alta compatibilidade com Oracle.

Pré-requisitos

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

Um cluster e uma tabela do PolarDB for PostgreSQL (compatível com Oracle) 1.0. Consulte Criar um cluster do PolarDB for PostgreSQL (compatível com Oracle) e Criar uma tabela
Uma lista de permissões configurada para o cluster. Consulte Configure a cluster whitelist

Limitações

Este conector suporta apenas o PolarDB 1.0. Para o PolarDB 2.0, use o conector JDBC.
Este conector requer o Ververica Runtime (VVR) 8.0.5 ou superior.

Sintaxe

CREATE TABLE polardbo_table (
  id INT,
  len INT,
  content VARCHAR,
  PRIMARY KEY(id)
) WITH (
  'connector'='polardbo',
  'url'='jdbc:postgresql://<Address>:<PortId>/<DatabaseName>',
  'tableName'='<yourDatabaseTableName>',
  'userName'='<yourDatabaseUserName>',
  'password'='<yourDatabasePassword>'
);

Opções do conector

Opção	Tipo	Obrigatório	Padrão	Descrição
`connector`	String	Sim	N/A	Deve ser `polardbo`.
`url`	String	Sim	N/A	URL JDBC do banco de dados. Formato: `jdbc:postgresql://<Address>:<PortId>/<DatabaseName>`.
`tableName`	String	Sim	N/A	Nome da tabela do banco de dados.
`userName`	String	Sim	N/A	Nome de usuário para conexão ao banco de dados.
`password`	String	Sim	N/A	Senha para conexão ao banco de dados. Armazene as credenciais usando project variables em vez de codificá-las em texto simples.
`maxRetryTimes`	Integer	Não	3	Número máximo de tentativas para uma operação de gravação com falha.
`targetSchema`	String	Não	public	Nome do schema.
`caseSensitive`	String	Não	false	Define se os nomes de tabelas e campos diferenciam maiúsculas de minúsculas. Valores válidos: `true`, `false`.
`connectionMaxActive`	Integer	Não	5	Quantidade máxima de conexões ativas no pool de conexões. O sistema libera automaticamente as conexões ociosas. Definir esse valor muito alto pode sobrecarregar o servidor de banco de dados.
`retryWaitTime`	Integer	Não	100	Intervalo entre novas tentativas, em milissegundos.
`batchSize`	Integer	Não	500	Quantidade de registros gravados em um único lote.
`flushIntervalMs`	Integer	Não	N/A	Intervalo de liberação, em milissegundos. Se a quantidade de registros no buffer não atingir `batchSize` dentro desse intervalo, o conector liberará todos os dados armazenados no buffer.
`writeMode`	String	Não	insert	Modo de gravação. Consulte Modos de gravação para obter detalhes.
`conflictMode`	String	Não	strict	Política de tratamento de conflitos para operações de `insert`. Consulte Modos de gravação para obter detalhes.

Modos de gravação

O conector oferece dois modos de gravação, controlados por writeMode e conflictMode.

writeMode=insert (padrão)

Neste modo, o conector executa uma instrução INSERT direta. Quando ocorre um conflito de chave primária ou índice exclusivo, o comportamento é determinado por conflictMode:

`conflictMode`	Comportamento
`strict` (padrão)	Gera um erro em caso de conflito.
`ignore`	Ignora o registro conflitante.
`update`	Atualize o registro existente quando há conflito. Funciona em tabelas sem chave primária, mas apresenta menor desempenho de gravação.

writeMode=upsert

Neste modo, o conector realiza um INSERT ou UPDATE dependendo da existência de conflitos. Esse modo exige que a tabela sink tenha uma chave primária definida na DDL.

Métricas

As métricas abaixo estão disponíveis para tabelas sink. Para descrições detalhadas de cada métrica, consulte Métricas de monitoramento.

numRecordsOut
numRecordsOutPerSecond
numBytesOut
numBytesOutPerSecond
currentSendTime

Mapeamentos de tipos de dados

A tabela a seguir mostra o mapeamento entre os tipos de dados do PolarDB for PostgreSQL (compatível com Oracle) 1.0 e os tipos de dados do Flink SQL para tabelas sink.

Tipo do PolarDB	Tipo do Flink
BOOLEAN	BOOLEAN
INT	INT
NUMBER	BIGINT
NUMBER	DOUBLE
VARCHAR	VARCHAR
TIMESTAMP	TIMESTAMP
VARCHAR	DATE

Exemplo

O exemplo abaixo gera dados usando uma source datagen e os grava em uma tabela sink do PolarDB.

CREATE TEMPORARY TABLE datagen_source (
  `name` VARCHAR,
  `age` INT
)
COMMENT 'datagen source table'
WITH (
  'connector' = 'datagen'
);

CREATE TABLE polardbo_sink (
  name VARCHAR,
  age INT
) WITH (
  'connector'='polardbo',
  'url'='jdbc:postgresql://<Address>:<PortId>/<DatabaseName>',
  'tableName'='<yourDatabaseTableName>',
  'userName'='<yourDatabaseUserName>',
  'password'='<yourDatabasePassword>'
);

INSERT INTO polardbo_sink
SELECT * FROM datagen_source;