All Products
Search

This Product

    Realtime Compute for Apache Flink:PolarDB-X CDC (Pratinjau publik)

    Document Center

    Realtime Compute for Apache Flink:PolarDB-X CDC (Pratinjau publik)

    Last Updated:Jun 04, 2026

    Pelajari cara menggunakan konektor PolarDB-X.

    Latar Belakang

    PolarDB for Xscale (PolarDB-X) adalah database terdistribusi cloud-native berkinerja tinggi dari Alibaba Cloud yang menawarkan throughput tinggi, kapasitas penyimpanan besar, latensi rendah, skalabilitas, dan ketersediaan ultra-tinggi.

    Penting

    Konektor ini memerlukan VVR 11.5 atau lebih baru dan PolarDB-X 2.0 atau lebih baru.

    Konektor PolarDB-X CDC hanya mendukung tabel sumber. Untuk mengkueri instans PolarDB-X sebagai tabel dimensi atau menulis ke dalamnya sebagai tabel sink, gunakan konektor MySQL (pratinjau publik).

    Kategori

    Deskripsi

    Jenis yang didukung

    Tabel sumber, sumber data ingestion

    Mode eksekusi

    Hanya mode streaming

    Format data

    N/A

    Metrik pemantauan spesifik

    • currentFetchEventTimeLag: Waktu antara pembuatan data dan saat Operator Sumber mengambilnya.

      Metrik ini hanya berlaku untuk fase binlog dan bernilai 0 selama fase snapshot.

    • currentEmitEventTimeLag: Waktu antara pembuatan data dan saat Operator Sumber mengirimkannya.

      Metrik ini hanya berlaku untuk fase binlog dan bernilai 0 selama fase snapshot.

    • sourceIdleTime: Waktu yang telah berlalu sejak tabel sumber terakhir menghasilkan data.

    Jenis API

    SQL, pekerjaan YAML data ingestion

    Mendukung pembaruan atau penghapusan data di tabel sink

    Tidak

    Fitur utama

    Konektor PolarDB-X CDC mengoptimalkan penguraian Binlog dengan filtering di sisi server. Konektor ini memangkas data Binlog yang tidak relevan di server PolarDB-X, sehingga meningkatkan throughput dan menghemat bandwidth jaringan.

    Langganan Binlog sesuai permintaan

    Dengan filtering Binlog di sisi server, konektor hanya mengirim log perubahan yang diperlukan ke client. Hal ini mengurangi lalu lintas jaringan dan meningkatkan throughput konsumsi data.

    Sebagai contoh, untuk berlangganan hanya pada perubahan dari tabel db.table1 dan db.table2 di server PolarDB-X Anda, konfigurasikan pekerjaan Flink SQL Anda sebagai berikut:

    CREATE TABLE polardbx_table_foo (
  ... -- Definisikan skema tabel di sini
) WITH (
  'connector' = 'polardbx-cdc',
  'database-name' = 'db',
  'table-name' = '.*',
  ..., -- Parameter lainnya
  'polardbx.binlog.include.tables' = 'db.table1,db.table2' -- Menangkap perubahan hanya dari tabel yang ditentukan
);

    Berbeda dengan konektor MySQL CDC yang memuat seluruh Binlog instans untuk filtering di sisi client, konektor PolarDB-X CDC melakukan filtering di sisi server. Kemampuan ini memungkinkan langganan client sesuai permintaan, sehingga secara signifikan mengurangi overhead I/O jaringan.

    Batasan penggunaan

    Filtering Binlog di sisi server dan langganan tingkat tabel memerlukan versi server PolarDB-X 2.5.0 atau lebih baru serta versi komponen Log Service 5.4.20 atau lebih baru.

    SQL

    Sintaksis

    CREATE TABLE polardbx_customer_table(
  `id` STRING,
  [columnName dataType,]*
  PRIMARY KEY(`id`) NOT ENFORCED
) WITH (
  'connector' = 'polardbx-cdc',
  'hosts' = 'pxc-**************-pub.polarx.rds.aliyuncs.com',
  'username' = 'pdx_user',
  'password' = 'pdx_password',
  'database' = 'full_db',
  'collection' = 'customers'
)

    Parameter WITH

    Parameter

    Deskripsi

    Tipe

    Wajib

    Default

    Keterangan

    connector

    Nama konektor.

    STRING

    Ya

    None

    Nilainya harus polardbx-cdc.

    hostname

    Alamat IP atau hostname database PolarDB-X.

    STRING

    Ya

    None

    Gunakan alamat kluster dari informasi koneksi instans.

    port

    Nomor port database PolarDB-X.

    INTEGER

    Tidak

    3306

    None

    username

    Username untuk database PolarDB-X.

    STRING

    Ya

    None

    None

    password

    Password untuk database PolarDB-X.

    STRING

    Ya

    None

    None

    database-name

    Nama database PolarDB-X.

    STRING

    Ya

    None

    Anda dapat menggunakan ekspresi reguler untuk membaca data dari beberapa database.

    Catatan

    Saat menggunakan ekspresi reguler, jangan gunakan karakter ^ dan $ untuk mencocokkan awal dan akhir string.

    table-name

    Nama tabel PolarDB-X.

    STRING

    Ya

    None

    Anda dapat menggunakan ekspresi reguler untuk membaca data dari beberapa tabel.

    Catatan

    Saat menggunakan ekspresi reguler, jangan gunakan karakter ^ dan $ untuk mencocokkan awal dan akhir string.

    server-time-zone

    Zona waktu sesi untuk koneksi database.

    STRING

    Tidak

    Zona waktu lingkungan runtime pekerjaan.

    Tentukan identifier zona waktu IANA, seperti Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP dari tabel sumber dikonversi menjadi STRING.

    scan.incremental.snapshot.chunk.size

    Jumlah baris dalam setiap chunk untuk pembacaan snapshot inkremental.

    INTEGER

    Tidak

    8096

    Selama fase snapshot, konektor membagi tabel menjadi chunk dan menyimpannya dalam cache di memori. Ukuran chunk yang lebih kecil meningkatkan jumlah total chunk. Meskipun hal ini meningkatkan granularitas toleransi kesalahan, risiko error out of memory (OOM) juga meningkat dan throughput berkurang. Tetapkan nilai yang sesuai untuk menyeimbangkan kinerja, toleransi kesalahan, dan konsumsi memori.

    scan.snapshot.fetch.size

    Jumlah maksimum baris yang diambil sekaligus saat membaca snapshot tabel.

    INTEGER

    Tidak

    1024

    None

    connect.timeout

    Timeout untuk membuat koneksi ke database PolarDB-X.

    DURATION

    Tidak

    30s

    None

    connection.pool.size

    Ukuran kolam koneksi database.

    INTEGER

    Tidak

    20

    Kolam koneksi database menggunakan kembali koneksi untuk mengurangi overhead koneksi.

    connect.max-retries

    Jumlah maksimum percobaan ulang setelah gagal terhubung ke database PolarDB-X.

    INTEGER

    Tidak

    3

    None

    scan.startup.mode

    Mode startup untuk konsumsi data.

    STRING

    Tidak

    initial

    Nilai yang valid:

    • initial (default): Saat startup pertama kali, konektor mengambil snapshot lengkap tabel lalu membaca perubahan inkremental dari Binlog.

    • latest-offset: Tidak melakukan snapshot. Saat startup pertama kali, konektor mulai membaca dari akhir Binlog, hanya menangkap perubahan yang terjadi setelah pekerjaan dimulai.

    • earliest-offset: Tidak pernah melakukan snapshot. Mulai membaca dari offset Binlog paling awal yang tersedia.

    • specific-offset: Tidak pernah melakukan snapshot. Mulai membaca dari offset tertentu. Tentukan offset dengan menggunakan scan.startup.specific-offset.file dan scan.startup.specific-offset.pos, atau gunakan scan.startup.specific-offset.gtid-set.

    • timestamp: Tidak pernah melakukan snapshot. Mulai membaca Binlog dari timestamp tertentu. Timestamp ditentukan dalam milidetik menggunakan scan.startup.timestamp-millis.

    Penting

    Untuk mode startup earliest-offset, specific-offset, dan timestamp, skema tabel saat startup harus sesuai dengan skema pada offset yang ditentukan. Ketidaksesuaian skema akan menyebabkan pekerjaan gagal. Pastikan skema tabel tidak berubah antara offset Binlog yang ditentukan dan waktu startup pekerjaan.

    scan.startup.specific-offset.file

    Nama file Binlog tempat memulai ketika scan.startup.mode adalah specific-offset.

    STRING

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke specific-offset. Contoh: mysql-bin.000003.

    scan.startup.specific-offset.pos

    Offset dalam file Binlog yang ditentukan ketika scan.startup.mode diatur ke specific-offset.

    INTEGER

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke specific-offset.

    scan.startup.specific-offset.gtid-set

    Set GTID tempat memulai ketika scan.startup.mode diatur ke specific-offset.

    STRING

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke specific-offset. Contoh: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.

    scan.startup.timestamp-millis

    Timestamp startup dalam milidetik ketika scan.startup.mode diatur ke timestamp.

    LONG

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke timestamp. Satuannya adalah milidetik.

    scan.startup.specific-offset.skip-events

    Jumlah event Binlog yang dilewati saat membaca dari offset tertentu.

    INTEGER

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke specific-offset.

    scan.startup.specific-offset.skip-rows

    Jumlah perubahan baris yang dilewati saat membaca dari offset tertentu. Satu event Binlog dapat berisi beberapa perubahan baris.

    INTEGER

    Tidak

    None

    Parameter ini hanya digunakan ketika scan.startup.mode diatur ke specific-offset.

    heartbeat.interval

    Interval pengiriman event heartbeat oleh sumber untuk memajukan offset Binlog.

    DURATION

    Tidak

    None

    Event heartbeat memajukan offset Binlog sumber. Hal ini mencegah Binlog di-purge pada sumber yang idle. Jika Binlog di-purge, pekerjaan gagal dan memerlukan restart tanpa status untuk pulih.

    chunk-meta.group.size

    Ukuran metadata chunk.

    INTEGER

    Tidak

    1000

    Jika ukuran metadata melebihi nilai ini, metadata tersebut dibagi menjadi beberapa bagian untuk transmisi.

    chunk-key.even-distribution.factor.upper-bound

    Batas atas faktor distribusi chunk untuk partisi merata.

    DOUBLE

    Tidak

    1000.0

    Jika faktor distribusi lebih besar dari nilai ini, chunking tidak merata digunakan.

    Faktor distribusi chunk = (MAX(kunci-chunk) - MIN(kunci-chunk) + 1) / jumlah total baris.

    chunk-key.even-distribution.factor.lower-bound

    Batas bawah faktor distribusi chunk untuk partisi merata.

    DOUBLE

    Tidak

    0.05

    Jika faktor distribusi kurang dari nilai ini, chunking tidak merata digunakan.

    Faktor distribusi chunk = (MAX(kunci-chunk) - MIN(kunci-chunk) + 1) / jumlah total baris.

    scan.newly-added-table.enabled

    Menentukan apakah akan memindai tabel baru yang sesuai kriteria tangkapan saat restart dari Checkpoint.

    BOOLEAN

    Tidak

    false

    Jika diaktifkan, pekerjaan yang direstart dari Checkpoint atau Savepoint akan memindai dan menyinkronkan tabel baru yang sesuai pola 'table-name'. Pekerjaan juga akan berhenti melacak tabel yang tidak lagi sesuai.

    scan.incremental.snapshot.chunk.key-column

    Menentukan kolom yang digunakan untuk pembagian data selama fase snapshot.

    STRING

    Lihat Keterangan

    None

    • Wajib untuk tabel tanpa primary key. Kolom yang ditentukan harus non-null.

    • Opsional untuk tabel dengan primary key. Hanya satu kolom dari primary key yang didukung.

    scan.incremental.close-idle-reader.enabled

    Menentukan apakah akan menutup reader yang idle setelah fase snapshot selesai.

    BOOLEAN

    Tidak

    false

    Agar pengaturan ini berlaku, execution.checkpointing.checkpoints-after-tasks-finish.enabled juga harus diatur ke true.

    scan.incremental.snapshot.backfill.skip

    Menentukan apakah akan melewati proses backfill selama fase pembacaan snapshot.

    BOOLEAN

    Tidak

    false

    Nilai yang valid:

    • true: Melewati proses backfill selama fase pembacaan snapshot.

    • false (default): Tidak melewati proses backfill selama fase pembacaan snapshot.

    Jika backfill dilewati, perubahan yang terjadi pada tabel selama fase snapshot akan dibaca dalam fase inkremental berikutnya, bukan digabungkan ke dalam snapshot.

    Penting

    Melewati backfill dapat menyebabkan ketidakkonsistenan data karena perubahan yang terjadi selama snapshot diputar ulang dalam fase inkremental. Mode ini hanya menjamin semantik at-least-once.

    scan.parse.online.schema.changes.enabled

    Menentukan apakah akan mencoba mengurai event DDL untuk perubahan skema tanpa lock RDS selama fase inkremental.

    BOOLEAN

    Tidak

    false

    Nilai yang valid:

    • true: Mengurai event DDL untuk perubahan skema tanpa lock RDS.

    • false (default): Tidak mengurai event DDL untuk perubahan skema tanpa lock RDS.

    Ini adalah fitur eksperimental. Sebelum melakukan perubahan skema tanpa lock di produksi, buat Savepoint untuk pekerjaan Flink agar dapat dipulihkan jika diperlukan.

    scan.only.deserialize.captured.tables.changelog.enabled

    Menentukan apakah hanya akan mendeserialisasi event perubahan untuk tabel yang ditangkap selama fase inkremental.

    BOOLEAN

    Tidak

    true

    Nilai yang valid:

    • true (default): Mendeserialisasi data perubahan hanya untuk tabel target, yang dapat mempercepat pembacaan Binlog.

    • false: Mendeserialisasi data perubahan untuk semua tabel.

    scan.read-changelog-as-append-only.enabled

    Menentukan apakah akan mengonversi aliran changelog menjadi aliran append-only.

    BOOLEAN

    Tidak

    false

    Nilai yang valid:

    • true: Mengonversi semua jenis pesan (termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER) menjadi pesan INSERT. Gunakan opsi ini hanya dalam kasus tertentu, misalnya untuk menyimpan catatan baris yang dihapus dari sumber.

    • false (default): Semua jenis pesan dikirim ke downstream apa adanya.

    scan.parallel-deserialize-changelog.enabled

    Menentukan apakah akan menggunakan beberapa thread untuk mendeserialisasi event perubahan selama fase inkremental.

    BOOLEAN

    Tidak

    false

    Nilai yang valid:

    • true: Menggunakan beberapa thread untuk deserialisasi guna mempercepat pembacaan sekaligus mempertahankan urutan event Binlog.

    • false (default): Menggunakan satu thread untuk deserialisasi.

    scan.parallel-deserialize-changelog.handler.size

    Jumlah handler event yang digunakan saat mendeserialisasi event perubahan secara paralel.

    INTEGER

    Tidak

    2

    None

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

    Menentukan apakah akan mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    BOOLEAN

    Tidak

    false

    Nilai yang valid:

    • true: Mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    • false (default): Tidak mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    Ini adalah fitur eksperimental. Mengaktifkannya dapat mengurangi risiko error out of memory (OOM) pada TaskManager saat menyinkronkan chunk terakhir snapshot. Pengaturan ini harus dikonfigurasi sebelum startup pertama pekerjaan.

    polardbx.binlog.ignore.archive-events.enabled

    Menentukan apakah akan mengabaikan event arsip (terutama event DELETE) dalam Binlog PolarDB-X.

    BOOLEAN

    Tidak

    false

    polardbx.binlog.ignore.query-events.enabled

    Menentukan apakah akan mengabaikan event ROWS_QUERY_LOG_EVENT dalam Binlog PolarDB-X.

    BOOLEAN

    Tidak

    false

    polardbx.binlog.include.tables

    Hanya membaca event Binlog dari tabel yang ditentukan. Gunakan koma (,) untuk memisahkan beberapa nama tabel.

    STRING

    Tidak

    None

    polardbx.binlog.exclude.tables

    Mengabaikan event Binlog dari tabel yang ditentukan. Gunakan koma (,) untuk memisahkan beberapa nama tabel.

    STRING

    Tidak

    None

    Pemetaan tipe

    Tipe PolarDB-X

    Tipe Flink

    TINYINT

    TINYINT

    SMALLINT

    SMALLINT

    TINYINT UNSIGNED

    TINYINT UNSIGNED ZEROFILL

    INT

    INT

    MEDIUMINT

    SMALLINT UNSIGNED

    SMALLINT UNSIGNED ZEROFILL

    BIGINT

    BIGINT

    INT UNSIGNED

    INT UNSIGNED ZEROFILL

    MEDIUMINT UNSIGNED

    MEDIUMINT UNSIGNED ZEROFILL

    BIGINT UNSIGNED

    DECIMAL(20, 0)

    BIGINT UNSIGNED ZEROFILL

    SERIAL

    FLOAT [UNSIGNED] [ZEROFILL]

    FLOAT

    DOUBLE [UNSIGNED] [ZEROFILL]

    DOUBLE

    DOUBLE PRECISION [UNSIGNED] [ZEROFILL]

    REAL [UNSIGNED] [ZEROFILL]

    NUMERIC(p, s) [UNSIGNED] [ZEROFILL]

    DECIMAL(p, s)

    DECIMAL(p, s) [UNSIGNED] [ZEROFILL]

    BOOLEAN

    BOOLEAN

    TINYINT(1)

    DATE

    DATE

    TIME [(p)]

    TIME [(p)] [WITHOUT TIME ZONE]

    DATETIME [(p)]

    TIMESTAMP [(p)] [WITHOUT TIME ZONE]

    TIMESTAMP [(p)]

    TIMESTAMP_LTZ [(p)]

    TIMESTAMP [(p)] WITH LOCAL TIME ZONE

    CHAR(n)

    STRING

    VARCHAR(n)

    TEXT

    BINARY

    BYTES

    VARBINARY

    BLOB

    Data ingestion

    Realtime Compute for Apache Flink 11.6 kini mendukung konektor PolarDB-X sebagai sumber data untuk pekerjaan YAML data ingestion.

    Sintaksis

    source:
   type: polardbx
   name: PolarDB-X Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: pdb.order_table
   # Abaikan event arsip dalam binlog
   polardbx.binlog.ignore.archive-events.enabled: true
   # Abaikan event kueri dalam binlog
   polardbx.binlog.ignore.query-events.enabled: true
   # Berlangganan hanya pada binlog untuk pdb.order_table untuk menghemat bandwidth.
   polardbx.binlog.include.tables: pdb.order_table

sink:
  type: values

    Parameter

    Parameter

    Deskripsi

    Wajib

    Tipe

    Default

    Keterangan

    type

    Jenis sumber data.

    Ya

    STRING

    N/A

    Nilainya harus polardbx.

    name

    Nama sumber data.

    Tidak

    STRING

    N/A

    N/A

    hostname

    Alamat IP atau hostname instans PolarDB-X.

    Ya

    STRING

    N/A

    Alamat VPC disarankan.

    Catatan

    Jika instans PolarDB-X dan ruang kerja Realtime Compute for Apache Flink Anda tidak berada dalam VPC yang sama, Anda harus membuat koneksi cross-VPC atau mengakses instans melalui jaringan publik. Untuk detailnya, lihat Manajemen dan Operasi Ruang Kerja dan Bagaimana kluster Flink yang dikelola sepenuhnya mengakses jaringan publik?.

    username

    Username untuk terhubung ke database PolarDB-X.

    Ya

    STRING

    N/A

    N/A

    password

    Password untuk username yang ditentukan.

    Ya

    STRING

    N/A

    N/A

    tables

    Tabel PolarDB-X yang akan disinkronkan.

    Ya

    STRING

    N/A

    • Mendukung ekspresi reguler untuk membaca data dari beberapa tabel.

    • Anda dapat memisahkan beberapa ekspresi reguler dengan koma (,).

    Catatan

    • Jangan gunakan anchor awal dan akhir ^ dan $ dalam ekspresi reguler.

    • Titik (.) memisahkan nama database dan nama tabel. Untuk mencocokkan titik literal, Anda harus meng-escape-nya dengan backslash. Contoh: db0.\.*, db1.user_table_[0-9]+, atau db[1-2].[app|web]order_\.*.

    tables.exclude

    Tabel yang dikecualikan dari sinkronisasi.

    Tidak

    STRING

    N/A

    • Mendukung ekspresi reguler untuk mengecualikan beberapa tabel.

    • Anda dapat memisahkan beberapa ekspresi reguler dengan koma (,).

    Catatan

    Titik (.) memisahkan nama database dan nama tabel. Untuk mencocokkan titik literal, Anda harus meng-escape-nya dengan backslash. Contoh: db0.\.*, db1.user_table_[0-9]+, atau db[1-2].[app|web]order_\.*.

    port

    Port instans PolarDB-X.

    Tidak

    INTEGER

    3306

    N/A

    schema-change.enabled

    Menentukan apakah akan mengirim event perubahan skema.

    Tidak

    BOOLEAN

    true

    N/A

    jdbc.properties.*

    Parameter koneksi kustom untuk URL JDBC.

    Tidak

    STRING

    N/A

    Anda dapat meneruskan parameter koneksi kustom. Misalnya, untuk menonaktifkan SSL, atur 'jdbc.properties.useSSL' = 'false'.

    debezium.*

    Parameter Debezium kustom untuk membaca binlog.

    Tidak

    STRING

    N/A

    Anda dapat meneruskan parameter Debezium kustom. Misalnya, atur 'debezium.event.deserialization.failure.handling.mode' = 'ignore' untuk menentukan cara konektor menangani error deserialisasi.

    scan.incremental.snapshot.chunk.size

    Ukuran setiap chunk, dalam jumlah baris.

    Tidak

    INTEGER

    8096

    Tabel PolarDB-X dibagi menjadi beberapa chunk untuk dibaca. Data dalam setiap chunk disimpan dalam cache di memori hingga sepenuhnya dibaca.

    Ukuran chunk yang lebih kecil meningkatkan jumlah total chunk, yang memberikan pemulihan kesalahan lebih granular tetapi dapat meningkatkan risiko OOM dan mengurangi throughput keseluruhan. Anda harus menyeimbangkan trade-off ini dan mengonfigurasi ukuran chunk yang wajar.

    scan.snapshot.fetch.size

    Jumlah maksimum catatan yang diambil sekaligus selama snapshot tabel lengkap.

    Tidak

    INTEGER

    1024

    N/A

    scan.startup.mode

    Mode startup untuk mengonsumsi data.

    Tidak

    STRING

    initial

    Nilai yang valid:

    • initial (Default): Mengambil snapshot awal tabel lalu membaca perubahan terbaru dari binlog.

    • latest-offset: Melewati fase snapshot dan membaca binlog dari posisi terbaru, hanya menangkap perubahan yang terjadi setelah konektor dimulai.

    • earliest-offset: Melewati fase snapshot dan mulai membaca dari posisi binlog paling awal yang tersedia.

    • specific-offset: Melewati fase snapshot dan mulai membaca dari posisi tertentu. Tentukan posisi dengan menggunakan scan.startup.specific-offset.file dan scan.startup.specific-offset.pos, atau dengan menggunakan scan.startup.specific-offset.gtid-set.

    • timestamp: Melewati fase snapshot dan mulai membaca binlog dari timestamp tertentu, yang ditentukan dalam milidetik oleh parameter scan.startup.timestamp-millis.

    Penting

    Untuk mode startup earliest-offset, specific-offset, dan timestamp, pekerjaan gagal dengan error jika skema tabel saat startup berbeda dari skema tabel pada waktu posisi startup yang ditentukan. Dengan kata lain, saat menggunakan ketiga mode startup ini, Anda harus memastikan bahwa skema tabel yang sesuai tidak berubah antara posisi konsumsi Binlog yang ditentukan dan waktu startup pekerjaan.

    scan.startup.specific-offset.file

    Nama file binlog untuk posisi startup.

    Tidak

    STRING

    N/A

    Wajib saat scan.startup.mode diatur ke specific-offset. Format contoh: mysql-bin.000003.

    scan.startup.specific-offset.pos

    Offset dalam file binlog yang ditentukan untuk memulai.

    Tidak

    INTEGER

    N/A

    Wajib saat scan.startup.mode diatur ke specific-offset.

    scan.startup.specific-offset.gtid-set

    Set GTID untuk posisi startup.

    Tidak

    STRING

    N/A

    Digunakan saat scan.startup.mode diatur ke specific-offset. Format contoh set GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.

    scan.startup.timestamp-millis

    Timestamp dalam milidetik untuk posisi startup.

    Tidak

    LONG

    N/A

    Wajib saat scan.startup.mode diatur ke timestamp.

    Penting

    Saat Anda menentukan timestamp, konektor menemukan file awal yang benar dengan membaca event pertama setiap file binlog untuk memeriksa timestamp-nya. Pastikan file binlog target untuk timestamp yang ditentukan belum di-purge dari database dan masih dapat dibaca.

    server-time-zone

    Zona waktu sesi yang digunakan oleh database.

    Tidak

    STRING

    Zona waktu lingkungan runtime pekerjaan Flink (zona ketersediaan ruang kerja Anda).

    Contoh: Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP di PolarDB-X dikonversi menjadi STRING. Untuk informasi lebih lanjut, lihat dokumentasi Debezium tentang tipe waktu.

    scan.startup.specific-offset.skip-events

    Jumlah event binlog yang dilewati saat membaca dari posisi tertentu.

    Tidak

    INTEGER

    N/A

    Digunakan saat scan.startup.mode diatur ke specific-offset.

    scan.startup.specific-offset.skip-rows

    Jumlah perubahan baris yang dilewati saat membaca dari posisi tertentu. Satu event binlog dapat berisi beberapa perubahan baris.

    Tidak

    INTEGER

    N/A

    Digunakan saat scan.startup.mode diatur ke specific-offset.

    connect.timeout

    Waktu maksimum menunggu koneksi database sebelum mencoba ulang.

    Tidak

    DURATION

    30s

    N/A

    connect.max-retries

    Jumlah maksimum percobaan ulang untuk koneksi database yang gagal.

    Tidak

    INTEGER

    3

    N/A

    connection.pool.size

    Ukuran kolam koneksi database.

    Tidak

    INTEGER

    20

    Kolam koneksi menggunakan kembali koneksi untuk mengurangi overhead koneksi.

    heartbeat.interval

    Interval pengiriman event heartbeat oleh sumber untuk memajukan posisi binlog.

    Tidak

    DURATION

    30s

    Event heartbeat memajukan posisi binlog untuk tabel yang jarang diperbarui. Tanpa event ini, posisi binlog mungkin tidak maju, yang dapat menyebabkan binlog kedaluwarsa. Binlog yang kedaluwarsa menyebabkan pekerjaan gagal, sehingga memerlukan restart tanpa status untuk pulih.

    scan.incremental.snapshot.chunk.key-column

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

    Tidak

    STRING

    N/A

    Anda hanya dapat memilih satu kolom dari primary key.

    chunk-meta.group.size

    Ukuran metadata chunk.

    Tidak

    INTEGER

    1000

    Jika ukuran metadata melebihi nilai ini, metadata tersebut dibagi menjadi beberapa bagian untuk transmisi.

    chunk-key.even-distribution.factor.lower-bound

    Batas bawah faktor distribusi chunk untuk menentukan apakah akan menggunakan chunking merata.

    Tidak

    DOUBLE

    0.05

    Jika faktor distribusi kurang dari nilai ini, chunking tidak merata digunakan.

    Faktor distribusi chunk = (MAX(kunci-chunk) - MIN(kunci-chunk) + 1) / jumlah total baris data.

    chunk-key.even-distribution.factor.upper-bound

    Batas atas faktor distribusi chunk untuk menentukan apakah akan menggunakan chunking merata.

    Tidak

    DOUBLE

    1000.0

    Jika faktor distribusi lebih besar dari nilai ini, chunking tidak merata digunakan.

    Faktor distribusi chunk = (MAX(kunci-chunk) - MIN(kunci-chunk) + 1) / jumlah total baris data.

    scan.incremental.close-idle-reader.enabled

    Menentukan apakah akan menutup reader yang idle setelah fase snapshot selesai.

    Tidak

    BOOLEAN

    false

    Agar pengaturan ini berlaku, Anda harus mengatur execution.checkpointing.checkpoints-after-tasks-finish.enabled ke true.

    scan.only.deserialize.captured.tables.changelog.enabled

    Menentukan apakah hanya akan mendeserialisasi event perubahan untuk tabel yang ditentukan selama fase inkremental.

    Tidak

    BOOLEAN

    true

    Nilai yang valid:

    • true: Mendeserialisasi data perubahan hanya untuk tabel target, yang mempercepat pembacaan binlog.

    • false: Mendeserialisasi data perubahan untuk semua tabel.

    scan.parallel-deserialize-changelog.enabled

    Menentukan apakah akan menggunakan beberapa thread untuk mengurai event perubahan selama fase inkremental.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Menggunakan beberapa thread untuk deserialisasi event perubahan sekaligus mempertahankan urutan event binlog, yang mempercepat pembacaan.

    • false (Default): Menggunakan satu thread untuk deserialisasi event.

    scan.parallel-deserialize-changelog.handler.size

    Jumlah handler event yang digunakan saat mengurai event perubahan secara paralel.

    Tidak

    INTEGER

    2

    N/A

    metadata-column.include-list

    Kolom metadata yang diteruskan ke sink downstream.

    Tidak

    STRING

    N/A

    Metadata yang tersedia meliputi op_ts, es_ts, query_log, file, dan pos. Anda dapat menggunakan koma untuk memisahkan beberapa kolom metadata.

    Catatan

    Konektor YAML CDC PolarDB-X tidak memerlukan atau mendukung penambahan kolom metadata untuk nama database, nama tabel, atau op_type. Anda dapat menggunakan __data_event_type__ dalam ekspresi transformasi untuk mendapatkan tipe data perubahan, atau menggunakan __schema_name__ dan __table_name__ untuk mendapatkan nama database dan tabel.

    Penting

    Kolom file berisi nama file binlog (fase inkremental) atau string kosong (fase snapshot). Kolom pos berisi offset binlog (fase inkremental) atau 0 (fase snapshot).

    scan.newly-added-table.enabled

    Saat restart dari checkpoint atau savepoint, menentukan apakah akan menyinkronkan tabel baru yang sesuai pola tabel dan berhenti melacak tabel yang tidak lagi sesuai.

    Tidak

    BOOLEAN

    false

    Pengaturan ini hanya berlaku saat restart dari checkpoint atau savepoint.

    scan.binlog.newly-added-table.enabled

    Selama fase inkremental, menentukan apakah akan mengirim data untuk tabel baru yang sesuai.

    Tidak

    BOOLEAN

    false

    Tidak dapat diaktifkan secara bersamaan dengan scan.newly-added-table.enabled.

    scan.incremental.snapshot.chunk.key-column

    Menentukan kolom kunci pemisah untuk chunking selama fase snapshot untuk tabel tertentu.

    Tidak

    STRING

    N/A

    • Gunakan titik dua (:) untuk menghubungkan nama tabel dan nama kolom guna mendefinisikan aturan. Nama tabel dapat berupa ekspresi reguler. Anda dapat mendefinisikan beberapa aturan yang dipisahkan dengan titik koma (;). Contoh: db1.user_table_[0-9]+:col1;db[1-2].[app|web]_order_\\.*:col2.

    • Parameter ini wajib untuk tabel tanpa primary key, dan kolom yang dipilih harus bertipe non-null (NOT NULL). Untuk tabel dengan primary key, parameter ini opsional, dan Anda hanya dapat memilih satu kolom dari primary key.

    scan.parse.online.schema.changes.enabled

    Selama fase inkremental, menentukan apakah akan mengurai event DDL untuk perubahan skema tanpa lock ApsaraDB RDS.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Mengurai event DDL untuk perubahan skema tanpa lock ApsaraDB RDS.

    • false (Default): Tidak mengurai event DDL untuk perubahan skema tanpa lock ApsaraDB RDS.

    Ini adalah fitur eksperimental. Sebelum melakukan perubahan tanpa lock online, kami menyarankan Anda membuat savepoint untuk pekerjaan Flink agar dapat dipulihkan jika diperlukan.

    scan.incremental.snapshot.backfill.skip

    Menentukan apakah akan melewati backfill selama fase pembacaan snapshot.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Melewati backfill selama fase pembacaan snapshot.

    • false (Default): Tidak melewati backfill selama fase pembacaan snapshot.

    Jika backfill dilewati, perubahan pada tabel yang terjadi selama fase snapshot akan dibaca dalam fase inkremental berikutnya, bukan digabungkan ke dalam snapshot.

    Penting

    Melewati backfill dapat menyebabkan ketidakkonsistenan data karena perubahan dari fase snapshot mungkin diputar ulang. Mode ini menyediakan semantik at-least-once.

    treat-tinyint1-as-boolean.enabled

    Menentukan apakah akan memperlakukan tipe data TINYINT(1) sebagai BOOLEAN.

    Tidak

    BOOLEAN

    true

    Nilai yang valid:

    • true (Default): Memperlakukan tipe TINYINT(1) sebagai BOOLEAN.

    • false: Tidak memperlakukan tipe TINYINT(1) sebagai BOOLEAN.

    treat-timestamp-as-datetime-enabled

    Menentukan apakah akan memperlakukan tipe data TIMESTAMP sebagai DATETIME.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Memperlakukan tipe TIMESTAMP PolarDB-X sebagai DATETIME dan memetakannya ke tipe CDC TIMESTAMP.

    • false (Default): Memetakan tipe TIMESTAMP PolarDB-X ke tipe CDC TIMESTAMP_LTZ.

    Tipe TIMESTAMP PolarDB-X menyimpan waktu UTC dan dipengaruhi oleh zona waktu. Tipe DATETIME PolarDB-X menyimpan waktu literal dan tidak dipengaruhi oleh zona waktu.

    Saat diaktifkan, parameter ini mengonversi data TIMESTAMP PolarDB-X ke tipe DATETIME berdasarkan server-time-zone.

    include-comments.enabled

    Menentukan apakah akan menyinkronkan komentar tabel dan kolom.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Menyinkronkan komentar tabel dan kolom.

    • false (Default): Tidak menyinkronkan komentar tabel dan kolom.

    Mengaktifkan parameter ini meningkatkan penggunaan memori pekerjaan.

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

    Menentukan apakah akan mengirimkan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    Tidak

    BOOLEAN

    false

    Nilai yang valid:

    • true: Mengirimkan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    • false (Default): Tidak mengirimkan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.

    Ini adalah fitur eksperimental. Mengaktifkannya dapat mengurangi risiko error OOM pada TaskManager saat chunk terakhir sedang disinkronkan. Kami menyarankan Anda menambahkan parameter ini sebelum startup pertama pekerjaan.

    binlog.session.network.timeout

    Timeout jaringan untuk koneksi binlog.

    Tidak

    DURATION

    10m

    Jika diatur ke 0s, timeout default sisi server digunakan.

    scan.rate-limit.records-per-second

    Jumlah maksimum catatan yang dapat dipancarkan sumber per detik.

    Tidak

    LONG

    N/A

    Gunakan parameter ini untuk membatasi throughput pembacaan data. Batasan ini berlaku untuk fase snapshot maupun inkremental.

    Metrik numRecordsOutPerSecond sumber menunjukkan jumlah catatan yang dipancarkan per detik untuk seluruh aliran data. Anda dapat menggunakan metrik ini untuk menyesuaikan parameter ini.

    Selama fase pembacaan snapshot, Anda mungkin juga perlu mengurangi jumlah baris yang dibaca per batch dengan menurunkan nilai parameter scan.incremental.snapshot.chunk.size.

    include-binlog-meta.enable

    Menentukan apakah akan menyertakan metadata binlog PolarDB-X mentah, seperti GTID dan posisi binlog, dalam catatan yang dipancarkan.

    Tidak

    BOOLEAN

    false

    Berlaku untuk skenario sinkronisasi binlog mentah, seperti mengganti pipeline sinkronisasi berbasis Canal yang sudah ada.

    polardbx.binlog.ignore.archive-events.enabled

    Menentukan apakah akan mengabaikan event arsip (terutama event DELETE) dalam binlog PolarDB-X.

    Tidak

    BOOLEAN

    false

    polardbx.binlog.ignore.query-events.enabled

    Menentukan apakah akan mengabaikan event ROWS_QUERY_LOG_EVENT dalam binlog PolarDB-X.

    Tidak

    BOOLEAN

    false

    polardbx.binlog.include.tables

    Daftar tabel yang dipisahkan koma untuk berlangganan event binlog.

    Tidak

    STRING

    N/A
    Catatan

    Parameter ini hanya berlaku untuk fase inkremental (binlog) dan tidak berlaku untuk fase snapshot.

    polardbx.binlog.exclude.tables

    Daftar tabel yang dipisahkan koma yang event binlog-nya harus diabaikan.

    Tidak

    STRING

    N/A
    Catatan

    Parameter ini hanya berlaku untuk fase inkremental (binlog) dan tidak berlaku untuk fase snapshot.