All Products
Search
Document Center

Realtime Compute for Apache Flink:PostgreSQL CDC

Last Updated:Sep 25, 2026

Konektor Postgres CDC membaca snapshot lengkap dari database PostgreSQL, lalu menangkap data perubahan dengan semantik pemrosesan tepat-sekali (exactly-once).

Ikhtisar

Konektor Postgres CDC mendukung kemampuan berikut:

Kategori

Detail

Jenis yang didukung

Sumber SQL, sumber Flink CDC

Catatan

Gunakan konektor JDBC untuk tabel sink dan tabel lookup (dimensi).

Mode eksekusi

Streaming

Format data

Tidak berlaku

Metrik

Metrik pemantauan

  • currentFetchEventTimeLag: Interval antara saat data dihasilkan hingga ditarik oleh operator sumber.

  • currentEmitEventTimeLag: Interval antara saat data dihasilkan hingga meninggalkan operator sumber.

  • sourceIdleTime: Durasi sumber tidak menghasilkan data baru.

Catatan
  • Metrik currentFetchEventTimeLag dan currentEmitEventTimeLag hanya berlaku pada fase inkremental. Pada fase snapshot, nilainya selalu 0.

  • Untuk informasi lebih lanjut tentang metrik, lihat Deskripsi metrik.

Jenis API

SQL dan Flink CDC

Pembaruan/penghapusan sink

Tidak berlaku

Fitur

Mulai dari VVR 8.0.6, konektor Postgres CDC terintegrasi dengan kerangka kerja snapshot inkremental. Konektor ini membaca data historis lengkap, lalu secara otomatis beralih ke pembacaan log perubahan dari WAL dengan semantik tepat-sekali.

Fitur utama:

  • Pemrosesan aliran dan batch terpadu: membaca data lengkap dan inkremental dalam satu pekerjaan.

  • Pembacaan snapshot konkuren: dapat diskalakan horizontal untuk kinerja lebih cepat.

  • Peralihan mulus dari lengkap ke inkremental: secara otomatis mengurangi penggunaan sumber daya.

  • Pembacaan dapat dilanjutkan: melanjutkan dari titik henti selama fase snapshot untuk stabilitas yang lebih baik.

  • Pembacaan tanpa lock: tidak memerlukan lock, sehingga tidak mengganggu operasi online.

Prasyarat

Konektor Postgres CDC membaca aliran CDC melalui replikasi logis PostgreSQL. Konektor ini mendukung ApsaraDB RDS for PostgreSQL, Amazon RDS for PostgreSQL, dan PostgreSQL yang dikelola sendiri.

Penting

Konfigurasi berbeda tergantung jenis penerapan. Konfigurasi Postgres.

Setelah dikonfigurasi, verifikasi hal berikut:

  • wal_level diatur ke logical untuk mengaktifkan logical decoding.

  • REPLICA IDENTITY setiap tabel yang dilangganan diatur ke FULL, sehingga event INSERT dan UPDATE menyertakan nilai kolom sebelumnya demi konsistensi data.

    Catatan

    REPLICA IDENTITY adalah pengaturan tingkat tabel PostgreSQL yang mengontrol apakah event INSERT dan UPDATE menyertakan nilai kolom sebelumnya. REPLICA IDENTITY.

  • Nilai max_wal_senders dan max_replication_slots melebihi jumlah slot yang sedang digunakan ditambah slot yang dibutuhkan oleh pekerjaan Flink.

  • Akun memiliki hak istimewa SUPERUSER, atau memiliki izin LOGIN dan REPLICATION, serta izin SELECT pada tabel yang dilanggankan.

  • Jika tabel Postgres Anda berisi kolom yang di-generate, atur parameter publish_generated_columns ke stored saat membuat slot. Jika tidak, skema pada fase snapshot dan inkremental mungkin berbeda.

Catatan penggunaan

Fitur snapshot inkremental memerlukan VVR 8.0.6 atau versi yang lebih baru.

Slot replikasi

Pekerjaan Flink PostgreSQL CDC menggunakan slot replikasi untuk mencegah pembersihan WAL yang prematur dan memastikan konsistensi data. Slot yang dikelola dengan buruk dapat menyebabkan penggunaan disk berlebihan atau penundaan pembacaan. Praktik terbaik:

  • Bersihkan slot yang tidak digunakan segera

    • Flink tidak secara otomatis menghapus slot replikasi setelah pekerjaan berhenti atau restart secara stateless, untuk mencegah kehilangan data WAL.

    • Jika pekerjaan tidak akan dijalankan ulang, hapus slot replikasinya secara manual untuk membebaskan ruang disk.

      Catatan

      Manajemen siklus hidup: Perlakukan slot replikasi sebagai resource tingkat pekerjaan dan kelolanya bersamaan dengan start dan stop pekerjaan.

  • Hindari penggunaan ulang slot lama

    • Selalu gunakan nama slot baru. Penggunaan ulang slot lama memaksa pekerjaan membaca data WAL historis yang terakumulasi saat startup, sehingga menunda pemrosesan data baru.

    • PostgreSQL memerlukan satu slot per koneksi. Setiap pekerjaan harus menggunakan nama slot unik.

      Catatan

      Konvensi penamaan: Saat menyesuaikan slot.name, hindari nama dengan akhiran numerik seperti my_slot_1, untuk mencegah konflik dengan slot sementara.

  • Perilaku slot dengan snapshot inkremental diaktifkan

    • Prasyarat: Checkpoint harus diaktifkan, dan tabel sumber harus memiliki primary key yang telah ditentukan.

    • Aturan pembuatan slot:

      • Snapshot inkremental dinonaktifkan: Hanya mendukung parallelism 1. Menggunakan satu slot global.

      • Snapshot inkremental diaktifkan:

        • Fase snapshot: Setiap subtask sumber konkuren membuat slot sementara. Format penamaannya adalah ${slot.name}_${task_id}.

        • Fase inkremental: Semua slot sementara secara otomatis diklaim kembali. Hanya satu slot global yang dipertahankan.

    • Jumlah maksimum slot: Parallelism sumber + 1 (selama fase snapshot)

  • Sumber daya dan kinerja

    • Jika jumlah slot atau ruang disk terbatas, kurangi parallelism snapshot untuk menggunakan lebih sedikit slot sementara. Hal ini mengurangi kecepatan pembacaan snapshot.

    • Jika sink downstream mendukung penulisan idempoten, atur scan.incremental.snapshot.backfill.skip = true untuk melewati backfill WAL selama fase snapshot dan mempercepat startup.

      Pengaturan ini hanya memberikan semantik paling sedikit sekali (at-least-once) dan tidak cocok untuk komputasi stateful (agregasi atau join lookup) karena perubahan historis yang diperlukan mungkin hilang.

  • Saat snapshot inkremental dinonaktifkan, checkpoint tidak didukung selama fase snapshot.

    Konfigurasi untuk menghindari timeout selama fase snapshot

    Saat snapshot inkremental dinonaktifkan, checkpoint selama fase snapshot dapat menyebabkan failover karena timeout. Konfigurasikan parameter berikut di Other Configuration (Konfigurasi parameter running kustom):

    execution.checkpointing.interval: 10min
    execution.checkpointing.tolerable-failed-checkpoints: 100
    restart-strategy: fixed-delay
    restart-strategy.fixed-delay.attempts: 2147483647

    Parameter:

    Parameter

    Deskripsi

    Keterangan

    execution.checkpointing.interval

    Interval antar checkpoint.

    Unitnya adalah nilai durasi, seperti 10min atau 30s.

    execution.checkpointing.tolerable-failed-checkpoints

    Jumlah kegagalan checkpoint yang dapat ditoleransi sebelum pekerjaan gagal.

    Hasil perkalian parameter ini dengan interval penjadwalan checkpoint merupakan waktu pembacaan snapshot yang diizinkan.

    Catatan

    Jika tabel sangat besar, atur parameter ini ke nilai yang lebih besar.

    restart-strategy

    Strategi restart pekerjaan.

    Nilai yang valid:

    • fixed-delay: Strategi restart dengan delay tetap.

    • failure-rate: Strategi restart berdasarkan tingkat kegagalan.

    • exponential-delay: Strategi restart dengan delay eksponensial.

    Restart Strategies.

    restart-strategy.fixed-delay.attempts

    Maksimum upaya restart untuk strategi restart fixed-delay.

    –

Gunakan kembali langganan Postgres

Konektor Postgres CDC bergantung pada publication untuk menentukan perubahan tabel mana yang didorong ke slot. Jika beberapa pekerjaan berbagi publication yang sama, konfigurasi mereka akan saling menimpa.

Penyebab

publication.autocreate.mode default adalah filtered, yang hanya mencakup tabel dalam konfigurasi konektor. Ini memodifikasi publication saat startup pekerjaan, yang dapat memengaruhi pekerjaan lain.

Solusi

  1. Buat publication di PostgreSQL yang mencakup semua tabel yang dipantau, atau buat publication terpisah untuk setiap pekerjaan.

    -- Buat publication bernama my_flink_pub yang mencakup semua tabel (atau tabel tertentu, buat satu publication per pekerjaan)
    CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
    -- Atau lebih sederhana, sertakan semua tabel dalam database
    CREATE PUBLICATION my_flink_pub FOR ALL TABLES;
    Catatan

    Berlangganan ke semua tabel tidak disarankan untuk database besar karena penggunaan bandwidth dan CPU berlebihan pada kluster Flink.

  2. Tambahkan konfigurasi Flink berikut:

    • debezium.publication.name = 'my_flink_pub' (Menentukan nama publication)

    • debezium.publication.autocreate.mode = 'disabled' (Mencegah Flink mencoba membuat atau memodifikasi publication saat startup)

Pengaturan ini memberikan isolasi penuh dan mencegah pekerjaan baru memengaruhi pekerjaan yang sudah ada.

SQL

Sintaksis

CREATE TABLE postgrescdc_source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- Melewatkan backfill dapat mempercepat pembacaan dan mengurangi penggunaan sumber daya, tetapi dapat menyebabkan duplikasi data. Aktifkan ini jika sink downstream bersifat idempoten.
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- Di lingkungan produksi, atur ini ke 'filtered' atau 'disabled' dan kelola publication secara manual, bukan melalui Flink.
  'debezium-publication.autocreate.mode' = 'disabled'
  -- Jika Anda memiliki beberapa sumber, konfigurasikan publication berbeda untuk setiap sumber.
  --'debezium.publication.name' = 'my_flink_pub'
);

Opsi konektor

Opsi

Deskripsi

Tipe data

Wajib

Default

Keterangan

connector

Nama konektor.

STRING

Ya

–

Nilainya harus postgres-cdc.

hostname

Alamat IP atau hostname database PostgreSQL.

STRING

Ya

–

–

username

Username untuk layanan database PostgreSQL.

STRING

Ya

–

–

password

Password untuk layanan database PostgreSQL.

STRING

Ya

–

–

database-name

Nama database PostgreSQL.

STRING

Ya

–

Nama database.

schema-name

Nama skema PostgreSQL. Regex didukung.

STRING

Ya

–

Nama skema mendukung ekspresi reguler untuk membaca data dari beberapa skema.

table-name

Nama tabel PostgreSQL. Regex didukung.

STRING

Ya

–

Nama tabel mendukung ekspresi reguler untuk membaca data dari beberapa tabel.

port

Nomor port.

INTEGER

Tidak

5432

–

decoding.plugin.name

Nama plugin logical decoding PostgreSQL.

STRING

Tidak

decoderbufs

Ini ditentukan oleh plugin yang diinstal pada layanan PostgreSQL. Plugin yang didukung adalah:

  • decoderbufs: Didukung pada PostgreSQL 9.6 dan versi lebih baru. Plugin ini harus diinstal.

  • pgoutput (direkomendasikan): Plugin resmi bawaan untuk PostgreSQL 10 dan versi lebih baru.

slot.name

Nama slot logical decoding.

STRING

Wajib untuk VVR 8.0.1 dan versi lebih baru. Opsional untuk versi sebelumnya.

flink (Sebelum 8.0.1)

Atur slot.name unik untuk setiap tabel untuk menghindari error PSQLException: ERROR: replication slot "debezium" is active for PID 974. Slot replikasi.

Tidak ada nilai default untuk VVR 8.0.1+.

debezium.*

Properti dan parameter Debezium

STRING

Tidak

–

Memberikan kontrol lebih rinci atas perilaku klien Debezium. Misalnya, 'debezium.snapshot.mode' = 'never'. Properti konfigurasi.

scan.incremental.snapshot.enabled

Menentukan apakah snapshot inkremental diaktifkan.

BOOLEAN

Tidak

false

Catatan

scan.startup.mode

Mode startup untuk konsumsi data.

STRING

Tidak

initial

Nilai yang valid:

  • initial: Memindai data historis lengkap saat startup pertama, lalu membaca data WAL terbaru.

  • latest-offset: Tidak memindai data historis lengkap saat startup pertama. Mulai membaca dari akhir WAL, artinya hanya membaca perubahan terbaru setelah konektor dimulai.

  • snapshot: Memindai data historis lengkap, membaca data WAL baru yang dihasilkan selama fase snapshot, lalu pekerjaan berhenti.

changelog-mode

Mode changelog untuk mengenkripsi perubahan aliran.

String

Tidak

all

Mode changelog yang didukung:

  • ALL: Mendukung semua tipe, termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER.

  • UPSERT: Hanya mendukung tipe upsert, yang mencakup INSERT, DELETE, dan UPDATE_AFTER.

heartbeat.interval.ms

Interval pengiriman paket heartbeat.

Duration

Tidak

30s

Unitnya adalah milidetik.

Konektor Postgres CDC secara aktif mengirim heartbeat ke database untuk memajukan offset slot. Saat perubahan tabel jarang terjadi, pengaturan nilai ini memastikan pembersihan log WAL yang tepat waktu.

scan.incremental.snapshot.chunk.key-column

Menentukan kolom yang digunakan sebagai kunci chunk untuk memisahkan shard selama fase snapshot.

STRING

Tidak

–

Secara default, kolom pertama dari primary key dipilih.

scan.incremental.close-idle-reader.enabled

Menentukan apakah reader idle ditutup setelah snapshot selesai.

Boolean

Tidak

false

Untuk mengaktifkan konfigurasi ini, atur execution.checkpointing.checkpoints-after-tasks-finish.enabled ke true.

scan.incremental.snapshot.backfill.skip

Menentukan apakah pembacaan log dilewati selama fase snapshot.

Boolean

Tidak

false

Nilai yang valid:

  • true: dilewati.

    Pada fase inkremental, log dibaca mulai dari watermark rendah.

    Jika operator atau penyimpanan downstream mendukung idempotensi, kami merekomendasikan melewatkan pembacaan log pada fase lengkap. Ini mengurangi jumlah slot WAL, tetapi hanya semantik paling sedikit sekali (at-least-once) yang dapat dijamin.

  • false: tidak dilewati.

    Saat membaca split pada fase lengkap, log antara watermark rendah dan watermark tinggi dibaca untuk menjamin konsistensi.

    Jika SQL melakukan agregasi, join, atau operasi serupa, kami tidak merekomendasikan melewatkan pembacaan log pada fase lengkap.

Pemetaan tipe

Pemetaan tipe PostgreSQL ke Flink:

PostgreSQL CDC

Flink

SMALLINT

SMALLINT

INT2

SMALLSERIAL

SERIAL2

INTEGER

INT

SERIAL

BIGINT

BIGINT

BIGSERIAL

REAL

FLOAT

FLOAT4

FLOAT8

DOUBLE

DOUBLE PRECISION

NUMERIC(p, s)

DECIMAL(p, s)

DECIMAL(p, s)

BOOLEAN

BOOLEAN

DATE

DATE

TIME [(p)] [WITHOUT TIMEZONE]

TIME [(p)] [WITHOUT TIMEZONE]

TIMESTAMP [(p)] [WITHOUT TIMEZONE]

TIMESTAMP [(p)] [WITHOUT TIMEZONE]

CHAR(n)

STRING

CHARACTER(n)

VARCHAR(n)

CHARACTER VARYING(n)

TEXT

BYTEA

BYTES

Contoh

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

Flink CDC

VVR V11.4+ mendukung konektor PostgreSQL sebagai sumber Flink CDC.

Sintaksis

source:
  type: postgres
  name: PostgreSQL Source
  hostname: localhost
  port: 5432
  username: pg_username
  password: pg_password
  tables: db.scm.tbl
  slot.name: test_slot
  scan.startup.mode: initial
  server-time-zone: UTC
  connect.timeout: 120s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

Opsi konektor

Opsi

Deskripsi

Wajib

Tipe data

Default

Keterangan

type

Nama konektor.

Ya

STRING

–

Harus postgres.

name

Nama sumber data.

Tidak

STRING

–

–

hostname

Nama domain atau alamat IP server database PostgreSQL.

Ya

STRING

–

–

port

Port database PostgreSQL.

Tidak

INTEGER

5432

–

username

Username PostgreSQL.

Ya

STRING

–

–

password

Password PostgreSQL.

Ya

STRING

–

–

tables

Nama tabel yang akan ditangkap.

Regex didukung.

Ya

STRING

–

Penting

Saat ini, hanya tabel dalam database yang sama yang dapat ditangkap.

Titik (.) berfungsi sebagai pemisah dalam nama yang sepenuhnya memenuhi kualifikasi. Untuk menggunakan titik (.) dalam ekspresi reguler agar mencocokkan karakter apa pun, tambahkan backslash sebagai escape character. Contohnya: bdb.schema_\.*.order_\.*.

slot.name

Nama slot replikasi PostgreSQL.

Ya

STRING

–

Nama harus sesuai dengan aturan penamaan slot replikasi PostgreSQL dan dapat berisi huruf kecil, angka, dan garis bawah.

decoding.plugin.name

Nama plugin logical decoding PostgreSQL yang diinstal di server.

Tidak

STRING

pgoutput

Nilai yang valid: decoderbufs dan pgoutput.

tables.exclude

Tabel yang dikecualikan. Opsi ini berlaku setelah opsi tables. Regex didukung.

Tidak

STRING

–

Lihat opsi tables.

server-time-zone

Zona waktu sesi server database, seperti "Asia/Shanghai".

Tidak

STRING

–

Jika tidak diatur, zona waktu default sistem (ZoneId.systemDefault()) digunakan.

scan.incremental.snapshot.chunk.size

Ukuran (jumlah baris) setiap chunk dalam kerangka kerja snapshot inkremental.

Tidak

INTEGER

8096

Saat snapshot inkremental diaktifkan, tabel dibagi menjadi beberapa chunk untuk dibaca. Data chunk disimpan dalam cache di memori sebelum sepenuhnya dikonsumsi.

Chunk yang lebih kecil menghasilkan jumlah total chunk yang lebih besar untuk tabel tersebut. Meskipun ini mengurangi granularitas pemulihan kesalahan, hal ini dapat menyebabkan error kehabisan memori (OOM) dan throughput keseluruhan yang lebih rendah. Oleh karena itu, Anda perlu menyeimbangkan dan mengatur ukuran chunk yang wajar.

scan.snapshot.fetch.size

Jumlah maksimum catatan yang diambil sekaligus saat membaca data lengkap tabel.

Tidak

INTEGER

1024

–

scan.startup.mode

Mode startup untuk konsumsi data.

Tidak

STRING

initial

Nilai yang valid:

  • initial (default): Memindai snapshot saat startup pertama, lalu beralih ke data WAL terbaru.

  • latest-offset: Melewatkan pembacaan snapshot; Mulai membaca dari akhir WAL, artinya hanya membaca perubahan terbaru setelah konektor dimulai.

  • committed-offset: Melewatkan pembacaan snapshot; Mengonsumsi data WAL dari offset tertentu.

  • snapshot: Hanya mengonsumsi snapshot, bukan data inkremental.

scan.incremental.close-idle-reader.enabled

Menentukan apakah reader idle ditutup setelah snapshot selesai.

Tidak

BOOLEAN

false

Untuk mengaktifkan konfigurasi ini, atur execution.checkpointing.checkpoints-after-tasks-finish.enabled ke true.

scan.lsn-commit.checkpoints-num-delay

Jumlah checkpoint yang ditunda sebelum mulai melakukan commit offset LSN.

Tidak

INTEGER

3

Offset LSN checkpoint dilakukan commit secara bergilir untuk mencegah ketidakmampuan memulihkan dari state.

connect.timeout

Waktu maksimum konektor menunggu untuk terhubung ke server database PostgreSQL sebelum timeout.

Tidak

DURATION

30s

Nilai ini tidak boleh kurang dari 250 milidetik.

connect.max-retries

Maksimum upaya retry untuk konektor membangun koneksi.

Tidak

INTEGER

3

–

connection.pool.size

Ukuran kolam koneksi.

Tidak

INTEGER

20

–

jdbc.properties.*

Memungkinkan pengguna meneruskan properti URL JDBC kustom.

Tidak

STRING

20

Pengguna dapat meneruskan properti kustom, seperti 'jdbc.properties.useSSL' = 'false'.

heartbeat.interval

Interval pengiriman event heartbeat untuk melacak offset log WAL terbaru yang tersedia.

Tidak

DURATION

30s

–

debezium.*

Meneruskan properti Debezium ke Debezium Embedded Engine, yang digunakan untuk menangkap perubahan data dari server PostgreSQL.

Tidak

STRING

–

Properti konektor PostgreSQL Debezium: Dokumentasi Debezium.

chunk-meta.group.size

Ukuran metadata chunk.

Tidak

STRING

1000

Jika metadata lebih besar dari nilai ini, metadata tersebut diteruskan dalam beberapa bagian.

metadata.list

Daftar metadata yang dapat dibaca yang diteruskan ke downstream, yang dapat digunakan dalam modul transform.

Tidak

STRING

false

Gunakan koma (,) sebagai pemisah. Metadata berikut tersedia:

  • op_ts

  • table_name

  • database_name

  • schema_name

Catatan

Kolom metadata hanya tersedia di VVR 11.9 dan versi lebih baru.

scan.incremental.snapshot.unbounded-chunk-first.enabled

Mengirimkan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

Tidak

STRING

false

Ini adalah fitur eksperimental. Mengaktifkannya dapat mengurangi risiko error OOM saat TaskManager menyinkronkan chunk terakhir selama fase snapshot. Kami merekomendasikan menambahkan ini sebelum startup pertama pekerjaan.

schema-change.enabled

Menentukan apakah konektor secara otomatis melakukan inferensi perubahan skema.

Tidak

BOOLEAN

false

Setelah fitur ini diaktifkan, konektor membandingkan pesan yang diterima oleh pgoutput dengan skema tabel yang di-cache untuk melakukan inferensi event perubahan skema.

Catatan

Perubahan skema yang dapat diinferensi meliputi penambahan kolom, penghapusan kolom, penggantian nama kolom, dan perubahan tipe data kolom.

Parameter ini hanya tersedia di VVR 11.9 dan versi lebih baru.

Referensi