PolarDB for PostgreSQL connector - Realtime Compute for Apache Flink

Importante

El conector de PolarDB for PostgreSQL (Compatible with Oracle) 1.0 está en proceso de desuso y se eliminará en una versión futura. Ya no recibirá actualizaciones ni mantenimiento, y sus opciones de configuración se quitarán de la consola. Para más detalles, consulte End-of-support (EOS) for the PolarDB for PostgreSQL (Compatible with Oracle) 1.0 connector. Migre sus cargas de trabajo a la brevedad para evitar interrupciones en el servicio.

Tipo compatible: Sink | Modos de ejecución: Streaming y por lotes | Tipo de API: SQL | Compatibilidad con actualización y eliminación: Compatible

Utilice este conector para escribir datos de Flink en un clúster de PolarDB for PostgreSQL (Compatible with Oracle) 1.0. PolarDB for PostgreSQL (Compatible with Oracle) es una base de datos nativa de la nube basada en una arquitectura de separación de almacenamiento y cómputo, con alta compatibilidad con Oracle.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un clúster de PolarDB for PostgreSQL (Compatible with Oracle) 1.0 y una tabla. Consulte Crear un clúster de PolarDB for PostgreSQL (Compatible with Oracle) y Crear una tabla
Una lista de permitidos configurada para el clúster. Consulte Configure a cluster whitelist

Limitaciones

Este conector solo es compatible con PolarDB 1.0. Para PolarDB 2.0, utilice el conector JDBC en su lugar.
Este conector requiere Ververica Runtime (VVR) 8.0.5 o posterior.

Sintaxis

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

Opciones del conector

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`connector`	String	Sí	N/A	Debe ser `polardbo`.
`url`	String	Sí	N/A	La URL JDBC de la base de datos. Formato: `jdbc:postgresql://<Address>:<PortId>/<DatabaseName>`.
`tableName`	String	Sí	N/A	El nombre de la tabla de la base de datos.
`userName`	String	Sí	N/A	El nombre de usuario para conectarse a la base de datos.
`password`	String	Sí	N/A	La contraseña para conectarse a la base de datos. Almacene las credenciales mediante project variables en lugar de codificarlas en texto plano.
`maxRetryTimes`	Integer	No	3	El número máximo de reintentos para una operación de escritura con errores.
`targetSchema`	String	No	public	El nombre del esquema.
`caseSensitive`	String	No	false	Especifica si los nombres de tablas y campos distinguen entre mayúsculas y minúsculas. Valores válidos: `true`, `false`.
`connectionMaxActive`	Integer	No	5	El número máximo de conexiones activas en el pool de conexiones. El sistema libera automáticamente las conexiones inactivas. Establecer este valor demasiado alto puede sobrecargar el servidor de base de datos.
`retryWaitTime`	Integer	No	100	El intervalo entre reintentos, en milisegundos.
`batchSize`	Integer	No	500	El número de registros escritos en un solo lote.
`flushIntervalMs`	Integer	No	N/A	El intervalo de vaciado, en milisegundos. Si el número de registros en búfer no alcanza `batchSize` dentro de este intervalo, el conector vacía todos los datos almacenados en el búfer.
`writeMode`	String	No	insert	El modo de escritura. Consulte Modos de escritura para más detalles.
`conflictMode`	String	No	strict	La directiva de manejo de conflictos para operaciones `insert`. Consulte Modos de escritura para más detalles.

Modos de escritura

El conector admite dos modos de escritura, controlados por writeMode y conflictMode.

writeMode=insert (predeterminado)

El conector realiza una operación INSERT directa. Cuando ocurre un conflicto de clave primaria o índice único, el comportamiento se controla mediante conflictMode:

`conflictMode`	Comportamiento
`strict` (predeterminado)	Genera un error en caso de conflicto.
`ignore`	Ignora el registro en conflicto.
`update`	Actualiza el registro existente en caso de conflicto. Funciona en tablas sin clave primaria, pero tiene un menor rendimiento de escritura.

writeMode=upsert

El conector realiza una operación INSERT o UPDATE según exista o no un conflicto. Este modo requiere que la tabla sink tenga una clave primaria definida en el DDL.

Métricas

Las siguientes métricas están disponibles para las tablas sink. Para consultar la descripción de cada métrica, véase Métricas de monitoreo.

numRecordsOut
numRecordsOutPerSecond
numBytesOut
numBytesOutPerSecond
currentSendTime

Mapeo de tipos de datos

La siguiente tabla muestra el mapeo de los tipos de datos de PolarDB for PostgreSQL (Compatible with Oracle) 1.0 a los tipos de datos de Flink SQL para tablas sink.

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

Ejemplo

El siguiente ejemplo genera datos mediante un origen datagen y los escribe en una tabla sink de 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;