Realtime Compute for Apache Flink:AnalyticDB for PostgreSQL connector

Başlangıç modları

Değişiklik günlüğü modları

Yuva adı rehberi

Çakışmaları önlemek için çoğaltma yuvalarının dikkatlice yönetilmesi gerekir:

• Tek bir Flink işi içinde: İş içindeki tüm kaynak tablolar için aynı slot.name değerini kullanın.

• Farklı Flink işleri arasında: Her iş için benzersiz bir slot.name değeri atayın. Farklı işlerde aynı yuva adının yeniden kullanılması PSQLException: ERROR: replication slot "debezium" is active for PID 974 hatasına neden olur.

Kalp atışı ve WAL saklama

Bir kaynak tablodaki güncellemeler seyrek olduğunda, connector çoğaltma yuvası ofsetini ilerletmeyebilir. Veritabanı, yuva ofseti ilgili dosyaların üzerine gelene kadar WAL dosyalarını tuttuğu için bu durum AnalyticDB for PostgreSQL'in WAL disk alanını geri kazanmasını engeller. Periyodik kalp atışı göndermek, yuva ofsetinin ileriye doğru hareket etmesini sağlayarak eski WAL dosyalarının serbest bırakılmasına olanak tanır. Tablo değişikliklerinin seyrek olduğu durumlarda heartbeat.interval.ms değerini güncelleme sıklığınıza uygun olacak şekilde ayarlayın.

Yazma modları ve çakışma çözümü

writeMode ve conflictMode parametreleri, kayıtların nasıl yazılacağını ve çakışmaların nasıl ele alınacağını kontrol etmek için birlikte çalışır.

Önbellek ilkeleri

Doğru önbellek ilkesini seçmek, sorgu aktarım hızı ile veri güncelliği arasında denge kurmanızı sağlar.

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;