Connecteur PolarDB for PostgreSQL - Realtime Compute for Apache Flink

Important

Le connecteur PolarDB for PostgreSQL (Compatible with Oracle) 1,0 est en cours de dépréciation et sera supprimé dans une prochaine version. Il ne recevra plus de mises à jour ni de maintenance, et ses options de configuration seront retirées de la console. Pour plus de détails, consultez la rubrique Fin du support (EOS) pour le connecteur PolarDB for PostgreSQL (Compatible with Oracle) 1,0. Migrez vos charges de travail rapidement afin d'éviter toute interruption de service.

Type pris en charge : Sink | Modes d'exécution : Streaming et batch | Type d'API : SQL | Prise en charge des mises à jour et suppressions : Oui

Utilisez ce connecteur pour écrire les données Flink dans un cluster PolarDB for PostgreSQL (Compatible with Oracle) 1,0. PolarDB for PostgreSQL (Compatible with Oracle) est une base de données cloud-native reposant sur une architecture de séparation du stockage et du calcul, offrant une haute compatibilité avec Oracle.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un cluster PolarDB for PostgreSQL (Compatible with Oracle) 1,0 et une table. Consultez les rubriques Créer un cluster PolarDB for PostgreSQL (Compatible with Oracle) et Créer une table
Une liste d'autorisation configurée pour le cluster. Consultez la rubrique Configurer la liste d'autorisation d'un cluster

Limitations

Ce connecteur prend uniquement en charge PolarDB 1.0. Pour PolarDB 2.0, utilisez plutôt le connecteur JDBC.
Ce connecteur nécessite Ververica Runtime (VVR) 8.0.5 ou une version ultérieure.

Syntaxe

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

Options du connecteur

Option	Type	Obligatoire	Par défaut	Description
`connector`	String	Oui	N/A	Doit être `polardbo`.
`url`	String	Oui	N/A	L'URL JDBC de la base de données. Format : `jdbc:postgresql://<Address>:<PortId>/<DatabaseName>`.
`tableName`	String	Oui	N/A	Le nom de la table de la base de données.
`userName`	String	Oui	N/A	Le nom d'utilisateur pour la connexion à la base de données.
`password`	String	Oui	N/A	Le mot de passe pour la connexion à la base de données. Stockez les identifiants à l'aide de variables de projet plutôt que de les coder en dur en texte clair.
`maxRetryTimes`	Integer	Non	3	Le nombre maximal de tentatives pour une opération d'écriture ayant échoué.
`targetSchema`	String	Non	public	Le nom du schéma.
`caseSensitive`	String	Non	false	Indique si les noms de tables et de champs sont sensibles à la casse. Valeurs valides : `true`, `false`.
`connectionMaxActive`	Integer	Non	5	Le nombre maximal de connexions actives dans le pool de connexions. Le système libère automatiquement les connexions inactives. Une valeur trop élevée risque de surcharger le serveur de base de données.
`retryWaitTime`	Integer	Non	100	L'intervalle entre les tentatives, en millisecondes.
`batchSize`	Integer	Non	500	Le nombre d'enregistrements écrits en un seul lot.
`flushIntervalMs`	Integer	Non	N/A	L'intervalle de vidage, en millisecondes. Si le nombre d'enregistrements mis en mémoire tampon n'atteint pas `batchSize` durant cet intervalle, le connecteur vide toutes les données tamponnées.
`writeMode`	String	Non	insert	Le mode d'écriture. Consultez la section Modes d'écriture pour plus de détails.
`conflictMode`	String	Non	strict	La politique de gestion des conflits pour les opérations `insert`. Consultez la section Modes d'écriture pour plus de détails.

Modes d'écriture

Le connecteur prend en charge deux modes d'écriture, contrôlés par writeMode et conflictMode.

writeMode=insert (par défaut)

Le connecteur effectue une insertion directe (INSERT). En cas de conflit de clé primaire ou d'index unique, le comportement est déterminé par conflictMode :

`conflictMode`	Comportement
`strict` (par défaut)	Génère une erreur en cas de conflit.
`ignore`	Ignore l'enregistrement conflictuel.
`update`	Met à jour l'enregistrement existant en cas de conflit. Fonctionne sur les tables sans clé primaire, mais offre des performances d'écriture inférieures.

writeMode=upsert

Le connecteur effectue une insertion (INSERT) ou une mise à jour (UPDATE) selon qu'un conflit existe ou non. Ce mode requiert que la table sink dispose d'une clé primaire définie dans le DDL.

Métriques

Les métriques suivantes sont disponibles pour les tables sink. Pour obtenir la description de chaque métrique, consultez la rubrique Métriques de surveillance.

numRecordsOut
numRecordsOutPerSecond
numBytesOut
numBytesOutPerSecond
currentSendTime

Mappages de types de données

Le tableau suivant établit la correspondance entre les types de données PolarDB for PostgreSQL (Compatible with Oracle) 1,0 et les types de données Flink SQL pour les tables sink.

Type PolarDB	Type Flink
BOOLEAN	BOOLEAN
INT	INT
NUMBER	BIGINT
NUMBER	DOUBLE
VARCHAR	VARCHAR
TIMESTAMP	TIMESTAMP
VARCHAR	DATE

Exemple

L'exemple suivant génère des données à l'aide d'une source datagen et les écrit dans une table sink 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;