AnalyticDB for PostgreSQL connector - Realtime Compute for Apache Flink

El conector de AnalyticDB for PostgreSQL permite utilizar AnalyticDB for PostgreSQL como tabla de origen (beta), de dimensiones o sink en trabajos SQL de Realtime Compute for Apache Flink. AnalyticDB for PostgreSQL es un almacén de datos con procesamiento masivamente paralelo (MPP) diseñado para análisis en línea a gran escala.

Compatible con: Origen (beta) · Dimensiones · Sink | Streaming · Lotes | API SQL | El sink admite actualizaciones y eliminaciones

La lectura desde una tabla de origen requiere un conector personalizado configurado mediante Flink CDC. Para consultar las instrucciones de configuración, consulte Uso de Flink CDC para suscribirse a datos completos e incrementales en tiempo real .

Requisitos previos

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

Una instancia y una tabla de AnalyticDB for PostgreSQL. Consulte Creación de una instancia y CREATE TABLE
Una lista de permitidos de direcciones IP configurada para la instancia. Consulte Configure an IP address whitelist

Limitaciones

AnalyticDB for PostgreSQL V7.0 requiere VVR 8.0.1 o posterior.
No se admiten bases de datos PostgreSQL autogestionadas.

Sintaxis

Utilice connector='adbpg' para tablas de dimensiones y sink, y connector='adbpg-cdc' para tablas de origen.

CREATE TEMPORARY TABLE adbpg_table (
  id      INT,
  len     INT,
  content VARCHAR,
  PRIMARY KEY (id)
) WITH (
  'connector'  = 'adbpg',
  'url'        = 'jdbc:postgresql://<host>:<port>/<database>',
  'tableName'  = '<table>',
  'userName'   = '<username>',
  'password'   = '<password>'
);

Opciones del conector

Generales

Estas opciones aplican a todos los tipos de tabla.

Opción	Obligatorio	Predeterminado	Tipo	Descripción
`connector`	Sí	—	STRING	Establezca este valor en `adbpg-cdc` para tablas de origen, o en `adbpg` para tablas de dimensiones y sink.
`url`	Sí	—	STRING	La URL de Java Database Connectivity (JDBC) con el formato `jdbc:postgresql://<host>:<port>/<database>`.
`tableName`	Sí	—	STRING	El nombre de la tabla en la base de datos.
`userName`	Sí	—	STRING	El nombre de usuario de la base de datos de AnalyticDB for PostgreSQL.
`password`	Sí	—	STRING	La contraseña de la base de datos de AnalyticDB for PostgreSQL.
`maxRetryTimes`	No	`3`	INTEGER	El número máximo de intentos de reintento cuando falla una escritura.
`targetSchema`	No	`public`	STRING	El nombre del esquema de la base de datos.
`caseSensitive`	No	`false`	STRING	Especifica si los identificadores distinguen entre mayúsculas y minúsculas. Valores válidos: `true`, `false`.
`connectionMaxActive`	No	`5`	INTEGER	El número máximo de conexiones en el pool de conexiones. El conector libera automáticamente las conexiones inactivas. Un valor demasiado alto puede provocar recuentos anómalos de conexiones en el servidor.

Específicas del origen (beta)

Estas opciones aplican únicamente a las tablas de origen (connector='adbpg-cdc').

Opción	Obligatorio	Predeterminado	Tipo	Descripción
`schema-name`	Sí	—	STRING	El nombre del esquema. Admite expresiones regulares para suscribirse a varios esquemas simultáneamente.
`port`	Sí	`5432`	INTEGER	El puerto de la instancia de AnalyticDB for PostgreSQL.
`decoding.plugin.name`	Sí	`pgoutput`	STRING	El plugin de decodificación lógica de PostgreSQL. Establezca este valor en `pgoutput`.
`slot.name`	Sí	—	STRING	El nombre del slot de decodificación lógica. Consulte Guía sobre nombres de slots.
`debezium.*`	Sí	—	STRING	Propiedades de configuración del cliente Debezium. Por ejemplo, establezca `'debezium.snapshot.mode' = 'never'` para omitir la instantánea inicial. Consulte Propiedades del conector.
`scan.incremental.snapshot.enabled`	No	`false`	BOOLEAN	Especifica si se habilita la instantánea incremental. Valores válidos: `true`, `false`.
`scan.startup.mode`	No	`initial`	STRING	El modo de inicio para el consumo de datos. Consulte Modos de inicio.
`changelog-mode`	No	`ALL`	STRING	Define cómo se codifican los eventos de cambio en el flujo de cambios. Valores válidos: `ALL`, `UPSERT`.
`heartbeat.interval.ms`	No	30 segundos	DURATION	El intervalo para enviar paquetes de latido, en milisegundos. Consulte Latidos y retención de WAL.
`scan.incremental.snapshot.chunk.key-column`	No	Primera columna de clave primaria	STRING	La columna que se utiliza como clave de fragmento durante las lecturas de instantáneas.

Modos de inicio

Valor	Comportamiento
`initial` (predeterminado)	Realiza un escaneo completo de los datos históricos en el primer inicio y luego lee desde la posición más reciente del registro de escritura anticipada (WAL).
`latest-offset`	Omite el escaneo de datos históricos y lee solo desde el extremo actual del WAL.
`snapshot`	Ejecuta un escaneo completo de los datos históricos y captura los datos del WAL generados durante dicho proceso. Se detiene al finalizar el escaneo.

Modos de registro de cambios

Valor	Eventos capturados
`ALL` (predeterminado)	`INSERT`, `DELETE`, `UPDATE_BEFORE`, `UPDATE_AFTER`
`UPSERT`	`INSERT`, `DELETE`, `UPDATE_AFTER`

Guía sobre nombres de slots

Los slots de replicación deben gestionarse con cuidado para evitar conflictos:

Dentro de un mismo trabajo de Flink: Utilice el mismo slot.name para todas las tablas de origen del trabajo.
Entre distintos trabajos de Flink: Asigne un slot.name único por trabajo. Reutilizar un nombre de slot entre trabajos provoca el error PSQLException: ERROR: replication slot "debezium" is active for PID 974.

Latidos y retención de WAL

Cuando una tabla de origen tiene actualizaciones poco frecuentes, es posible que el conector no avance el desplazamiento del slot de replicación. Esto impide que AnalyticDB for PostgreSQL recupere espacio en disco del WAL, ya que la base de datos conserva los archivos WAL hasta que el desplazamiento del slot avanza más allá de ellos. El envío periódico de latidos mantiene el avance del desplazamiento del slot y permite liberar los archivos WAL antiguos. Si los cambios en la tabla son esporádicos, configure heartbeat.interval.ms con un valor adecuado a su frecuencia de actualización.

Específicas del sink

Estas opciones aplican únicamente a las tablas sink (connector='adbpg').

Opción	Obligatorio	Predeterminado	Tipo	Descripción
`writeMode`	No	`insert`	STRING	La estrategia de escritura para el primer intento. Consulte Modos de escritura y resolución de conflictos.
`conflictMode`	No	`strict`	STRING	Define cómo manejar los conflictos de clave primaria o índice. Consulte Modos de escritura y resolución de conflictos.
`batchSize`	No	`500`	INTEGER	El número de registros escritos por lote.
`flushIntervalMs`	No	—	INTEGER	El tiempo máximo que espera el conector antes de volcar los registros en caché, en milisegundos. Si el búfer no alcanza el valor de `batchSize` dentro de este periodo, todos los registros almacenados se escriben de inmediato.
`retryWaitTime`	No	`100`	INTEGER	El intervalo entre intentos de reintento, en milisegundos.

Modos de escritura y resolución de conflictos

Las opciones writeMode y conflictMode funcionan en conjunto para controlar la escritura de registros y la gestión de conflictos.

`writeMode`	Descripción	Notas
`insert` (predeterminado)	Inserta los registros directamente. La gestión de conflictos depende de `conflictMode`.	Adecuado para la mayoría de los casos de uso.
`upsert`	Actualiza automáticamente los registros existentes cuando hay un conflicto.	Requiere una clave primaria.
`copy`	Inserta registros mediante el comando `COPY` de PostgreSQL.	Requiere VVR 11.1 o posterior.

`conflictMode`	Comportamiento ante conflictos	Notas
`strict` (predeterminado)	Reporta un error.
`ignore`	Descarta silenciosamente el registro en conflicto.
`update`	Actualiza el registro en conflicto.	Útil para tablas sin clave primaria. Reduce el throughput de escritura.
`upsert`	Actualiza el registro en conflicto.	Requiere una clave primaria. Es más eficiente que `update` en tablas con clave.

Específicas de tablas de dimensiones

Estas opciones aplican únicamente a las tablas de dimensiones (connector='adbpg' utilizado en un JOIN con FOR SYSTEM_TIME AS OF).

Opción	Obligatorio	Predeterminado	Tipo	Descripción
`cache`	No	`ALL`	STRING	La directiva de caché para las búsquedas en la tabla de dimensiones. Consulte Directivas de caché.
`cacheSize`	No	`100000`	LONG	El número máximo de filas mantenidas en la caché. Solo surte efecto cuando `cache=LRU`.
`cacheTTLMs`	No	`Long.MAX_VALUE`	LONG	El tiempo de vida de las entradas de caché en milisegundos. Su comportamiento depende de la configuración de `cache`. Consulte Directivas de caché.
`maxJoinRows`	No	`1024`	INTEGER	El número máximo de filas para unir por cada registro de entrada.

Directivas de caché

Elegir la directiva de caché adecuada permite equilibrar el throughput de consultas con la actualidad de los datos.

Valor de `cache`	Comportamiento	Cuándo utilizarla
`ALL` (predeterminado)	Carga toda la tabla de dimensiones en memoria antes de iniciar el trabajo. Todas las búsquedas se resuelven desde la caché. Al expirar la caché (`cacheTTLMs`), se recarga la tabla completa. Si una clave de unión no se encuentra en la caché, significa que no existe.	Tablas de dimensiones pequeñas o medianas donde la actualización periódica de datos es aceptable.
`LRU`	Almacena en caché un subconjunto de filas. Ante un fallo de caché, obtiene los datos de la base de datos y actualiza la caché. Expulsa las entradas menos utilizadas recientemente cuando la caché alcanza el límite de `cacheSize`. El parámetro `cacheTTLMs` controla la expiración individual de cada entrada.	Tablas de dimensiones grandes donde almacenar todo en caché no es viable, o cuando resulta aceptable una caché parcial de baja latencia.
`None`	Sin caché. Cada búsqueda consulta directamente la base de datos.	Cuando los datos siempre deben leerse actualizados desde la base de datos.

Mapeo de tipos de datos

Tipo de AnalyticDB for PostgreSQL	Tipo de Flink SQL
`BOOLEAN`	`BOOLEAN`
`SMALLINT`	`INT`
`INT`	`INT`
`BIGINT`	`BIGINT`
`FLOAT`	`DOUBLE`
`VARCHAR`	`VARCHAR`
`TEXT`	`VARCHAR`
`TIMESTAMP`	`TIMESTAMP`
`DATE`	`DATE`

Ejemplos

Tabla sink

Este ejemplo lee desde un origen datagen y escribe en una tabla sink de AnalyticDB for PostgreSQL.

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

CREATE TEMPORARY TABLE adbpg_sink (
  name VARCHAR,
  age  INT
) WITH (
  'connector' = 'adbpg',
  'url'       = 'jdbc:postgresql://<host>:<port>/<database>',
  'tableName' = '<table>',
  'userName'  = '<username>',
  'password'  = '<password>'
);

INSERT INTO adbpg_sink
SELECT * FROM datagen_source;

Tabla de dimensiones

Este ejemplo une un flujo datagen con una tabla de dimensiones de AnalyticDB for PostgreSQL mediante una unión temporal.

CREATE TEMPORARY TABLE datagen_source (
  a         INT,
  b         BIGINT,
  c         STRING,
  `proctime` AS PROCTIME()
)
COMMENT 'datagen source table'
WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE adbpg_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'adbpg',
  'url'       = 'jdbc:postgresql://<host>:<port>/<database>',
  'tableName' = '<table>',
  'userName'  = '<username>',
  'password'  = '<password>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
)
COMMENT 'blackhole sink table'
WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN adbpg_dim FOR SYSTEM_TIME AS OF T.proctime AS H
  ON T.a = H.a;

Tabla de origen (beta)

Para consultar la configuración y los ejemplos de tablas de origen, consulte Uso de Flink CDC para suscribirse a datos completos e incrementales en tiempo real.

Métricas

Las métricas de las tablas sink están disponibles en la consola de Realtime Compute for Apache Flink. Las tablas de dimensiones no exponen métricas. Para consultar las definiciones de las métricas, consulte Métricas.

Métrica	Descripción
`numRecordsOut`	Número total de registros escritos en el sink.
`numRecordsOutPerSecond`	Registros escritos por segundo.
`numBytesOut`	Total de bytes escritos en el sink.
`numBytesOutPerSecond`	Bytes escritos por segundo.
`currentSendTime`	Tiempo empleado en la operación de escritura más reciente.

Realtime Compute for Apache Flink:AnalyticDB for PostgreSQL connector