Realtime Compute for Apache Flink:Konektor MySQL YAML

Parameter

Deskripsi

Wajib

Tipe data

Nilai default

Catatan

type

Jenis sumber data.

Ya

STRING

None

Nilainya harus mysql.

name

Nama sumber data.

Tidak

STRING

None

None.

hostname

Alamat IP atau hostname database MySQL.

Ya

STRING

None

Kami menyarankan agar Anda menentukan alamat VPC.

username

Username untuk layanan database MySQL.

Ya

STRING

None

None.

password

Password untuk layanan database MySQL.

Ya

STRING

None

None.

tables

Tabel data MySQL yang akan disinkronkan.

Ya

STRING

None

• Parameter ini mendukung ekspresi reguler untuk membaca data dari beberapa tabel.

• Anda dapat menggunakan koma untuk memisahkan beberapa ekspresi reguler.

tables.exclude

Tabel yang akan dikecualikan dari sinkronisasi.

Tidak

STRING

None

• Parameter ini mendukung ekspresi reguler untuk mengecualikan beberapa tabel.

• Anda dapat menggunakan koma untuk memisahkan beberapa ekspresi reguler.

port

Nomor port layanan database MySQL.

Tidak

INTEGER

3306

None.

schema-change.enabled

Menentukan apakah akan mengirimkan event perubahan skema.

Tidak

BOOLEAN

true

None.

server-id

ID numerik atau rentang untuk klien database yang digunakan untuk sinkronisasi.

Tidak

STRING

Nilai acak antara 5400 dan 6400 dihasilkan.

ID ini harus unik secara global dalam kluster MySQL. Tetapkan ID berbeda untuk setiap pekerjaan yang terhubung ke database yang sama. Parameter ini juga mendukung format rentang ID, seperti 5400-5408.

jdbc.properties.*

Parameter koneksi kustom dalam URL JDBC.

Tidak

STRING

None

Anda dapat meneruskan parameter koneksi kustom. Misalnya, untuk tidak menggunakan protokol SSL, Anda dapat mengonfigurasi 'jdbc.properties.useSSL' = 'false'.

Untuk informasi lebih lanjut tentang parameter koneksi yang didukung, lihat MySQL Configuration Properties.

debezium.*

Parameter kustom untuk Debezium guna membaca log biner.

Tidak

STRING

None

Anda dapat meneruskan parameter Debezium kustom. Misalnya, gunakan 'debezium.event.deserialization.failure.handling.mode'='ignore' untuk menentukan logika penanganan error parsing.

scan.incremental.snapshot.chunk.size

Ukuran setiap chunk dalam jumlah baris.

Tidak

INTEGER

8096

Tabel MySQL dibagi menjadi beberapa chunk untuk dibaca. Data sebuah chunk di-cache di memori sebelum sepenuhnya dibaca.

Semakin sedikit baris yang dimiliki setiap chunk, semakin besar jumlah total chunk dalam tabel. Meskipun hal ini mengurangi granularitas pemulihan kesalahan, namun dapat menyebabkan error OOM dan menurunkan throughput keseluruhan. Oleh karena itu, Anda perlu membuat pertimbangan dan menetapkan ukuran chunk yang wajar.

scan.snapshot.fetch.size

Jumlah maksimum catatan yang ditarik sekaligus saat membaca data lengkap sebuah tabel.

Tidak

INTEGER

1024

None.

scan.startup.mode

Mode startup untuk konsumsi data.

Tidak

STRING

initial

Nilai yang valid:

• initial (default): Saat startup pertama kali atau startup tanpa status, konektor memindai data historis lengkap lalu membaca data log biner terbaru.

• latest-offset: Saat startup pertama kali atau startup tanpa status, konektor tidak memindai data historis. Konektor mulai membaca dari akhir log biner, artinya hanya membaca perubahan terbaru setelah konektor dimulai.

• earliest-offset: Konektor tidak memindai data historis. Konektor mulai membaca dari log biner paling awal yang tersedia.

• specific-offset: Konektor tidak memindai data historis. Konektor mulai dari offset log biner tertentu. Anda dapat menentukan offset dengan mengonfigurasi scan.startup.specific-offset.file dan scan.startup.specific-offset.pos, atau hanya mengonfigurasi scan.startup.specific-offset.gtid-set untuk memulai dari set GTID tertentu.

• timestamp: Konektor tidak memindai data historis. Konektor mulai membaca log biner dari timestamp tertentu. Timestamp ditentukan oleh scan.startup.timestamp-millis dalam milidetik.

scan.startup.specific-offset.file

Nama file log biner untuk offset awal saat menggunakan mode startup specific-offset.

Tidak

STRING

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke specific-offset. Contoh format nama file: mysql-bin.000003.

scan.startup.specific-offset.pos

Offset dalam file log biner yang ditentukan untuk offset awal saat menggunakan mode startup specific-offset.

Tidak

INTEGER

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke specific-offset.

scan.startup.specific-offset.gtid-set

Set GTID untuk offset awal saat menggunakan mode startup specific-offset.

Tidak

STRING

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke specific-offset. Contoh format set GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.

scan.startup.timestamp-millis

Timestamp dalam milidetik untuk offset awal saat menggunakan mode startup timestamp.

Tidak

LONG

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke timestamp. Satuan timestamp adalah milidetik.

server-time-zone

Zona waktu sesi yang digunakan oleh database.

Tidak

STRING

Jika Anda tidak menentukan parameter ini, sistem menggunakan zona waktu lingkungan runtime pekerjaan Flink sebagai zona waktu server database. Ini adalah zona waktu yang Anda pilih.

Contoh: Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP di MySQL dikonversi ke tipe STRING. Untuk informasi lebih lanjut, lihat Debezium temporal values.

scan.startup.specific-offset.skip-events

Jumlah event log biner yang dilewati saat membaca dari offset tertentu.

Tidak

INTEGER

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke specific-offset.

scan.startup.specific-offset.skip-rows

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

Tidak

INTEGER

None

Saat menggunakan parameter ini, Anda harus mengatur scan.startup.mode ke specific-offset.

connect.timeout

Waktu maksimum menunggu koneksi ke server database MySQL hingga timeout sebelum mencoba ulang.

Tidak

DURATION

30 s

None.

connect.max-retries

Jumlah maksimum percobaan ulang setelah koneksi ke layanan database MySQL gagal.

Tidak

INTEGER

3

None.

connection.pool.size

Ukuran kolam koneksi database.

Tidak

INTEGER

20

Kolam koneksi database digunakan untuk menggunakan kembali koneksi, yang dapat mengurangi jumlah koneksi database.

heartbeat.interval

Interval di mana sumber memajukan offset log biner menggunakan event heartbeat.

Tidak

DURATION

30s

Event heartbeat digunakan untuk memajukan offset log biner di sumber. Hal ini sangat berguna untuk tabel di MySQL yang jarang diperbarui. Untuk tabel seperti itu, offset log biner tidak dapat maju secara otomatis. Event heartbeat dapat mendorong offset log biner maju, yang mencegah masalah akibat offset log biner yang kedaluwarsa. Offset log biner yang kedaluwarsa dapat menyebabkan pekerjaan gagal dan tidak dapat dipulihkan, sehingga memerlukan restart tanpa status.

rds.region-id

ID wilayah instans Alibaba Cloud ApsaraDB RDS for MySQL.

Diperlukan saat menggunakan fitur untuk membaca log arsip dari OSS.

STRING

None

Untuk informasi lebih lanjut tentang ID wilayah, lihat Wilayah dan zona.

rds.access-key-id

ID AccessKey akun Alibaba Cloud ApsaraDB RDS for MySQL.

Diperlukan saat menggunakan fitur untuk membaca log arsip dari OSS.

STRING

None

Untuk informasi lebih lanjut, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey?

rds.access-key-secret

Rahasia AccessKey akun Alibaba Cloud ApsaraDB RDS for MySQL.

Diperlukan saat menggunakan fitur untuk membaca log arsip dari OSS.

STRING

None

Untuk informasi lebih lanjut, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey?

rds.db-instance-id

ID instans Alibaba Cloud ApsaraDB RDS for MySQL.

Diperlukan saat menggunakan fitur untuk membaca log arsip dari OSS.

STRING

None

None.

rds.main-db-id

Nomor database utama instans Alibaba Cloud ApsaraDB RDS for MySQL.

Tidak

STRING

None

Untuk informasi lebih lanjut tentang cara mendapatkan nomor database utama, lihat Cadangan log ApsaraDB RDS for MySQL.

rds.download.timeout

Periode timeout untuk mengunduh satu log arsip dari OSS.

Tidak

DURATION

60s

None.

rds.endpoint

Titik akhir layanan untuk mendapatkan informasi log biner OSS.

Tidak

STRING

None

Untuk informasi lebih lanjut tentang nilai yang valid, lihat Titik akhir.

rds.binlog-directory-prefix

Awalan direktori untuk menyimpan file log biner.

Tidak

STRING

rds-binlog-

None.

rds.use-intranet-link

Menentukan apakah akan menggunakan jaringan internal untuk mengunduh file log biner.

Tidak

BOOLEAN

true

None.

rds.binlog-directories-parent-path

Jalur mutlak direktori induk untuk menyimpan file log biner.

Tidak

STRING

None

None.

chunk-meta.group.size

Ukuran metadata chunk.

Tidak

INTEGER

1000

Jika metadata lebih besar dari nilai ini, metadata tersebut dibagi menjadi beberapa bagian untuk transmisi.

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

Batas bawah faktor distribusi chunk untuk sharding merata.

Tidak

DOUBLE

0.05

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

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

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

Batas atas faktor distribusi chunk untuk sharding merata.

Tidak

DOUBLE

1000.0

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

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

scan.incremental.close-idle-reader.enabled

Menentukan apakah akan menutup pembaca idle setelah snapshot selesai.

Tidak

BOOLEAN

false

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

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

Pada fase inkremental, menentukan apakah hanya akan mendeserialisasi event perubahan dari tabel yang ditentukan.

Tidak

BOOLEAN

• Nilai default adalah false di versi VVR 8.x.

• Nilai default adalah true di VVR 11.1 dan versi yang lebih baru.

Nilai yang valid:

• true: Hanya mendeserialisasi data perubahan dari tabel target untuk mempercepat pembacaan log biner.

• false (default): Mendeserialisasi data perubahan dari semua tabel.

scan.parallel-deserialize-changelog.enabled

Pada fase inkremental, menentukan apakah akan menggunakan beberapa thread untuk mengurai event perubahan.

Tidak

BOOLEAN

false

Nilai yang valid:

• true: Menggunakan beberapa thread pada fase deserialisasi event perubahan sambil mempertahankan urutan event log biner untuk mempercepat pembacaan.

• false (default): Menggunakan satu thread pada fase deserialisasi event.

scan.parallel-deserialize-changelog.handler.size

Jumlah penanganan event saat menggunakan beberapa thread untuk mengurai event perubahan.

Tidak

INTEGER

2

metadata-column.include-list

Kolom metadata yang akan diteruskan ke downstream.

Tidak

STRING

None

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

scan.newly-added-table.enabled

Saat me-restart dari checkpoint, menentukan apakah akan menyinkronkan tabel baru yang tidak cocok selama startup sebelumnya atau menghapus tabel dari state yang tidak lagi cocok.

Tidak

BOOLEAN

false

Ini berlaku saat me-restart dari checkpoint atau savepoint.

scan.binlog.newly-added-table.enabled

Pada fase inkremental, menentukan apakah akan mengirimkan data dari tabel baru yang cocok.

Tidak

BOOLEAN

false

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

scan.incremental.snapshot.chunk.key-column

Menentukan kolom untuk tabel tertentu yang akan digunakan sebagai kolom pemisah untuk sharding selama fase snapshot.

Tidak

STRING

None

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

• Wajib untuk tabel tanpa primary key. Kolom yang dipilih harus bertipe non-null (NOT NULL). Opsional untuk tabel dengan primary key. Hanya satu kolom yang dapat dipilih dari primary key.

scan.parse.online.schema.changes.enabled

Pada fase inkremental, menentukan apakah akan mencoba mengurai event DDL perubahan tanpa lock RDS.

Tidak

BOOLEAN

false

Nilai yang valid:

• true: Mengurai event DDL perubahan tanpa lock RDS.

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

Ini adalah fitur eksperimen. Sebelum melakukan perubahan tanpa lock online, ambil snapshot pekerjaan Flink untuk pemulihan.

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.

Backfill hanya berlaku selama kueri snapshot satu chunk dan tidak mencakup seluruh fase pembacaan penuh. Saat backfill dilewati, kueri snapshot setiap chunk membaca data tabel terbaru pada saat itu; pembaruan yang terjadi pada chunk setelah dibaca tidak digabungkan selama fase pembacaan penuh dan dibaca dari Binlog setelah memasuki fase inkremental. Misalnya, pembaruan pada chunk5 yang terjadi saat chunk5 sedang di-snapshot tercermin langsung dalam snapshot chunk5; jika chunk5 diperbarui setelah pembaca maju ke chunk80, pembaruan tersebut diterapkan kemudian dari Binlog selama fase inkremental.

treat-tinyint1-as-boolean.enabled

Menentukan apakah akan memperlakukan tipe TINYINT(1) sebagai tipe Boolean.

Tidak

BOOLEAN

true

Nilai yang valid:

• true (default): Memperlakukan tipe TINYINT(1) sebagai tipe Boolean.

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

treat-timestamp-as-datetime-enabled

Menentukan apakah akan memperlakukan tipe TIMESTAMP sebagai tipe DATETIME.

Tidak

BOOLEAN

false

Nilai yang valid:

• true: Memperlakukan tipe TIMESTAMP MySQL sebagai tipe DATETIME dan memetakannya ke tipe CDC TIMESTAMP.

• false (default): Memetakan tipe TIMESTAMP MySQL ke tipe CDC TIMESTAMP_LTZ.

Tipe TIMESTAMP MySQL menyimpan waktu UTC dan dipengaruhi oleh zona waktu. Tipe DATETIME MySQL menyimpan waktu literal dan tidak dipengaruhi oleh zona waktu.

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

include-comments.enabled

Menentukan apakah akan menyinkronkan komentar tabel dan kolom.

Tidak

BOOELEAN

false

Nilai yang valid:

• true: Menyinkronkan komentar tabel dan kolom.

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

Mengaktifkan opsi 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

BOOELEAN

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 eksperimen. Mengaktifkannya dapat mengurangi risiko error OOM pada Pengelola Tugas saat menyinkronkan chunk terakhir selama fase snapshot. Tambahkan parameter ini sebelum startup pertama pekerjaan.

binlog.session.network.timeout

Timeout jaringan untuk koneksi log biner.

Tidak

DURATION

10m

Jika diatur ke 0s, timeout default server MySQL yang digunakan.

scan.rate-limit.records-per-second

Membatasi jumlah maksimum catatan yang dikirim oleh sumber per detik.

Tidak

LONG

None

Ini berlaku untuk skenario di mana pembacaan data perlu dibatasi. Batasan ini efektif baik pada fase penuh maupun inkremental.

Metrik numRecordsOutPerSecond dari sumber mencerminkan jumlah catatan yang di-output oleh seluruh aliran data per detik. Anda dapat menyesuaikan parameter ini berdasarkan metrik tersebut.

Pada fase pembacaan data penuh, biasanya Anda perlu mengurangi jumlah baris yang dibaca dalam setiap batch. Anda dapat mengurangi nilai parameter scan.incremental.snapshot.chunk.size.

include-binlog-meta.enable

Menentukan apakah akan menyertakan informasi log biner MySQL asli, seperti GTID dan offset log biner, dalam pesan.

Tidak

Boolean

false

Ini berlaku untuk skenario sinkronisasi log biner asli, seperti mengganti tautan sinkronisasi Canal yang sudah ada.

scan.binlog.tolerate.gtid-holes

Mengaktifkan parameter ini mengabaikan celah dalam urutan GTID, memungkinkan pekerjaan melewati event yang tidak kontinu dan terus berjalan.

Tidak

Boolean

false

Sebelum mengaktifkan parameter ini, Anda harus memastikan bahwa offset awal pekerjaan belum kedaluwarsa. Jika pekerjaan dimulai dari offset GTID yang telah dihapus atau kedaluwarsa, mesin akan diam-diam melewati log yang hilang, yang akan menyebabkan kehilangan data.

scan.emit.create-table-events.in-batch.enabled

Menentukan apakah akan mengirimkan skema tabel secara batch selama fase inisialisasi pekerjaan.

Tidak

Boolean

false

Ini adalah fitur eksperimen. Aktifkan opsi ini saat satu pekerjaan menyinkronkan banyak tabel.

Tipe bidang CDC MySQL

Tipe bidang CDC

TINYINT(n)

TINYINT

SMALLINT

SMALLINT

TINYINT UNSIGNED

TINYINT UNSIGNED ZEROFILL

YEAR

INT

INT

MEDIUMINT

MEDIUMINT UNSIGNED

MEDIUMINT UNSIGNED ZEROFILL

SMALLINT UNSIGNED

SMALLINT UNSIGNED ZEROFILL

BIGINT

BIGINT

INT UNSIGNED

INT 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] dan p <= 38

DECIMAL(p, s)

DECIMAL(p, s) [UNSIGNED] [ZEROFILL] dan p <= 38

FIXED(p, s) [UNSIGNED] [ZEROFILL] dan p <= 38

BOOLEAN

BOOLEAN

BIT(1)

TINYINT(1)

DATE

DATE

TIME [(p)]

TIME [(p)]

DATETIME [(p)]

TIMESTAMP [(p)]

TIMESTAMP [(p)]

Pemetaan tergantung pada nilai parameter treat-timestamp-as-datetime-enabled:

true:TIMESTAMP[(p)]

false:TIMESTAMP_LTZ[(p)]

CHAR(n)

CHAR(n)

VARCHAR(n)

VARCHAR(n)

BIT(n)

BINARY(⌈(n + 7) / 8⌉)

BINARY(n)

BINARY(n)

VARBINARY(N)

VARBINARY(N)

NUMERIC(p, s) [UNSIGNED] [ZEROFILL] dan 38 < p <= 65

STRING

DECIMAL(p, s) [UNSIGNED] [ZEROFILL] dan 38 < p <= 65

FIXED(p, s) [UNSIGNED] [ZEROFILL] dan 38 < p <= 65

TINYTEXT

STRING

TEXT

MEDIUMTEXT

LONGTEXT

ENUM

JSON

STRING

GEOMETRY

STRING

POINT

LINESTRING

POLYGON

MULTIPOINT

MULTILINESTRING

MULTIPOLYGON

GEOMETRYCOLLECTION

TINYBLOB

BYTES

BLOB

MEDIUMBLOB

LONGBLOB