Realtime Compute for Apache Flink:Konektor SQL MySQL

SQL

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

Sintaks

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	Tidak ada	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	Tidak ada	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 lebih lanjut, lihat Kelola dan operasikan ruang kerja dan Bagaimana kluster Flink yang sepenuhnya dikelola mengakses Internet?.
  username	Username untuk layanan database MySQL.	Ya	STRING	Tidak ada	Tidak ada.
  password	Password untuk layanan database MySQL.	Ya	STRING	Tidak ada	Tidak ada.
  database-name	Nama database MySQL.	Ya	STRING	Tidak ada	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 lebih lanjut, lihat catatan untuk parameter table-name.
  table-name	Nama tabel MySQL.	Ya	STRING	Tidak ada	Anda dapat menggunakan ekspresi reguler untuk nama tabel sumber guna membaca data dari beberapa tabel. Saat menggunakan ekspresi reguler, jangan gunakan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk informasi lebih lanjut, 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 jalur lengkap. Sebelum VVR 8.0.1, karakter . digunakan. Konektor kemudian menggunakan ekspresi reguler ini untuk mencocokkan nama tabel yang memenuhi syarat penuh 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 penuh 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	Tidak ada.

• 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, tetapkan rentang ID sehingga setiap pembaca konkuren menggunakan ID berbeda. Untuk informasi lebih lanjut, 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 kunci 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 menetapkan ukuran chunk yang wajar.
  scan.snapshot.fetch.size	Jumlah maksimum catatan yang ditarik sekaligus saat membaca data penuh tabel.	Tidak	INTEGER	1024	Tidak ada.
  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 yang disebabkan oleh ketidakcocokan skema.
  scan.startup.specific-offset.file	Nama file log biner untuk offset awal saat menggunakan mode startup specific-offset.	Tidak	STRING	Tidak ada	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	Tidak ada	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	Tidak ada	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	Tidak ada	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 lebih lanjut, 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: Pembacaan 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. Pembacaan 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	Tidak ada.
  connect.max-retries	Jumlah maksimum percobaan ulang setelah koneksi ke layanan database MySQL gagal.	Tidak	INTEGER	3	Tidak ada.
  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	Tidak ada	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	Tidak ada	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 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 yang disebabkan oleh 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	Tidak ada	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	Tidak ada	Untuk informasi lebih lanjut 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	Tidak ada	Untuk informasi lebih lanjut, 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 lebih lanjut, 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	Tidak ada	Untuk informasi lebih lanjut, 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 lebih lanjut, 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	Tidak ada	Tidak ada.
  rds.main-db-id	Nomor database utama instans Alibaba Cloud ApsaraDB RDS for MySQL.	Tidak	STRING	Tidak ada	Untuk informasi lebih lanjut 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	Tidak ada.
  rds.endpoint	Titik akhir layanan untuk mendapatkan informasi log biner OSS.	Tidak	STRING	Tidak ada	Untuk informasi lebih lanjut 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 menganggur 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 hapus dari tabel hulu. false (default): Semua jenis pesan dikirim ke hilir sebagaimana 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 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 tabel target untuk mempercepat pembacaan log biner. false (default): Mendeserialisasi data perubahan 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 kunci RDS.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengurai event DDL perubahan tanpa kunci RDS. false (default): Tidak mengurai event DDL perubahan tanpa kunci RDS. Ini adalah fitur eksperimental. Sebelum melakukan perubahan tanpa kunci 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 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 nanti 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 hilir 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	Tidak ada	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 berurutan dan terus berjalan.	Tidak	BOOLEAN	false	Sebelum mengaktifkan parameter ini, Anda harus memastikan bahwa offset startup 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.
  scan.filter-unchanged-row.enabled	Menentukan apakah akan menyaring event perubahan bidang yang tidak digunakan dalam SQL.	Tidak	BOOLEAN	false	Jika Anda mengatur parameter ini ke true, tabel sumber CDC hanya membandingkan kolom yang dirujuk dalam pernyataan SQL saat menerima event UPDATE. Jika nilai kolom yang dirujuk tidak berubah, event tersebut tidak dikirim ke hilir. Parameter ini berlaku dalam skenario JOIN multi-aliran. Jika tabel hulu memperbarui bidang yang tidak dirujuk dalam pernyataan SQL, event DELETE dan INSERT yang tidak perlu tidak dihasilkan. Hal ini mengurangi amplifikasi data dan jitter hasil pada operator hilir. Catatan Parameter ini hanya didukung di VVR 11.8 dan versi yang lebih baru.
  scan.projection-pushdown.enabled	Menentukan apakah akan mengaktifkan projection pushdown.	Tidak	BOOLEAN	false	Jika Anda mengatur parameter ini ke true, tabel sumber CDC hanya mengurai dan mentransmisikan kolom yang dirujuk dalam pernyataan SQL. Kolom yang tidak dirujuk tidak diserialisasi atau ditransmisikan melalui jaringan. Hal ini mengurangi overhead I/O dan komputasi. Parameter ini bekerja lebih baik jika digunakan bersama dengan scan.filter-unchanged-row.enabled. Projection pushdown pertama-tama mengurangi volume data, lalu baris yang tidak berubah disaring untuk lebih meringankan beban pemrosesan pada operator hilir. Catatan Parameter ini hanya didukung di VVR 11.8 dan versi yang lebih baru.

• Parameter khusus tabel dimensi

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  url	URL JDBC MySQL.	Tidak	STRING	Tidak ada	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	Tidak ada	Kebijakan cache yang didukung adalah None, LRU, dan ALL. Untuk informasi lebih lanjut 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	Tidak ada.
  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 menyaring 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.
  lookup.cache.cache-empty	Menentukan apakah akan menyimpan hasil kueri kosong dalam cache.	Tidak	BOOLEAN	true	Hanya berlaku saat kebijakan cache adalah LRU. Nilai yang valid: true (default): jika kueri tidak mengembalikan data, hasil kosong di-cache dan dikembalikan langsung untuk kueri berikutnya. false: jika kueri tidak mengembalikan data, tidak ada hasil yang di-cache dan kueri berikutnya mencari data lagi. Catatan Parameter ini hanya didukung di VVR 11.9.0 dan versi yang lebih baru.

• Hanya untuk tabel sink

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  url	URL JDBC MySQL.	Tidak	STRING	Tidak ada	Format URL adalah: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Jumlah maksimum percobaan ulang setelah penulisan data gagal.	Tidak	INTEGER	3	Tidak ada.
  sink.buffer-flush.batch-size	Jumlah baris dalam satu penulisan batch.	Tidak	INTEGER	4096	Tidak ada.
  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	Tidak ada.
  sink.ignore-delete	Menentukan apakah akan mengabaikan operasi DELETE data.	Tidak	BOOLEAN	false	Saat aliran yang dihasilkan oleh Flink SQL mencakup 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 hulu untuk menghindari masalah tersebut. Catatan UPDATE_BEFORE adalah bagian dari mekanisme penarikan 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 hapus 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 hanya dapat diatur ke true 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.
  sink.force-batch-on-non-primary-table	Menentukan apakah akan memaksa buffering catatan dan memasukkannya secara batch menggunakan JDBC Batch saat data ditulis ke tabel sink MySQL yang tidak memiliki primary key.	Tidak	BOOLEAN	false	Parameter ini hanya berlaku pada tabel yang tidak memiliki primary key. Jika Anda mengatur parameter ini ke true, event DELETE dan UPDATE_BEFORE diam-diam dibuang, dan event INSERT dan UPDATE_AFTER ditulis secara batch sebagai pernyataan INSERT. Kami menyarankan agar Anda mengaktifkan parameter ini hanya jika datanya append-only atau semantik sebelumnya dapat diterima oleh Anda. Catatan Parameter ini hanya didukung di VVR 11.8 dan versi yang lebih baru. Aktifkan parameter ini hanya jika datanya append-only atau semantik sebelumnya dapat diterima oleh Anda. Jika tidak, data mungkin menjadi tidak konsisten.

Pemetaan tipe

• Tabel sumber CDC

  Tipe bidang CDC MySQL	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 di mana 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 mewakili latensi dalam membaca data dari log biner. Metrik currentEmitEventTimeLag mewakili 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 oleh 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 selanjutnya dalam bagian ini untuk optimasi.

2. Tekanan balik dapat mengurangi laju sumber mengirim data ke operator hilir. Anda mungkin mengamati bahwa sourceIdleTime meningkat secara berkala, dan kedua currentFetchEventTimeLag dan currentEmitEventTimeLag terus bertambah. Untuk mengatasi hal ini, tingkatkan parallelisme node tempat tekanan balik 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 lebih lanjut, 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 lebih lanjut, lihat Hindari penggunaan SinkUpsertMaterializer. Menghapus operator SinkUpsertMaterializer dari pekerjaan yang ada memerlukan restart tanpa status. Hal ini karena topologi pekerjaan berubah, dan memulai dari state yang ada dapat menyebabkan pekerjaan gagal atau 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 lebih lanjut, 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 Memori TaskManager 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 gabungan, 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.