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

Importante

O conector do PolarDB for PostgreSQL (Compatible with 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 mais detalhes, consulte {{XREF_0}}. Migre suas cargas de trabalho o mais rápido possível 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 do {{XREF_1}} 1.0. O PolarDB for PostgreSQL (Compatible with Oracle) é um banco de dados nativo da cloud, desenvolvido com uma arquitetura de separação de armazenamento e computação, com alta compatibilidade com o Oracle.

Pré-requisitos

Antes de começar, certifique-se de ter:

Um cluster do PolarDB for PostgreSQL (Compatible with Oracle) 1.0 e uma tabela. Consulte Criar um cluster do PolarDB for PostgreSQL (Compatible with Oracle) e Criar uma tabela
Uma lista de permissões configurada para o cluster. Consulte {{XREF_2}}

Limitações

Este conector oferece suporte apenas ao PolarDB 1.0. Para o PolarDB 2.0, use o conector JDBC.
Este conector requer o Ververica Runtime (VVR) 8.0.5 ou posterior.

Sintaxe

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;

Opções do conector

Opção	Tipo	Obrigatório	Padrão	Descrição
`connector`	String	Sim	N/D	Deve ser `polardbo`.
`url`	String	Sim	N/D	A URL JDBC do banco de dados. Formato: `jdbc:postgresql://<Address>:<PortId>/<DatabaseName>`.
`tableName`	String	Sim	N/D	O nome da tabela do banco de dados.
`userName`	String	Sim	N/D	O nome de usuário para conexão com o banco de dados.
`password`	String	Sim	N/D	A senha para conexão com o banco de dados. Armazene as credenciais usando {{XREF_3}} em vez de codificá-las em texto simples.
`maxRetryTimes`	Integer	Não	3	O número máximo de tentativas para uma operação de gravação com falha.
`targetSchema`	String	Não	public	O nome do schema.
`caseSensitive`	String	Não	false	Especifica se os nomes de tabelas e campos diferenciam maiúsculas de minúsculas. Valores válidos: `true`, `false`.
`connectionMaxActive`	Integer	Não	5	O número máximo 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	O intervalo entre as tentativas, em milissegundos.
`batchSize`	Integer	Não	500	O número de registros gravados em um único lote.
`flushIntervalMs`	Integer	Não	N/D	O intervalo de liberação, em milissegundos. Se o número de registros armazenados em buffer não atingir `batchSize` dentro desse intervalo, o conector libera todos os dados em buffer.
`writeMode`	String	Não	insert	O modo de gravação. Consulte Modos de gravação para obter detalhes.
`conflictMode`	String	Não	strict	A 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 suporte a dois modos de gravação, controlados por writeMode e conflictMode.

writeMode=insert (padrão)

O conector executa um INSERT direto. Quando ocorre um conflito de chave primária ou índice único, o comportamento é controlado por conflictMode:

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

writeMode=upsert

O conector executa um INSERT ou UPDATE com base na existência de um conflito. Este modo exige que a tabela sink tenha uma chave primária definida na DDL.

Métricas

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

numRecordsOut
numRecordsOutPerSecond
numBytesOut
numBytesOutPerSecond
currentSendTime

Mapeamentos de tipos de dados

A tabela a seguir mapeia os tipos de dados do PolarDB for PostgreSQL (Compatible with Oracle) 1.0 para os tipos de dados do Flink SQL em 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 a seguir gera dados usando uma datagen source 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;