Realtime Compute for Apache Flink:AnalyticDB for PostgreSQL connector

Modes de démarrage

Modes de journal des modifications

Conseils sur le nom du slot

Gérez les slots de réplication avec soin pour éviter les conflits :

• Au sein d'un seul job Flink : Utilisez le même slot.name pour toutes les tables sources du job.

• Entre différents jobs Flink : Attribuez un slot.name unique par job. La réutilisation d'un nom de slot entre plusieurs jobs provoque l'erreur PSQLException: ERROR: replication slot "debezium" is active for PID 974.

Heartbeat et rétention WAL

Lorsqu'une table source reçoit peu de mises à jour, le connecteur peut ne pas faire avancer l'offset du slot de réplication. Cela empêche AnalyticDB for PostgreSQL de récupérer l'espace disque WAL, car la base de données conserve les fichiers WAL jusqu'à ce que l'offset du slot les dépasse. L'envoi périodique de heartbeats permet de faire avancer l'offset du slot, libérant ainsi les anciens fichiers WAL. Définissez heartbeat.interval.ms sur une valeur adaptée à votre fréquence de mise à jour lorsque les modifications de table sont rares.

Modes d'écriture et résolution des conflits

Les options writeMode et conflictMode fonctionnent conjointement pour contrôler l'écriture des enregistrements et la gestion des conflits.

Politiques de cache

Le choix de la politique de cache appropriée permet d'équilibrer le débit des requêtes et l'actualité des données.

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;

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;