Realtime Compute for Apache Flink: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	Tipe 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 selengkapnya, lihat Kelola dan operasikan ruang kerja dan Bagaimana kluster Flink yang sepenuhnya dikelola mengakses Internet?.
  username	Nama pengguna untuk layanan database MySQL.	Ya	STRING	Tidak ada	Tidak ada.
  password	Kata sandi 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 selengkapnya, 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 Anda membaca data dari beberapa tabel MySQL, kirim beberapa pernyataan CTAS sebagai satu Pekerjaan. Hal ini menghindari pengaktifan beberapa Pendengar log biner dan meningkatkan performa dan 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 jalur 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	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 yang 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 hal ini, tetapkan rentang ID sehingga setiap pembaca konkuren menggunakan ID yang 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 memperoleh 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 yang unik. Oleh karena itu, server-id harus berupa rentang, seperti 5400-6400, dan ukuran rentangnya harus lebih besar dari atau sama dengan tingkat konkurensi. Catatan Item konfigurasi ini dihapus di Ververica Runtime (VVR) 11.1 dan versi selanjutnya.
  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 menurunkan throughput keseluruhan. 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	Tidak ada.
  scan.startup.mode	Mode startup untuk konsumsi data.	Tidak	STRING	initial	Nilai yang valid: initial (default): Saat startup pertama, konektor memindai data historis penuh lalu membaca data log biner terbaru. latest-offset: Saat startup pertama, 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	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 selengkapnya, lihat Nilai temporal Debezium.
  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	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 selengkapnya tentang parameter koneksi yang didukung, lihat Properti Konfigurasi MySQL.
  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 penguraian. 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 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	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 untuk membaca log arsip dari OSS.	STRING	Tidak ada	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 penguraian 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 untuk membaca log arsip dari OSS.	STRING	Tidak ada	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 untuk membaca log arsip dari OSS.	STRING	Tidak ada	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 untuk membaca log arsip dari OSS.	STRING	Tidak ada	Tidak ada.
  rds.main-db-id	Nomor database primer instans Alibaba Cloud ApsaraDB RDS for MySQL.	Tidak	STRING	Tidak ada	Untuk informasi selengkapnya tentang cara mendapatkan nomor database primer, lihat Cadangan log ApsaraDB RDS for MySQL. Didukung hanya di VVR 8.0.7 dan versi selanjutnya. Catatan Jika parameter ini tidak ditentukan, VVR 11.7 dan versi selanjutnya secara otomatis mengkueri nomor database primer 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 selengkapnya tentang nilai yang valid, lihat Titik akhir. Didukung hanya di VVR 8.0.8 dan versi selanjutnya.
  scan.incremental.close-idle-reader.enabled	Menentukan apakah akan menutup pembaca idle setelah snapshot selesai.	Tidak	BOOLEAN	false	Didukung hanya di VVR 8.0.1 dan versi selanjutnya. 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 Didukung hanya di VVR 8.0.8 dan versi selanjutnya.
  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 selanjutnya.	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 Didukung hanya di VVR 8.0.7 dan versi selanjutnya. Saat Anda 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	Pada fase inkremental, menentukan apakah akan mencoba mengurai event DDL perubahan tanpa penguncian RDS.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengurai event DDL perubahan tanpa penguncian RDS. false (default): Tidak mengurai event DDL perubahan tanpa penguncian RDS. Ini adalah fitur eksperimental. Sebelum melakukan perubahan tanpa penguncian online, ambil snapshot pekerjaan Flink untuk pemulihan. Catatan Didukung hanya di VVR 11.1 dan versi selanjutnya.
  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 selama fase snapshot dibaca dalam fase inkremental berikutnya, bukan digabungkan ke dalam snapshot. Penting Melewati backfill dapat menyebabkan ketidakkonsistenan data karena perubahan yang terjadi selama fase snapshot mungkin diputar ulang. Hanya semantik at-least-once yang dijamin. Catatan Didukung hanya di VVR 11.1 dan versi selanjutnya.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Menentukan apakah akan mengirim chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.	Tidak	BOOELEAN	false	Nilai yang valid: true: Mengirim chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot. false (default): Tidak mengirim 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 Didukung hanya di VVR 11.1 dan versi selanjutnya.
  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 Didukung hanya di VVR 11.5 dan versi selanjutnya.
  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 Didukung hanya di VVR 11.5 dan versi selanjutnya.
  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 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 didukung hanya di VVR 11.6 dan versi selanjutnya.

• 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	Didukung hanya di VVR 6.0.7 dan versi selanjutnya.
  lookup.cache.strategy	Kebijakan cache.	Tidak	STRING	Tidak ada	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 primer cocok dengan catatan di tabel dimensi.	Tidak	INTEGER	1024	Tidak ada.
  lookup.filter-push-down.enabled	Menentukan apakah akan mengaktifkan pushdown filter untuk tabel dimensi.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengaktifkan pushdown filter 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 pushdown filter untuk tabel dimensi. Saat memuat data dari tabel database MySQL, tabel dimensi memuat semua data. Catatan Didukung hanya di VVR 8.0.7 dan versi selanjutnya. Penting Pushdown tabel dimensi hanya boleh diaktifkan saat tabel Flink digunakan sebagai tabel dimensi. Tabel sumber MySQL tidak mendukung pengaktifan pushdown filter. Jika tabel Flink digunakan sebagai tabel sumber dan tabel dimensi, dan pushdown filter 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	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 menyertakan catatan hapus atau update-before, jika beberapa tugas output memperbarui bidang berbeda dari tabel yang sama secara bersamaan, ketidakkonsistenan 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 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-null-when-update	Saat memperbarui data, menentukan apakah akan memperbarui bidang yang sesuai menjadi null atau melewati 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 selanjutnya, tabel sink mendukung penulisan batch. Penulisan batch dapat meningkatkan efisiensi penulisan dan throughput keseluruhan secara signifikan, 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 didukung hanya di VVR 8.0.5 dan versi selanjutnya.

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	Byte Penting Flink hanya mendukung catatan tipe BLOB MySQL yang ukurannya kurang dari atau sama dengan 2.147.483.647 (2^31 - 1) byte.
  BLOB
  MEDIUMBLOB
  LONGBLOB

Ingesti Data

Anda dapat menggunakan konektor MySQL sebagai sumber data dalam pekerjaan YAML ingesti data.

Sintaks

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: xxx

Item konfigurasi

Gunakan katalog yang sudah ada

Mulai dari VVR 11.5, Anda dapat langsung mereferensikan katalog MySQL bawaan yang dibuat di halaman Data Management dalam pekerjaan ingesti data CDC Flink. Hal ini mengurangi upaya manual dalam menulis properti koneksi.

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog

Saat ini, pekerjaan ingesti data mendukung penggunaan ulang otomatis parameter katalog MySQL berikut:

• hostname

• port

• username

• password

• catalog.table.metadata-columns

• catalog.table.treat-tinyint1-as-boolean

Jika Anda ingin mengganti salah satu parameter yang digunakan ulang secara otomatis ini, Anda dapat menulis eksplisit parameter YAML yang sesuai. Parameter yang ditulis eksplisit memiliki prioritas lebih tinggi.

Pemetaan tipe

Tabel berikut menunjukkan pemetaan tipe data untuk ingesti data.