All Products
Search

This Product

    Realtime Compute for Apache Flink:PostgreSQL CDC

    Document Center

    Realtime Compute for Apache Flink:PostgreSQL CDC

    Last Updated:Aug 21, 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 dari saat data dihasilkan hingga ditarik oleh operator sumber.

    • currentEmitEventTimeLag: Interval dari 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 menskalakan turun untuk 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 berlangganan 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 berlangganan.

    • 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 tanpa status, 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 sumber daya tingkat pekerjaan dan kelolanya bersamaan dengan memulai dan menghentikan pekerjaan.

    • Hindari penggunaan ulang slot lama

      • Selalu gunakan nama slot baru. Menggunakan 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 yang 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 ditentukan.

      • Aturan pembuatan slot:

        • Snapshot inkremental dinonaktifkan: Hanya mendukung parallelism 1. Digunakan 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 dikembalikan. 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.

        Ini hanya memberikan semantik setidaknya-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 adalah waktu baca 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 ulang 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 ditimpa.

    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)

    Hal 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. Mendukung regex.

    STRING

    Ya

    –

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

    table-name

    Nama tabel PostgreSQL. Mendukung regex.

    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 setelahnya. Plugin ini harus diinstal.

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

    slot.name

    Nama slot logical decoding.

    STRING

    Wajib untuk VVR 8.0.1 dan versi setelahnya. Opsional untuk versi sebelumnya.

    flink (Sebelum 8.0.1)

    Atur slot.name yang 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 pengodean 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 secara 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 pembaca 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 sarankan melewati pembacaan log pada fase lengkap. Ini mengurangi jumlah slot WAL, tetapi hanya semantik setidaknya-sekali 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 menyarankan melewati 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.

    Mendukung regex.

    Ya

    STRING

    –
    Penting

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

    Titik (.) diperlakukan sebagai pemisah untuk nama lengkap. Untuk menggunakan titik (.) dalam ekspresi reguler agar cocok dengan karakter apa pun, tambahkan backslash sebagai escape. Contoh: bdb.schema_\.*.order_\.*.

    slot.name

    Nama slot replikasi PostgreSQL.

    Ya

    STRING

    –

    Nama harus mematuhi 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. Mendukung regex.

    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 di-cache di memori sebelum sepenuhnya dikonsumsi.

    Chunk yang lebih kecil menghasilkan jumlah total chunk yang lebih besar untuk tabel. 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 pembaca 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 meng-commit offset LSN.

    Tidak

    INTEGER

    3

    Offset LSN checkpoint di-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 di modul transform.

    Tidak

    STRING

    false

    Gunakan koma (,) sebagai pemisah. Saat ini, metadata yang tersedia adalah: op_ts.

    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 Pengelola Tugas menyinkronkan chunk terakhir selama fase snapshot. Kami sarankan menambahkan ini sebelum startup pertama pekerjaan.

    Referensi