Realtime Compute for Apache Flink:MySQL SQL connector

SQL

Anda dapat menggunakan konektor MySQL dalam pekerjaan SQL sebagai tabel sumber, tabel dimensi, atau tabel sink.

Sintaksis

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

Parameter WITH

• Umum

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  connector	Jenis tabel.	Ya	STRING	None	Saat digunakan sebagai tabel sumber, Anda dapat mengatur parameter ini ke mysql-cdc atau mysql. Keduanya setara. Saat digunakan sebagai tabel dimensi atau tabel sink, nilainya harus mysql.
  hostname	Alamat IP atau hostname database MySQL.	Ya	STRING	None	Kami menyarankan agar Anda menentukan alamat virtual private cloud (VPC). Catatan Jika database MySQL dan Realtime Compute for Apache Flink tidak berada dalam VPC yang sama, Anda harus membuat koneksi jaringan cross-VPC atau menggunakan titik akhir publik untuk mengakses database. Untuk informasi selengkapnya, lihat Kelola dan operasikan ruang kerja dan Bagaimana kluster Flink yang sepenuhnya dikelola mengakses Internet?.
  username	Username untuk layanan database MySQL.	Ya	STRING	None	None.
  password	Password untuk layanan database MySQL.	Ya	STRING	None	None.
  database-name	Nama database MySQL.	Ya	STRING	None	Saat database digunakan sebagai tabel sumber, Anda dapat menggunakan ekspresi reguler untuk nama database guna membaca data dari beberapa database. Saat menggunakan ekspresi reguler, jangan gunakan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk informasi selengkapnya, lihat catatan untuk parameter table-name.
  table-name	Nama tabel MySQL.	Ya	STRING	None	Anda dapat menggunakan ekspresi reguler untuk nama tabel sumber guna membaca data dari beberapa tabel. Saat membaca data dari beberapa tabel MySQL, kirimkan beberapa pernyataan CTAS sebagai satu pekerjaan. Hal ini menghindari pengaktifan beberapa pendengar log biner dan meningkatkan kinerja serta efisiensi. Untuk informasi selengkapnya, lihat Beberapa pernyataan CTAS: Kirim sebagai satu pekerjaan. Saat menggunakan ekspresi reguler, jangan gunakan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk informasi selengkapnya, lihat catatan berikut. Catatan Saat tabel sumber CDC MySQL mencocokkan nama tabel menggunakan ekspresi reguler, konektor menggabungkan database-name dan table-name yang Anda tentukan dengan string \\. untuk membentuk ekspresi reguler path lengkap. Sebelum VVR 8.0.1, karakter . digunakan. Konektor kemudian menggunakan ekspresi reguler ini untuk mencocokkan nama tabel yang memenuhi syarat sepenuhnya di database MySQL. Misalnya, jika Anda mengatur 'database-name'='db_.*' dan 'table-name'='tb_.+', konektor menggunakan ekspresi reguler db_.*\\.tb_.+ untuk mencocokkan nama tabel yang memenuhi syarat sepenuhnya guna menentukan tabel mana yang akan dibaca. Sebelum VVR 8.0.1, ekspresi regulernya adalah db_.*.tb_.+.
  port	Nomor port layanan database MySQL.	Tidak	INTEGER	3306	None.

• Hanya untuk tabel sumber

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  server-id	ID numerik untuk klien database.	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. Saat pembacaan inkremental diaktifkan, pembacaan konkuren didukung. Dalam kasus ini, atur rentang ID sehingga setiap pembaca konkuren menggunakan ID berbeda. Untuk informasi selengkapnya, lihat Gunakan Server ID.
  scan.incremental.snapshot.enabled	Menentukan apakah akan mengaktifkan snapshot inkremental.	Tidak	BOOLEAN	true	Snapshot inkremental diaktifkan secara default. Snapshot inkremental adalah mekanisme baru untuk membaca snapshot data penuh. Dibandingkan dengan metode pembacaan snapshot lama, snapshot inkremental memiliki banyak keunggulan, termasuk: Sumber dapat membaca data penuh secara paralel. Sumber mendukung checkpoint tingkat chunk saat membaca data penuh. Sumber tidak perlu mengambil lock baca global (FLUSH TABLES WITH read lock) saat membaca data penuh. Jika Anda ingin sumber mendukung pembacaan konkuren, setiap pembaca konkuren memerlukan server ID unik. Oleh karena itu, server-id harus berupa rentang, seperti 5400-6400, dan ukuran rentang harus lebih besar dari atau sama dengan konkurensi. Catatan Item konfigurasi ini dihapus di Ververica Runtime (VVR) 11.1 dan versi yang lebih baru.
  scan.incremental.snapshot.chunk.size	Ukuran setiap chunk dalam jumlah baris.	Tidak	INTEGER	8096	Saat pembacaan snapshot inkremental diaktifkan, tabel dibagi menjadi beberapa chunk untuk dibaca. Data chunk di-cache dalam memori sebelum sepenuhnya dibaca. Semakin sedikit baris yang dikandung setiap chunk, semakin besar jumlah total chunk dalam tabel. Meskipun hal ini mengurangi granularitas pemulihan kesalahan, hal ini dapat menyebabkan error kehabisan memori (OOM) dan throughput keseluruhan yang lebih rendah. Oleh karena itu, Anda perlu membuat pertimbangan dan mengatur ukuran chunk yang wajar.
  scan.snapshot.fetch.size	Jumlah maksimum catatan yang ditarik sekaligus saat membaca data penuh tabel.	Tidak	INTEGER	1024	None.
  scan.startup.mode	Mode startup untuk konsumsi data.	Tidak	STRING	initial	Nilai yang valid: initial (default): Saat startup pertama atau startup tanpa status, konektor memindai data historis penuh lalu membaca data log biner terbaru. latest-offset: Saat startup pertama 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 yang tersedia paling awal. 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. Penting Saat menggunakan mode startup earliest-offset, specific-offset, atau timestamp, pastikan skema tabel yang sesuai tidak berubah antara posisi konsumsi log biner yang ditentukan dan waktu startup pekerjaan. Hal ini mencegah error akibat ketidakcocokan skema.
  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. Penting Saat Anda menentukan waktu, CDC MySQL mencoba membaca event awal setiap file log biner untuk menentukan timestamp-nya. Kemudian, CDC MySQL menemukan file log biner yang sesuai dengan waktu yang ditentukan. Pastikan file log biner yang sesuai dengan timestamp yang ditentukan belum dihapus dari database dan dapat dibaca.
  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 zona yang Anda pilih.	Contoh: Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP di MySQL dikonversi ke tipe STRING. Untuk informasi selengkapnya, lihat Debezium temporal values.
  debezium.min.row.count.to.stream.results	Saat jumlah baris dalam tabel lebih besar dari nilai ini, mode pembacaan batch digunakan.	Tidak	INTEGER	1000	Flink membaca data dari tabel sumber MySQL dengan salah satu cara berikut: Baca penuh: Membaca seluruh data tabel langsung ke memori. Metode ini cepat tetapi mengonsumsi memori yang sesuai. Jika tabel sumber sangat besar, ada risiko error OOM. Baca batch: Membaca data dalam beberapa batch, dengan jumlah baris tertentu per batch, hingga semua data dibaca. Metode ini menghindari risiko OOM saat membaca tabel besar tetapi relatif lambat.
  connect.timeout	Waktu maksimum menunggu koneksi ke server database MySQL hingga timeout sebelum mencoba lagi.	Tidak	DURATION	30s	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.
  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 selengkapnya 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. Peringatan Jangan mengubah parameter Debezium secara sembarangan. Hal ini dapat menyebabkan konektor membaca data secara salah. Misalnya, parameter debezium.binlog.buffer.size tidak boleh dikonfigurasi.
  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 tersebut, 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.
  scan.incremental.snapshot.chunk.key-column	Menentukan kolom yang akan digunakan sebagai kolom pemisah untuk sharding selama fase snapshot.	Lihat kolom Catatan.	STRING	None	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.
  rds.region-id	ID wilayah instans Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya tentang ID wilayah, lihat Wilayah dan zona. Penting Karena string GTID untuk CDC MySQL dihasilkan secara acak dan tidak meningkat secara monoton seperti offset file log biner, menemukan GTID dalam file memerlukan pengunduhan dan parsing semua log arsip dari OSS. Proses ini sangat intensif sumber daya dan memakan waktu, sehingga fitur yang bergantung pada offset GTID tidak layak. Oleh karena itu, fitur log arsip OSS hanya mendukung pemulaian dari timestamp tertentu atau offset file log biner tertentu. Fitur ini tidak mendukung pemulaian dari GTID tertentu, maupun skenario dengan alih bencana primer/sekunder dalam log arsip, karena alih bencana primer/sekunder MySQL bergantung pada GTID. Evaluasi fitur ini dengan hati-hati sebelum digunakan.
  rds.access-key-id	ID AccessKey akun Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey?. Penting Untuk mencegah Informasi AccessKey Anda bocor, gunakan fitur manajemen rahasia untuk menentukan ID AccessKey. Untuk informasi selengkapnya, lihat Kelola variabel.
  rds.access-key-secret	Rahasia AccessKey akun Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey? Penting Untuk mencegah Informasi AccessKey Anda bocor, gunakan fitur manajemen rahasia untuk menentukan Rahasia AccessKey. Untuk informasi selengkapnya, lihat Kelola variabel.
  rds.db-instance-id	ID instans Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat menggunakan fitur 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 selengkapnya tentang cara mendapatkan nomor database utama, lihat Cadangan log ApsaraDB RDS for MySQL. Hanya didukung di VVR 8.0.7 dan versi yang lebih baru. Catatan Jika parameter ini tidak ditentukan, VVR 11.7 dan versi yang lebih baru secara otomatis mengkueri nomor database utama berdasarkan informasi koneksi 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 selengkapnya tentang nilai yang valid, lihat Titik akhir. Hanya didukung di VVR 8.0.8 dan versi yang lebih baru.
  scan.incremental.close-idle-reader.enabled	Menentukan apakah akan menutup pembaca idle setelah snapshot selesai.	Tidak	BOOLEAN	false	Hanya didukung di VVR 8.0.1 dan versi yang lebih baru. Agar konfigurasi ini berlaku, Anda harus mengatur execution.checkpointing.checkpoints-after-tasks-finish.enabled ke true.
  scan.read-changelog-as-append-only.enabled	Menentukan apakah akan mengonversi aliran data changelog menjadi aliran data append-only.	Tidak	BOOLEAN	false	Nilai yang valid: true: Semua jenis pesan, termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER, dikonversi menjadi pesan INSERT. Aktifkan opsi ini hanya dalam skenario khusus, seperti saat Anda perlu menyimpan pesan delete dari tabel upstream. false (default): Semua jenis pesan dikirim downstream apa adanya. Catatan Hanya didukung di VVR 8.0.8 dan versi yang lebih baru.
  scan.only.deserialize.captured.tables.changelog.enabled	Dalam 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. Catatan Hanya didukung di VVR 8.0.7 dan versi yang lebih baru. Saat menggunakan VVR 8.0.8 atau versi sebelumnya, Anda harus mengubah nama parameter menjadi debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	Dalam 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 eksperimental. Sebelum melakukan perubahan tanpa lock online, ambil snapshot pekerjaan Flink untuk pemulihan. Catatan Hanya didukung di VVR 11.1 dan versi yang lebih baru.
  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 chunk tunggal dan tidak mencakup seluruh fase baca penuh. Saat backfill dilewati, kueri snapshot setiap chunk membaca data tabel terbaru pada saat itu; pembaruan yang terjadi pada chunk setelah dibaca tidak digabung selama fase baca 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. Penting Saat diaktifkan, perubahan yang terjadi selama atau setelah pemindaian chunk tetap dikirim dari Binlog dalam fase inkremental dan mungkin diduplikasi. Hanya semantik setidaknya-sekali yang dijamin. Aktifkan ini hanya saat sink downstream mendukung penulisan idempoten berdasarkan primary key. Catatan Hanya didukung di VVR 11.1 dan versi yang lebih baru.
  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 eksperimental. Mengaktifkannya dapat mengurangi risiko error OOM pada Pengelola Tugas saat menyinkronkan chunk terakhir selama fase snapshot. Tambahkan parameter ini sebelum startup pertama pekerjaan. Catatan Hanya didukung di VVR 11.1 dan versi yang lebih baru.
  binlog.session.network.timeout	Timeout baca/tulis jaringan untuk koneksi log biner.	Tidak	DURATION	10m	Jika diatur ke 0s, timeout default server MySQL digunakan. Catatan Hanya didukung di VVR 11.5 dan versi yang lebih baru.
  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 berlaku baik dalam fase penuh maupun inkremental. Metrik numRecordsOutPerSecond dari sumber mencerminkan jumlah catatan yang dikeluarkan oleh seluruh aliran data per detik. Anda dapat menyesuaikan parameter ini berdasarkan metrik tersebut. Dalam fase pembacaan data penuh, Anda biasanya perlu mengurangi jumlah baris yang dibaca dalam setiap batch. Anda dapat mengurangi nilai parameter scan.incremental.snapshot.chunk.size. Catatan Hanya didukung di VVR 11.5 dan versi yang lebih baru.
  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 dihapus atau kedaluwarsa, mesin akan diam-diam melewatkan log yang hilang, yang akan menyebabkan kehilangan data. Catatan Parameter ini hanya didukung di VVR 11.6 dan versi yang lebih baru.

• Parameter khusus tabel dimensi

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  url	URL JDBC MySQL.	Tidak	STRING	None	Format URL adalah: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	Jumlah maksimum percobaan ulang setelah pembacaan data gagal.	Tidak	INTEGER	3	Hanya didukung di VVR 6.0.7 dan versi yang lebih baru.
  lookup.cache.strategy	Kebijakan cache.	Tidak	STRING	None	Kebijakan cache yang didukung adalah None, LRU, dan ALL. Untuk informasi selengkapnya tentang nilai-nilai tersebut, lihat Pernyataan JOIN tabel dimensi. Catatan Saat Anda menggunakan kebijakan cache LRU, Anda juga harus mengonfigurasi parameter lookup.cache.max-rows.
  lookup.cache.max-rows	Jumlah maksimum baris yang di-cache.	Tidak	INTEGER	100000	Jika Anda memilih kebijakan cache LRU, Anda harus mengatur ukuran cache. Jika Anda memilih kebijakan cache ALL, Anda tidak perlu mengatur ukuran cache.
  lookup.cache.ttl	Waktu hidup (TTL) cache.	Tidak	DURATION	10 s	Konfigurasi lookup.cache.ttl bergantung pada lookup.cache.strategy: Jika lookup.cache.strategy diatur ke None, Anda tidak perlu mengonfigurasi lookup.cache.ttl. Artinya cache tidak kedaluwarsa. Jika lookup.cache.strategy diatur ke LRU, lookup.cache.ttl adalah TTL cache. Secara default, cache tidak kedaluwarsa. Jika lookup.cache.strategy diatur ke ALL, lookup.cache.ttl adalah waktu pemuatan cache. Secara default, cache tidak dimuat ulang. Gunakan format waktu, seperti 1min atau 10s.
  lookup.max-join-rows	Jumlah maksimum hasil yang dikembalikan saat catatan dari tabel utama cocok dengan catatan di tabel dimensi.	Tidak	INTEGER	1024	None.
  lookup.filter-push-down.enabled	Menentukan apakah akan mengaktifkan filter pushdown untuk tabel dimensi.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengaktifkan filter pushdown untuk tabel dimensi. Saat memuat data dari tabel database MySQL, tabel dimensi memfilter data terlebih dahulu berdasarkan kondisi yang ditetapkan dalam pekerjaan SQL. false (default): Menonaktifkan filter pushdown untuk tabel dimensi. Saat memuat data dari tabel database MySQL, tabel dimensi memuat semua data. Catatan Hanya didukung di VVR 8.0.7 dan versi yang lebih baru. Penting Pushdown tabel dimensi hanya boleh diaktifkan saat tabel Flink digunakan sebagai tabel dimensi. Tabel sumber MySQL tidak mendukung pengaktifan filter pushdown. Jika tabel Flink digunakan sebagai tabel sumber dan tabel dimensi, dan filter pushdown diaktifkan untuk tabel dimensi, Anda harus secara eksplisit mengatur item konfigurasi ini ke false untuk tabel sumber menggunakan Petunjuk SQL. Jika tidak, pekerjaan mungkin berjalan tidak normal.

• Hanya untuk tabel sink

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  url	URL JDBC MySQL.	Tidak	STRING	None	Format URL adalah: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Jumlah maksimum percobaan ulang setelah penulisan data gagal.	Tidak	INTEGER	3	None.
  sink.buffer-flush.batch-size	Jumlah baris dalam satu penulisan batch.	Tidak	INTEGER	4096	None.
  sink.buffer-flush.max-rows	Jumlah baris data yang di-cache dalam memori.	Tidak	INTEGER	10000	Parameter ini hanya berlaku setelah primary key ditentukan.
  sink.buffer-flush.interval	Interval untuk membersihkan cache. Jika data dalam cache tidak memenuhi kondisi output setelah waktu tunggu yang ditentukan, sistem secara otomatis mengeluarkan semua data dalam cache.	Tidak	DURATION	1s	None.
  sink.ignore-delete	Menentukan apakah akan mengabaikan operasi DELETE data.	Tidak	BOOLEAN	false	Saat aliran yang dihasilkan oleh Flink SQL menyertakan catatan delete atau update-before, jika beberapa tugas output memperbarui bidang berbeda dari tabel yang sama secara bersamaan, ketidaksesuaian data dapat terjadi. Misalnya, setelah catatan dihapus, tugas lain hanya memperbarui beberapa bidang. Bidang yang tidak diperbarui akan menjadi null atau nilai default, menyebabkan error data. Dengan mengatur sink.ignore-delete ke true, Anda dapat mengabaikan operasi DELETE dan UPDATE_BEFORE upstream untuk menghindari masalah tersebut. Catatan UPDATE_BEFORE adalah bagian dari mekanisme retraksi Flink, digunakan untuk "menarik kembali" nilai lama dalam operasi pembaruan. Saat ignoreDelete = true, semua catatan tipe DELETE dan UPDATE_BEFORE dilewati. Hanya catatan INSERT dan UPDATE_AFTER yang diproses.
  sink.ignore-delete-mode	Strategi untuk menangani catatan tipe delete setelah operasi DELETE diabaikan.	Tidak	STRING	ALL	Nilai yang valid: ALL: Mengabaikan catatan -D dan -U. REAL_DELETE: Hanya mengabaikan catatan -D. UPDATE_BEFORE: Hanya mengabaikan catatan -U. Catatan Opsi ini hanya didukung di mesin Realtime Compute VVR 11.8 dan versi yang lebih baru. Hanya berlaku saat sink.ignore-delete=true. Mengonfigurasinya sendiri menghasilkan error.
  sink.ignore-null-when-update	Saat memperbarui data, menentukan apakah akan memperbarui bidang yang sesuai menjadi null atau melewatkan pembaruan untuk bidang tersebut jika nilai bidang data yang masuk adalah null.	Tidak	BOOLEAN	false	Nilai yang valid: true: Tidak memperbarui bidang. Parameter ini dapat diatur ke true hanya saat primary key ditetapkan untuk tabel Flink. Saat diatur ke true: Untuk VVR 8.0.6 dan versi sebelumnya, tabel sink tidak mendukung penulisan batch. Untuk VVR 8.0.7 dan versi yang lebih baru, tabel sink mendukung penulisan batch. Penulisan batch dapat secara signifikan meningkatkan efisiensi penulisan dan throughput keseluruhan, tetapi memperkenalkan latensi data dan risiko error OOM. Oleh karena itu, Anda harus membuat pertimbangan berdasarkan skenario bisnis Anda. false: Memperbarui bidang menjadi null. Catatan Parameter ini hanya didukung di VVR 8.0.5 dan versi yang lebih baru.

Pemetaan tipe

• Tabel sumber CDC

  Tipe bidang MySQL CDC	Tipe bidang 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 [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  Penting

  Jangan gunakan tipe TINYINT(1) di MySQL untuk menyimpan nilai selain 0 dan 1. Saat property-version=0, tabel sumber CDC MySQL memetakan TINYINT(1) ke tipe BOOLEAN di Flink secara default. Hal ini dapat menyebabkan ketidakakuratan data. Untuk menggunakan tipe TINYINT(1) guna menyimpan nilai selain 0 dan 1, lihat parameter konfigurasi catalog.table.treat-tinyint1-as-boolean.

• Tabel dimensi dan tabel sink

  Tipe bidang MySQL	Tipe bidang Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) Catatan dengan p <= 38.
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES Penting Flink hanya mendukung catatan tipe BLOB MySQL yang kurang dari atau sama dengan 2.147.483.647 (2^31 - 1) byte.
  BLOB
  MEDIUMBLOB
  LONGBLOB

Diagnosis latensi data untuk mengoptimalkan throughput pekerjaan

Jika Anda mengalami latensi data selama fase inkremental, analisis masalah dengan mengikuti langkah-langkah berikut:

1. Periksa metrik currentFetchEventTimeLag dan currentEmitEventTimeLag di halaman Ikhtisar. Metrik currentFetchEventTimeLag merepresentasikan latensi dalam membaca data dari log biner. Metrik currentEmitEventTimeLag merepresentasikan latensi dalam membaca data untuk tabel yang relevan dengan pekerjaan dari log biner.

   Skenario	Deskripsi
   currentFetchEventTimeLag rendah, sedangkan currentEmitEventTimeLag tinggi dan jarang diperbarui.	currentFetchEventTimeLag yang rendah menunjukkan bahwa menarik log biner dari database efisien. Namun, log biner berisi sedikit data untuk tabel yang perlu dibaca pekerjaan. Oleh karena itu, currentEmitEventTimeLag jarang diperbarui. Ini adalah perilaku yang diharapkan.
   Kedua currentFetchEventTimeLag dan currentEmitEventTimeLag tinggi.	Hal ini menunjukkan bahwa tabel sumber memiliki kinerja baca yang buruk. Anda dapat melanjutkan ke langkah-langkah berikutnya dalam bagian ini untuk optimasi.

2. Backpressure dapat mengurangi laju pengiriman data sumber ke operator downstream. Anda mungkin mengamati bahwa sourceIdleTime meningkat secara berkala, dan kedua currentFetchEventTimeLag serta currentEmitEventTimeLag terus bertambah. Untuk mengatasi hal ini, tingkatkan parallelisme node tempat backpressure berasal.

3. Periksa metrik TM CPU Usage di halaman CPU dan metrik TM GC Time di halaman JVM untuk menentukan apakah sumber daya CPU atau memori tidak mencukupi. Anda dapat meningkatkan sumber daya pekerjaan untuk mengoptimalkan kinerja baca. Anda juga dapat mengaktifkan parameter mini-batch untuk meningkatkan throughput. Untuk informasi selengkapnya, lihat Teknik Optimasi SQL Flink Berkinerja Tinggi.

4. Jika operator SinkUpsertMaterializer dengan state besar ada dalam pekerjaan, hal ini dapat memengaruhi kinerja baca. Pertimbangkan untuk meningkatkan parallelisme pekerjaan atau menghindari operator SinkUpsertMaterializer. Untuk informasi selengkapnya, lihat Hindari penggunaan SinkUpsertMaterializer. Menghapus operator SinkUpsertMaterializer dari pekerjaan yang sudah ada memerlukan restart tanpa status. Hal ini karena topologi pekerjaan berubah, dan memulai dari status yang ada dapat menyebabkan pekerjaan gagal atau mengakibatkan kehilangan data.

Optimasi parameter chunk

Data dan metadata chunk disimpan dalam memori, yang kadang-kadang dapat menyebabkan error kehabisan memori (OOM). Anda dapat menyesuaikan parameter berdasarkan komponen mana yang mengalami error OOM:

• JobManager

  JobManager menyimpan metadata untuk semua chunk. Jumlah chunk yang berlebihan dapat menyebabkan error OOM. Untuk mengatasi hal ini, tingkatkan nilai scan.incremental.snapshot.chunk.size untuk mengurangi jumlah chunk. Atau, Anda dapat meningkatkan memori heap JobManager dengan mengatur jobmanager.memory.heap.size dalam konfigurasi runtime Anda. Untuk informasi selengkapnya, lihat Konfigurasi Parameter Flink.

• TaskManager

  • TaskManager membaca data untuk setiap chunk. Jika chunk berisi terlalu banyak baris, error OOM dapat terjadi. Untuk mengatasi hal ini, kurangi nilai scan.incremental.snapshot.chunk.size untuk mengurangi jumlah baris per chunk. Anda juga dapat meningkatkan memori heap TaskManager dengan meningkatkan nilai TaskManager Memory dalam konfigurasi runtime Anda.

  • Di VVR 8.0.8 dan versi sebelumnya, chunk terakhir mungkin berisi sejumlah besar data, yang dapat menyebabkan TaskManager mengalami error OOM. Kami menyarankan untuk meningkatkan ke VVR 8.0.9 atau versi yang lebih baru untuk menghindari masalah ini.

  • Untuk tabel sumber CDC MySQL dengan primary key komposit, chunk dibagi berdasarkan kolom pertama kunci secara default. Jika data sangat miring, dengan banyak baris yang memiliki nilai yang sama dalam kolom tersebut, chunk untuk nilai tersebut dapat menjadi sangat besar dan menyebabkan error OOM di TaskManager. Anda dapat mengatur scan.incremental.snapshot.chunk.key-column untuk menentukan kolom berbeda dari primary key untuk pemisahan.