Realtime Compute for Apache Flink：MySQL

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	Required	Tipe data	Nilai default	Catatan
  connector	Tipe tabel.	Ya	STRING	Tidak ada	Saat digunakan sebagai tabel sumber, Anda dapat mengatur ini ke mysql-cdc atau mysql. Kedua nilai ini setara. Saat digunakan sebagai tabel dimensi atau sink, nilainya harus mysql.
  hostname	Alamat IP atau nama host database MySQL.	Ya	STRING	Tidak ada	Kami menyarankan agar Anda memasukkan 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 akses. Untuk informasi selengkapnya, lihat Manajemen dan operasi penyimpanan dan Bagaimana kluster Flink yang sepenuhnya dikelola mengakses Internet?.
  username	Nama pengguna layanan database MySQL.	Ya	STRING	Tidak ada	Tidak ada.
  password	Kata sandi layanan database MySQL.	Ya	STRING	Tidak ada	Tidak ada.
  database-name	Nama database MySQL.	Ya	STRING	Tidak ada	Saat digunakan sebagai tabel sumber, nama database mendukung ekspresi reguler untuk membaca data dari beberapa database. Saat menggunakan ekspresi reguler, hindari penggunaan simbol ^ dan $ untuk mencocokkan awal dan akhir string. Untuk alasan spesifiknya, lihat keterangan untuk table-name.
  table-name	Nama tabel MySQL.	Ya	STRING	Tidak ada	Saat digunakan sebagai tabel sumber, nama tabel mendukung ekspresi reguler untuk membaca data dari beberapa tabel. Saat membaca dari beberapa tabel MySQL, kirimkan beberapa pernyataan CTAS sebagai satu pekerjaan untuk menghindari pengaktifan beberapa listener Binlog, yang 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. Alasannya dijelaskan di bawah ini. Catatan Saat mencocokkan nama tabel dengan ekspresi reguler, tabel sumber CDC MySQL menggabungkan database-name dan table-name yang Anda berikan dengan string \\. (atau karakter . sebelum Ververica Runtime (VVR) 8.0.1) untuk membentuk ekspresi reguler yang memenuhi syarat sepenuhnya. Ekspresi ini kemudian digunakan untuk mencocokkan nama tabel yang memenuhi syarat sepenuhnya dalam database MySQL. Misalnya, jika Anda mengonfigurasi 'database-name'='db_.*' dan 'table-name'='tb_.+', konektor akan menggunakan ekspresi reguler db_.*\\.tb_.+ (atau db_.*.tb_.+ sebelum versi 8.0.1) untuk mencocokkan nama tabel yang memenuhi syarat sepenuhnya guna menentukan tabel mana yang akan dibaca.
  port	Nomor port layanan database MySQL.	Tidak	INTEGER	3306	Tidak ada.

• Khusus tabel sumber

  Parameter	Deskripsi	Diperlukan	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. Kami menyarankan agar Anda menetapkan ID yang berbeda untuk setiap pekerjaan yang terhubung ke database yang sama. Parameter ini juga mendukung rentang ID, seperti 5400-5408. Saat pembacaan inkremental diaktifkan, beberapa pembaca konkuren didukung. Dalam kasus ini, kami menyarankan agar Anda menetapkan rentang ID sehingga setiap pembaca konkuren menggunakan ID yang berbeda. Untuk informasi selengkapnya, lihat Penggunaan server ID.
  scan.incremental.snapshot.enabled	Menentukan apakah akan mengaktifkan snapshot inkremental.	Tidak	BOOLEAN	true	Snapshot inkremental diaktifkan secara default. Ini adalah mekanisme baru untuk membaca snapshot data penuh. Dibandingkan dengan metode pembacaan snapshot lama, snapshot inkremental menawarkan beberapa keunggulan: 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 rentang harus lebih besar dari atau sama dengan tingkat paralelisme. Catatan Item konfigurasi ini dihapus di mesin komputasi Flink VVR 11.1 dan versi yang lebih baru.
  scan.incremental.snapshot.chunk.size	Ukuran setiap chunk (jumlah baris).	Tidak	INTEGER	8096	Saat pembacaan snapshot inkremental diaktifkan, tabel dibagi menjadi beberapa chunk untuk dibaca. Data dalam chunk di-cache dalam memori sebelum sepenuhnya dibaca. Jumlah baris per chunk yang lebih kecil menghasilkan total jumlah chunk yang lebih besar di tabel. Meskipun ini meningkatkan granularitas pemulihan kesalahan, ini dapat menyebabkan kesalahan kehabisan memori (OOM) dan throughput keseluruhan yang lebih rendah. Oleh karena itu, Anda perlu menemukan keseimbangan dan menetapkan ukuran chunk yang masuk akal.
  scan.snapshot.fetch.size	Jumlah maksimum record yang diambil sekaligus saat membaca data penuh dari tabel.	Tidak	INTEGER	1024	Tidak ada.
  scan.startup.mode	Mode startup untuk konsumsi data.	Tidak	STRING	initial	Nilai valid: initial (default): Saat startup pertama, memindai data historis penuh dan kemudian membaca data Binlog terbaru. latest-offset: Saat startup pertama, tidak memindai data historis dan mulai membaca dari akhir Binlog (posisi terbaru). Hanya membaca perubahan yang terjadi setelah konektor dimulai. earliest-offset: Tidak memindai data historis dan mulai membaca dari Binlog yang tersedia paling awal. specific-offset: Tidak memindai data historis dan mulai dari offset Binlog tertentu. Anda dapat menentukan offset dengan mengonfigurasi scan.startup.specific-offset.file dan scan.startup.specific-offset.pos, atau Anda dapat menentukan set GTID dengan hanya mengonfigurasi scan.startup.specific-offset.gtid-set. timestamp: Tidak memindai data historis dan mulai membaca Binlog 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 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 untuk offset awal saat menggunakan mode startup offset tertentu.	Tidak	STRING	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke specific-offset. Contoh format nama file: mysql-bin.000003.
  scan.startup.specific-offset.pos	Offset dalam file Binlog yang ditentukan untuk offset awal saat menggunakan mode startup offset tertentu.	Tidak	INTEGER	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke specific-offset.
  scan.startup.specific-offset.gtid-set	Set GTID untuk offset awal saat menggunakan mode startup offset tertentu.	Tidak	STRING	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke specific-offset. Contoh format set GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	Timestamp offset awal dalam milidetik saat menggunakan mode startup timestamp.	Tidak	LONG	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke timestamp. Timestamp dalam milidetik. Penting Saat menggunakan waktu tertentu, CDC MySQL mencoba membaca event awal setiap file Binlog untuk menentukan timestamp-nya, akhirnya menemukan file Binlog yang sesuai dengan waktu yang ditentukan. Pastikan file Binlog untuk timestamp yang ditentukan belum dibersihkan 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, yaitu zona waktu wilayah yang Anda pilih.	Misalnya, Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP MySQL dikonversi ke tipe STRING. Untuk informasi selengkapnya, lihat Debezium temporal values.
  debezium.min.row.count.to.stream.results	Ketika jumlah baris dalam tabel melebihi nilai ini, mode pembacaan batch digunakan.	Tidak	INTEGER	1.000	Flink membaca data dari tabel sumber MySQL dengan salah satu cara berikut: Baca penuh: Membaca seluruh data tabel langsung ke memori. Ini cepat tetapi mengonsumsi memori yang sesuai. Jika tabel sumber sangat besar, ada risiko error OOM. Baca batch: Membaca data dalam beberapa batch, mengambil sejumlah baris tertentu setiap kali hingga semua data dibaca. Ini menghindari risiko OOM untuk tabel besar tetapi relatif lebih lambat.
  connect.timeout	Jangka waktu maksimum untuk menunggu koneksi ke server database MySQL mengalami timeout sebelum mencoba kembali.	Tidak	DURATION	30s	Tidak ada.
  connect.max-retries	Jumlah maksimum percobaan ulang setelah koneksi gagal ke layanan database MySQL.	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 ke database.
  jdbc.properties.*	Parameter koneksi kustom untuk 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 MySQL Configuration Properties.
  debezium.*	Parameter kustom untuk Debezium guna membaca Binlog.	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.
  heartbeat.interval	Interval waktu sumber memajukan offset Binlog menggunakan event heartbeat.	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, mencegahnya kedaluwarsa. Offset Binlog yang kedaluwarsa akan menyebabkan pekerjaan gagal dan memerlukan restart tanpa status.
  scan.incremental.snapshot.chunk.key-column	Menentukan kolom yang akan digunakan sebagai kunci pemisahan untuk sharding selama fase snapshot.	Lihat Keterangan.	STRING	Tidak ada	Diperlukan untuk tabel tanpa kunci utama. Kolom yang dipilih harus bertipe non-null (NOT NULL). Opsional untuk tabel dengan primary key. Hanya satu kolom dari primary key yang dapat dipilih.
  rds.region-id	ID wilayah instans Alibaba Cloud 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.
  rds.access-key-id	ID AccessKey akun Alibaba Cloud RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat Informasi AccessKey?. Penting Untuk mencegah Informasi AccessKey Anda bocor, kami menyarankan agar Anda mengelola rahasia untuk menentukan ID AccessKey. Untuk informasi selengkapnya, lihat Manajemen variabel.
  rds.access-key-secret	Rahasia AccessKey akun Alibaba Cloud RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat Informasi AccessKey? Penting Untuk mencegah Informasi AccessKey Anda bocor, kami menyarankan agar Anda mengelola rahasia untuk menentukan Rahasia AccessKey. Untuk informasi selengkapnya, lihat Manajemen variabel.
  rds.db-instance-id	ID instans Alibaba Cloud RDS for MySQL.	Wajib saat menggunakan fitur membaca log arsip dari OSS.	STRING	Tidak ada	Tidak ada.
  rds.main-db-id	ID database utama instans Alibaba Cloud RDS for MySQL.	Tidak	STRING	Tidak ada	Untuk informasi selengkapnya tentang cara mendapatkan ID database utama, lihat Cadangan log RDS for MySQL. Hanya didukung di mesin komputasi Flink VVR 8.0.7 dan versi yang lebih baru.
  rds.download.timeout	Periode waktu habis untuk mengunduh satu log arsip dari OSS.	Tidak	DURASI	60s	Tidak ada.
  rds.endpoint	Titik akhir untuk mendapatkan informasi Binlog OSS.	Tidak	STRING	Tidak ada	Untuk informasi lebih lanjut tentang nilai yang valid, lihat Endpoints. Hanya didukung di mesin komputasi Flink VVR 8.0.8 dan versi yang lebih baru.
  scan.incremental.close-idle-reader.enabled	Menentukan apakah akan menutup pembaca tidak aktif setelah fase Snapshot berakhir.	Tidak	BOOLEAN	false	Hanya didukung di mesin komputasi Flink VVR 8.0.1 dan versi yang lebih baru. Agar konfigurasi ini berlaku, atur execution.checkpointing.checkpoints-after-tasks-finish.enabled ke true.
  scan.read-changelog-as-append-only.enabled	Menentukan apakah aliran data changelog diubah 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 ini hanya dalam skenario khusus, seperti ketika Anda perlu menyimpan pesan penghapusan dari tabel upstream. false (default): Semua jenis pesan dikirim ke hilir apa adanya. Catatan Hanya didukung di mesin komputasi Flink VVR 8.0.8 dan versi yang lebih baru.
  scan.only.deserialize.captured.tables.changelog.enabled	Pada fase inkremental, menentukan apakah hanya 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, yang mempercepat pembacaan Binlog. false (default): Mendeserialisasi data perubahan dari semua tabel. Catatan Hanya didukung di mesin komputasi Flink VVR 8.0.7 dan versi yang lebih baru. Saat menggunakan ini di mesin komputasi Flink VVR 8.0.8 dan 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 mencoba mengurai event DDL perubahan tanpa lock RDS.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengurai event DDL perubahan tanpa lock RDS. false (default): Tidak mengurai event DDL perubahan tanpa lock RDS. Ini adalah fitur eksperimen. Kami menyarankan agar Anda mengambil snapshot pekerjaan Flink untuk pemulihan sebelum melakukan perubahan online tanpa lock. Catatan Hanya didukung di mesin komputasi Flink 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. Jika backfill dilewati, perubahan pada tabel selama fase snapshot akan dibaca pada fase inkremental berikutnya, bukan digabungkan ke dalam snapshot. Penting Melewati backfill dapat menyebabkan ketidaksesuaian data karena perubahan yang terjadi selama fase snapshot mungkin diputar ulang. Hanya jaminan semantik setidaknya sekali yang diberikan. Catatan Hanya didukung di mesin komputasi Flink VVR 11.1 dan versi yang lebih baru.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Menentukan apakah mendistribusikan chunk tak terbatas terlebih dahulu selama fase pembacaan snapshot.	Tidak	BOOELEAN	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 eksperimen. Mengaktifkannya dapat mengurangi risiko error OOM pada Pengelola Tugas saat menyinkronkan chunk terakhir selama fase snapshot. Kami menyarankan menambahkan ini sebelum startup pertama pekerjaan. Catatan Hanya didukung di mesin komputasi Flink VVR 11.1 dan versi yang lebih baru.

• Khusus tabel dimensi

  Parameter	Deskripsi	Diperlukan	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	BILANGAN BULAT	3	Hanya didukung di mesin komputasi Flink VVR 6.0.7 dan versi yang lebih baru.
  lookup.cache.strategy	Kebijakan cache.	Tidak	STRING	Tidak ada	Mendukung tiga kebijakan cache: None, LRU, dan ALL. Untuk informasi lebih lanjut tentang nilai-nilai tersebut, lihat Informasi latar belakang. Catatan Saat menggunakan kebijakan cache least recently used (LRU), Anda juga harus mengonfigurasi parameter lookup.cache.max-rows.
  lookup.cache.max-rows	Jumlah maksimum baris yang di-cache.	Tidak	INTEGER	100.000	Jika Anda memilih kebijakan cache LRU, Anda harus menetapkan ukuran cache. Jika Anda memilih kebijakan cache ALL, Anda tidak perlu menetapkan ukuran cache.
  lookup.cache.ttl	Waktu hidup cache (TTL).	Tidak	DURASI	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. Ini berarti cache tidak akan kedaluwarsa. Jika lookup.cache.strategy diatur ke LRU, lookup.cache.ttl adalah waktu kedaluwarsa cache. Secara default, tidak kedaluwarsa. Jika lookup.cache.strategy diatur ke ALL, lookup.cache.ttl adalah waktu reload cache. Secara default, tidak reload. 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	1.024	Tidak ada.
  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. Tabel dimensi memuat semua data saat memuat dari tabel database MySQL. Catatan Hanya didukung di mesin komputasi Flink 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 menggunakan SQL Hints saat menggunakannya sebagai tabel sumber. Jika tidak, pekerjaan mungkin berjalan tidak normal.

• Khusus tabel sink

  Parameter	Deskripsi	Diperlukan	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 catatan yang ditulis dalam satu batch.	Tidak	INTEGER	4096	Tidak ada.
  sink.buffer-flush.max-rows	Jumlah catatan data yang di-cache dalam memori.	Tidak	INTEGER	10.000	Parameter ini hanya berlaku setelah primary key ditentukan.
  sink.buffer-flush.interval	Interval waktu untuk membersihkan cache. Jika data yang di-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 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-nya, menyebabkan error data. Dengan menyetel sink.ignore-delete ke true, Anda dapat mengabaikan operasi DELETE dan UPDATE_BEFORE dari 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-null-when-update	Saat memperbarui data, menentukan apakah memperbarui bidang yang sesuai ke 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 sebelumnya, penulisan data ke tabel sink tidak mendukung eksekusi batch. Untuk VVR 8.0.7 dan yang lebih baru, penulisan data ke tabel sink mendukung eksekusi batch. Meskipun penulisan batch dapat secara signifikan meningkatkan efisiensi penulisan dan throughput keseluruhan, hal ini dapat menyebabkan latensi data dan risiko error OOM. Oleh karena itu, buat pertimbangan berdasarkan skenario bisnis aktual Anda. false: Memperbarui bidang menjadi null. Catatan Parameter ini hanya didukung di mesin komputasi waktu nyata VVR 8.0.5 dan versi yang lebih baru.

Pemetaan tipe data

• 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

  Kami menyarankan agar Anda tidak menggunakan tipe TINYINT(1) di MySQL untuk menyimpan nilai selain 0 dan 1. Saat property-version diatur ke 0, tabel sumber CDC MySQL memetakan TINYINT(1) ke tipe BOOLEAN Flink secara default, yang dapat menyebabkan ketidakakuratan data. Untuk menggunakan TINYINT(1) guna menyimpan nilai selain 0 dan 1, lihat parameter konfigurasi catalog.table.treat-tinyint1-as-boolean.

• Tabel dimensi dan 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

Ingesti Data

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

Sintaksis

source:
   type: mysql
   name: Sumber MySQL
   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

Parameter

Pemetaan tipe

Tabel berikut menunjukkan pemetaan tipe data untuk Ingesti Data.