Gestión de catálogos de PostgreSQL - Realtime Compute for Apache Flink

Un catálogo de PostgreSQL gestiona los metadatos de PostgreSQL y permite el acceso directo de lectura y escritura de datos en Flink sin necesidad de reconstruir los esquemas de las tablas. Este tema describe cómo crear, usar y eliminar un catálogo de PostgreSQL.

Requisitos de versión

Los catálogos de PostgreSQL requieren Ververica Runtime (VVR) 11.4 o superior.

Crear un catálogo

En la página Scripts, ejecute la siguiente sentencia en el editor SQL para crear un catálogo:

CREATE CATALOG `postgres` WITH (
    'type' = 'postgres',
    'default-database' = 'postgres',                      
    'hostname' = '<yourHostname>',
    'port' = '5432',                                       
    'username' = '<yourUserName>',                        
    'password' = '<yourPassWord>'                           
);

Opción de configuración	Obligatorio	Valor predeterminado	Descripción
type	Sí	(Ninguna)	El tipo de catálogo. Establezca este valor en `postgres`.
hostname	Sí	(Ninguna)	El nombre de host de la base de datos PostgreSQL.
port	No	`5432`	El número de puerto de la base de datos.
username	Sí	(Ninguna)	El nombre de usuario para acceder a la base de datos.
password	Sí	(Ninguna)	La contraseña para acceder a la base de datos.
default-database	Sí	(Ninguna)	El nombre de la base de datos predeterminada a la que se conectará.

Consultar un catálogo

Después de crear un catálogo, ejecute los siguientes comandos para consultar sus bases de datos y tablas.

USE CATALOG `postgres`;
SHOW DATABASES;

USE `postgres`;
SHOW TABLES;

Uso de un catálogo

Lectura de datos desde PostreSQL

Con un catálogo de PostreSQL, puede leer datos directamente de las tablas de PostgreSQL mediante Flink. Para configurar opciones de Change Data Capture (CDC), como el replication slot, use SQL hints (OPTIONS) para sobrescribir la configuración.

SELECT * 
FROM `postgres`.`postgres`.`public.target_table` 
/*+ OPTIONS(
    'slot.name' = 'testName', 
    'debezium.publication.autocreate.mode' = 'filtered'
) */;

Escritura de datos en PostreSQL

INSERT INTO `postgres`.`postgres`.`public.target_table`
SELECT id, name 
FROM `source_table`;

Lookup join

INSERT INTO sink_table
SELECT 
  o.order_id,
  o.user_id,
  d.user_name,  
  o.amount
FROM pg_catalog.db.orders AS o
JOIN mysql_dim.db.users FOR SYSTEM_TIME AS OF o.proc_time AS d
ON o.user_id = d.user_id;

Eliminar un catálogo

Si ya no necesita el catálogo de PostreSQL, ejecute el siguiente comando para eliminarlo. Esta operación solo elimina la asignación de metadatos en Flink.

DROP CATALOG `postgres`;