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	Keterangan
  connector	Jenis tabel.	Ya	STRING	None	Saat digunakan sebagai tabel sumber, atur opsi ini ke mysql-cdc atau mysql. Keduanya setara. Saat digunakan sebagai tabel dimensi atau sink, atur opsi ini ke mysql.
  hostname	Alamat IP atau hostname database MySQL.	Ya	STRING	None	Kami menyarankan memasukkan alamat 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 jaringan publik untuk akses. Untuk informasi selengkapnya, lihat Kelola dan operasikan ruang kerja dan Bagaimana kluster Flink yang sepenuhnya dikelola mengakses jaringan publik?.
  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 Anda menggunakan database sebagai tabel sumber, Anda dapat menggunakan ekspresi reguler untuk nama database guna membaca data dari beberapa database. Saat menggunakan ekspresi reguler, hindari penggunaan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk informasi selengkapnya, lihat kolom Keterangan untuk opsi 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 Anda membaca data dari beberapa tabel MySQL, kirimkan beberapa pernyataan CTAS sebagai satu Pekerjaan. Ini menghindari pengaktifan beberapa Pendengar log biner dan meningkatkan kinerja dan efisiensi. Untuk informasi selengkapnya, lihat Beberapa pernyataan CTAS: Kirim sebagai satu Pekerjaan. Saat menggunakan ekspresi reguler, hindari penggunaan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk informasi selengkapnya, lihat catatan berikut. Catatan Saat tabel sumber CDC MySQL mencocokkan nama tabel, sistem menggabungkan database-name dan table-name yang Anda tentukan menjadi ekspresi reguler jalur lengkap menggunakan string \\. (karakter . digunakan dalam versi VVR sebelum 8.0.1). Sistem kemudian menggunakan ekspresi reguler ini untuk mencocokkan nama tabel lengkap di database MySQL. Misalnya, jika Anda mengatur 'database-name'='db_.*' dan 'table-name'='tb_.+', konektor menggunakan ekspresi reguler db_.*\\.tb_.+ (db_.*.tb_.+ dalam versi sebelum 8.0.1) untuk mencocokkan nama tabel lengkap guna menentukan tabel mana yang akan dibaca.
  port	Nomor port layanan database MySQL.	Tidak	INTEGER	3306	None.

• Khusus sumber

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Keterangan
  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. Kami merekomendasikan agar Anda menetapkan ID berbeda untuk setiap pekerjaan yang terhubung ke database yang sama. Opsi ini juga mendukung rentang ID, seperti 5400-5408. Saat pembacaan inkremental diaktifkan, pembacaan konkuren didukung. Dalam hal ini, kami merekomendasikan agar Anda menetapkan rentang ID sehingga setiap pembaca konkuren menggunakan ID berbeda. Untuk informasi selengkapnya, lihat Penggunaan Server ID.
  scan.incremental.snapshot.enabled	Menentukan apakah snapshot inkremental diaktifkan.	Tidak	BOOLEAN	true	Snapshot inkremental diaktifkan secara default. Snapshot inkremental adalah mekanisme baru untuk membaca snapshot data lengkap. Dibandingkan dengan metode pembacaan snapshot lama, snapshot inkremental menawarkan beberapa keunggulan: Sumber dapat membaca data lengkap secara paralel. Sumber mendukung checkpoint tingkat chunk saat membaca data lengkap. Sumber tidak perlu mengambil kunci baca global (FLUSH TABLES WITH read lock) saat membaca data lengkap. 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 rentang tersebut harus lebih besar dari atau sama dengan tingkat paralelisme. 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 disimpan dalam memori sebelum sepenuhnya dibaca. Semakin sedikit baris dalam chunk, semakin banyak jumlah chunk total dalam tabel. Meskipun ini mengurangi granularitas pemulihan kesalahan, hal ini dapat menyebabkan masalah kehabisan memori (OOM) dan mengurangi throughput keseluruhan. Oleh karena itu, Anda perlu mempertimbangkan trade-off dan menetapkan ukuran chunk yang wajar.
  scan.snapshot.fetch.size	Jumlah maksimum catatan yang diambil sekaligus saat membaca data lengkap tabel.	Tidak	INTEGER	1024	None.
  scan.startup.mode	Mode startup untuk konsumsi data.	Tidak	STRING	initial	Nilai yang valid: initial (Default): Memindai data historis lengkap terlebih dahulu, lalu membaca data binlog terbaru saat startup pertama kali. latest-offset: Tidak memindai data historis saat startup pertama kali. Mulai membaca dari akhir binlog, artinya hanya membaca perubahan terbaru setelah konektor dimulai. earliest-offset: Tidak memindai data historis. Mulai membaca dari posisi binlog paling awal yang tersedia. specific-offset: Tidak memindai data historis. Mulai dari offset binlog tertentu yang Anda tentukan. 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: Tidak memindai data historis. Mulai membaca binlog dari timestamp tertentu. Timestamp ditentukan oleh scan.startup.timestamp-millis dalam milidetik. Penting Saat Anda menggunakan mode startup earliest-offset, specific-offset, atau timestamp, pastikan skema tabel yang sesuai tidak berubah antara posisi konsumsi binlog yang ditentukan dan waktu startup pekerjaan untuk menghindari error akibat ketidakcocokan skema.
  scan.startup.specific-offset.file	Nama file binlog dari offset awal saat menggunakan mode startup offset tertentu.	Tidak	STRING	None	Saat menggunakan konfigurasi ini, Anda harus mengatur scan.startup.mode ke specific-offset. Format nama file contohnya mysql-bin.000003.
  scan.startup.specific-offset.pos	Offset dalam file binlog tertentu untuk memulai saat menggunakan mode startup offset tertentu.	Tidak	INTEGER	None	Saat menggunakan konfigurasi ini, Anda harus mengatur scan.startup.mode ke specific-offset.
  scan.startup.specific-offset.gtid-set	Set GTID dari offset awal saat menggunakan mode startup offset tertentu.	Tidak	STRING	None	Saat menggunakan konfigurasi ini, Anda harus mengatur scan.startup.mode ke specific-offset. Format set GTID contohnya 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	Offset awal sebagai timestamp milidetik saat menggunakan mode startup timestamp.	Tidak	LONG	None	Saat menggunakan konfigurasi ini, Anda harus mengatur scan.startup.mode ke timestamp. Timestamp dalam milidetik. Penting Saat Anda menentukan waktu, CDC MySQL mencoba membaca event awal setiap file binlog untuk menentukan timestamp-nya dan menemukan file binlog yang sesuai dengan waktu yang ditentukan. Pastikan file binlog untuk 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 opsi ini, sistem menggunakan zona waktu lingkungan runtime pekerjaan Flink sebagai zona waktu server database. Ini adalah zona waktu wilayah yang Anda pilih.	Contoh: Asia/Shanghai. Opsi 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 melebihi nilai ini, mode pembacaan batch digunakan.	Tidak	INTEGER	1000	Flink membaca data dari tabel sumber MySQL dengan cara berikut: Pembacaan lengkap: Membaca seluruh data tabel langsung ke memori. Metode ini cepat tetapi mengonsumsi memori yang sesuai. Jika tabel sumber sangat besar, hal ini dapat menimbulkan risiko OOM. Pembacaan batch: Membaca data dalam beberapa batch, dengan jumlah baris tertentu per batch, hingga semua data terbaca. Metode ini menghindari risiko OOM saat membaca tabel besar tetapi relatif lebih 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 ulang koneksi, yang dapat mengurangi jumlah koneksi database.
  jdbc.properties.*	Opsi koneksi kustom dalam URL JDBC.	Tidak	STRING	None	Anda dapat meneruskan opsi koneksi kustom. Misalnya, untuk tidak menggunakan protokol SSL, Anda dapat mengatur 'jdbc.properties.useSSL' = 'false'. Untuk informasi selengkapnya tentang opsi koneksi yang didukung, lihat MySQL Configuration Properties.
  debezium.*	Opsi kustom untuk Debezium guna membaca binlog.	Tidak	STRING	None	Anda dapat meneruskan opsi Debezium kustom. Misalnya, gunakan 'debezium.event.deserialization.failure.handling.mode'='ignore' untuk menentukan logika penanganan kesalahan penguraian.
  heartbeat.interval	Interval sumber menggunakan event heartbeat untuk memajukan offset binlog.	Tidak	DURATION	30s	Event heartbeat digunakan untuk memajukan offset binlog di sumber, yang sangat berguna untuk tabel MySQL yang diperbarui secara perlahan. Untuk tabel tersebut, offset binlog tidak maju secara otomatis. Event heartbeat dapat mendorong offset binlog maju, mencegah masalah di mana offset binlog kedaluwarsa menyebabkan pekerjaan gagal dan memerlukan restart tanpa status.
  scan.incremental.snapshot.chunk.key-column	Kolom yang digunakan untuk membagi chunk selama fase snapshot.	Lihat Keterangan.	STRING	None	Wajib untuk tabel tanpa primary key. Kolom yang dipilih harus bertipe non-null (NOT NULL). Opsional untuk tabel dengan primary key. Anda hanya dapat memilih satu kolom dari primary key.
  rds.region-id	ID wilayah instans ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya tentang ID wilayah, lihat Wilayah dan zona.
  rds.access-key-id	ID AccessKey akun untuk instans ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey?. Penting Untuk mencegah kebocoran informasi AccessKey Anda, kami merekomendasikan agar Anda mengelola ID AccessKey menggunakan manajemen rahasia. Untuk informasi selengkapnya, lihat Kelola variabel.
  rds.access-key-secret	Rahasia AccessKey akun untuk instans ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	None	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey? Penting Untuk mencegah kebocoran informasi AccessKey Anda, kami merekomendasikan agar Anda mengelola Rahasia AccessKey menggunakan manajemen rahasia. Untuk informasi selengkapnya, lihat Kelola variabel.
  rds.db-instance-id	ID instans instans ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	None	None.
  rds.main-db-id	ID database utama instans ApsaraDB RDS for MySQL.	Tidak	STRING	None	Untuk informasi selengkapnya tentang cara mendapatkan ID database utama, lihat Backup log untuk ApsaraDB RDS for MySQL. Didukung hanya di VVR 8.0.7 dan versi selanjutnya.
  rds.download.timeout	Waktu tunggu untuk mengunduh satu log arsip dari OSS.	Tidak	DURATION	60s	None.
  rds.endpoint	Titik akhir layanan untuk mendapatkan informasi binlog OSS.	Tidak	STRING	None	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 pembaca idle ditutup setelah fase snapshot berakhir.	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 aliran changelog diubah menjadi aliran append-only.	Tidak	BOOLEAN	false	Nilai yang valid: true: Semua jenis pesan, termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER, diubah menjadi pesan INSERT. Aktifkan ini hanya dalam skenario khusus, seperti saat Anda perlu menyimpan pesan hapus dari tabel upstream. false (Default): Semua jenis pesan dikirim downstream apa adanya. Catatan Didukung hanya di VVR 8.0.8 dan versi selanjutnya.
  scan.only.deserialize.captured.tables.changelog.enabled	Selama fase inkremental, menentukan apakah hanya mendeserialisasi event perubahan untuk 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: Mendeserialisasi data perubahan hanya untuk tabel target guna mempercepat pembacaan binlog. false (Default): Mendeserialisasi data perubahan untuk semua tabel. Catatan Didukung hanya di VVR 8.0.7 dan versi selanjutnya. Saat digunakan di VVR 8.0.8 dan versi sebelumnya, nama opsi harus diubah menjadi debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	Selama fase inkremental, menentukan apakah 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. Kami merekomendasikan agar Anda mengambil snapshot pekerjaan Flink untuk pemulihan sebelum melakukan perubahan online tanpa penguncian. Catatan Didukung hanya di VVR 11.1 dan versi selanjutnya.
  scan.incremental.snapshot.backfill.skip	Menentukan apakah 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 Anda melewati backfill, perubahan pada tabel selama fase snapshot dibaca pada 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 mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot. false (Default): Tidak mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot. Ini adalah fitur eksperimental. Mengaktifkannya dapat mengurangi risiko error kehabisan memori (OOM) saat Pengelola Tugas menyinkronkan chunk terakhir selama fase snapshot. Kami menyarankan agar Anda menambahkan ini sebelum startup pekerjaan pertama kali. Catatan Didukung hanya di VVR 11.1 dan versi selanjutnya.
  binlog.session.network.timeout	Timeout baca/tulis jaringan untuk koneksi binlog.	Tidak	DURATION	10m	Jika Anda mengatur opsi ini ke 0s, timeout default server MySQL digunakan. Catatan Didukung hanya di VVR 11.5 dan versi selanjutnya.
  scan.rate-limit.records-per-second	Jumlah maksimum catatan yang dapat dikirim sumber per detik.	Tidak	LONG	None	Opsi ini berlaku untuk skenario di mana Anda perlu membatasi pembacaan data. Batasan berlaku di kedua fase, lengkap dan inkremental. Metrik numRecordsOutPerSecond dari sumber mencerminkan jumlah catatan yang dikeluarkan aliran data per detik. Anda dapat menyesuaikan opsi ini berdasarkan metrik tersebut. Pada fase pembacaan data lengkap, Anda biasanya perlu mengurangi jumlah catatan data yang dibaca dalam setiap batch. Anda dapat mengurangi nilai opsi scan.incremental.snapshot.chunk.size. Catatan Didukung hanya di VVR 11.5 dan versi selanjutnya.

• Khusus tabel dimensi

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Keterangan
  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	Didukung hanya di VVR 6.0.7 dan versi selanjutnya.
  lookup.cache.strategy	Kebijakan cache.	Tidak	STRING	None	Mendukung tiga kebijakan cache: None, LRU, dan ALL. Untuk informasi selengkapnya tentang nilai-nilai tersebut, lihat Informasi latar belakang. Catatan Saat Anda menggunakan kebijakan cache LRU, Anda juga harus mengonfigurasi opsi lookup.cache.max-rows.
  lookup.cache.max-rows	Jumlah maksimum baris yang dicache.	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 tergantung 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, tidak kedaluwarsa. Jika lookup.cache.strategy diatur ke ALL, lookup.cache.ttl adalah waktu reload cache. Secara default, tidak direload. Gunakan format waktu, seperti 1min atau 10s.
  lookup.max-join-rows	Jumlah maksimum hasil yang dikembalikan saat mengkueri tabel dimensi untuk setiap catatan di tabel utama.	Tidak	INTEGER	1024	None.
  lookup.filter-push-down.enabled	Menentukan apakah 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 Didukung hanya di VVR 8.0.7 dan versi selanjutnya. Penting Filter pushdown untuk 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 SQL Hints. Jika tidak, pekerjaan mungkin berjalan abnormal.

• Khusus sink

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Keterangan
  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 catatan dalam satu batch penulisan.	Tidak	INTEGER	4096	None.
  sink.buffer-flush.max-rows	Jumlah catatan data yang disimpan dalam memori.	Tidak	INTEGER	10000	Opsi ini hanya berlaku setelah primary key ditentukan.
  sink.buffer-flush.interval	Interval waktu untuk mengosongkan buffer. Jika data dalam buffer tidak memenuhi kondisi output setelah waktu tunggu yang ditentukan, sistem secara otomatis mengeluarkan semua data dalam buffer.	Tidak	DURATION	1s	None.
  sink.ignore-delete	Menentukan apakah mengabaikan operasi DELETE.	Tidak	BOOLEAN	false	Saat aliran yang dihasilkan Flink SQL berisi catatan delete 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-nya, menyebabkan kesalahan 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 DELETE dan UPDATE_BEFORE dilewati. Hanya catatan INSERT dan UPDATE_AFTER yang diproses.
  sink.ignore-null-when-update	Saat memperbarui data, menentukan apakah memperbarui bidang yang sesuai menjadi null atau melewati pembaruan untuk bidang tersebut jika bidang data masuk adalah null.	Tidak	BOOLEAN	false	Nilai yang valid: true: Tidak memperbarui bidang. Opsi ini dapat diatur ke true hanya saat primary key ditetapkan untuk tabel Flink. Saat diatur ke true: Di VVR 8.0.6 dan versi sebelumnya, eksekusi batch tidak didukung untuk menulis data ke tabel sink. Di VVR 8.0.7 dan versi selanjutnya, eksekusi batch didukung untuk menulis data ke tabel sink. Meskipun penulisan batch dapat meningkatkan efisiensi penulisan dan throughput keseluruhan secara signifikan, hal ini dapat menyebabkan latensi data dan risiko OOM. Oleh karena itu, pertimbangkan trade-off berdasarkan skenario bisnis aktual Anda. false: Memperbarui bidang menjadi null. Catatan Opsi ini didukung hanya di VVR 8.0.5 dan versi selanjutnya.

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

  Kami menyarankan agar Anda tidak menggunakan 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 Flink secara default, yang dapat menyebabkan ketidakakuratan data. Untuk menggunakan tipe TINYINT(1) guna menyimpan nilai selain 0 dan 1, lihat opsi 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 p harus kurang dari atau sama dengan 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 hingga 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

Pemetaan tipe

Tabel berikut menunjukkan pemetaan tipe untuk ingesti data.