Manage PostgreSQL catalogs - Realtime Compute for Apache Flink

Un catalogue PostgreSQL gère les métadonnées PostgreSQL et permet un accès direct en lecture et en écriture aux données dans Flink sans reconstruire les schémas de table. Cette rubrique explique comment créer, utiliser et supprimer un catalogue PostgreSQL.

Prérequis de version

Les catalogues PostgreSQL nécessitent Ververica Runtime (VVR) 11.4 ou une version ultérieure.

Créer un catalogue

Sur la page Scripts, exécutez l'instruction suivante dans l'éditeur SQL pour créer un catalogue :

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

Option de configuration

Obligatoire

Valeur par défaut

Description

type

Oui

(aucune)

Type du catalogue. Définissez cette option sur postgres.

hostname

Oui

(aucune)

Nom d'hôte de la base de données PostgreSQL.

port

Non

5432

Numéro de port de la base de données.

username

Oui

(aucune)

Nom d'utilisateur pour accéder à la base de données.

password

Oui

(aucune)

Mot de passe pour accéder à la base de données.

default-database

Oui

(aucune)

Nom de la base de données par défaut à laquelle se connecter.

Consulter un catalogue

Une fois le catalogue créé, exécutez les commandes suivantes pour consulter ses bases de données et ses tables.

USE CATALOG `postgres`;
SHOW DATABASES;

USE `postgres`;
SHOW TABLES;

Utiliser un catalogue

Lire des données depuis PostgreSQL

Avec un catalogue PostgreSQL, vous pouvez lire directement les données des tables PostgreSQL via Flink. Pour configurer les options Change Data Capture (CDC), telles que le slot de réplication, utilisez des indications SQL (OPTIONS) afin de remplacer la configuration.

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

Écrire des données dans PostgreSQL

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

Jointure de recherche (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;

Supprimer un catalogue

Si vous n'avez plus besoin du catalogue PostgreSQL, exécutez la commande suivante pour le supprimer. Cette opération supprime uniquement le mappage des métadonnées dans Flink.

DROP CATALOG `postgres`;