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>'
);

dengan parameter

• Parameter umum

  Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Catatan
  connector	Jenis tabel.	Ya	STRING	Tidak ada	Saat digunakan sebagai tabel sumber, Anda dapat menentukan mysql-cdc atau mysql. Keduanya setara. Saat digunakan sebagai tabel dimensi atau sink, nilainya tetap mysql.
  hostname	Alamat IP atau hostname database MySQL.	Ya	STRING	Tidak ada	Kami merekomendasikan menentukan titik akhir virtual private cloud (VPC). Catatan Jika MySQL dan Realtime Compute for Apache Flink tidak berada dalam VPC yang sama, buat koneksi lintas-VPC atau gunakan titik akhir publik. Untuk informasi selengkapnya, lihat Manajemen dan operasi penyimpanan dan Bagaimana cara mengakses Internet dari kluster Flink yang sepenuhnya dikelola?.
  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 digunakan sebagai tabel sumber, nama database mendukung ekspresi reguler untuk membaca data dari beberapa database. Saat menggunakan ekspresi reguler, hindari penggunaan karakter ^ dan $ untuk mencocokkan awal dan akhir string. Lihat catatan untuk table-name untuk detailnya.
  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 beberapa tabel MySQL, kirim beberapa pernyataan CTAS sebagai satu Pekerjaan untuk menghindari peluncuran beberapa Pendengar log biner. Hal ini meningkatkan performa dan efisiensi. Untuk informasi selengkapnya, lihat Beberapa pernyataan CTAS: Kirim sebagai satu Pekerjaan. Saat menggunakan ekspresi reguler, hindari penggunaan karakter ^ dan $ untuk mencocokkan awal dan akhir string. Lihat catatan di bawah untuk detailnya. Catatan Saat mencocokkan nama tabel menggunakan ekspresi reguler, tabel sumber CDC MySQL menggabungkan nilai database-name dan table-name dengan string \\. (atau . untuk VVR 8.0.1 dan sebelumnya) untuk membentuk ekspresi reguler jalur lengkap. Ekspresi ini kemudian dicocokkan dengan nama tabel lengkap di MySQL. Misalnya, jika Anda mengonfigurasi 'database-name'='db_.*' dan 'table-name'='tb_.+', konektor menggunakan ekspresi reguler db_.*\\.tb_.+ (atau db_.*.tb_.+ untuk VVR 8.0.1 dan sebelumnya) untuk menentukan tabel mana yang akan dibaca.
  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 secara default.	ID ini harus unik secara global di seluruh kluster MySQL. Kami merekomendasikan memberikan ID berbeda untuk setiap pekerjaan yang mengakses database yang sama. Parameter ini juga mendukung format rentang, seperti 5400-5408. Saat pembacaan inkremental diaktifkan, beberapa pembaca konkuren didukung. Dalam kasus ini, tentukan rentang agar setiap pembaca menggunakan ID unik. Untuk informasi selengkapnya, lihat Menggunakan ID server.
  scan.incremental.snapshot.enabled	Apakah akan mengaktifkan snapshot inkremental.	Tidak	BOOLEAN	true	Snapshot inkremental diaktifkan secara default. Snapshot inkremental adalah mekanisme baru untuk membaca snapshot data lengkap. Dibandingkan dengan snapshot tradisional, snapshot inkremental menawarkan beberapa keunggulan: Pembacaan konkuren data lengkap. Checkpointing pada level chunk selama pembacaan data lengkap. Tidak memerlukan penguncian baca global (FLUSH TABLES WITH READ LOCK) selama pembacaan data lengkap. Jika Anda ingin sumber mendukung pembacaan konkuren, setiap pembaca memerlukan ID server unik. Oleh karena itu, server-id harus berupa rentang seperti 5400-6400, dan ukuran rentang harus minimal sama dengan tingkat paralelisme. Catatan Konfigurasi ini dihapus di VVR 11.1 dan seterusnya.
  scan.incremental.snapshot.chunk.size	Jumlah baris per chunk.	Tidak	INTEGER	8096	Saat snapshot inkremental diaktifkan, tabel dibagi menjadi chunk untuk dibaca. Data dari setiap chunk di-cache dalam memori hingga chunk tersebut sepenuhnya dibaca. Ukuran chunk yang lebih kecil meningkatkan jumlah total chunk. Meskipun ini mengurangi granularitas pemulihan, hal ini dapat menyebabkan kesalahan kehabisan memori (OOM) dan mengurangi throughput keseluruhan. Pertimbangkan pertukaran ini saat memilih ukuran chunk.
  scan.snapshot.fetch.size	Jumlah maksimum catatan yang diambil per bacaan saat memindai data tabel lengkap.	Tidak	INTEGER	1024	Tidak ada.
  scan.startup.mode	Mode startup untuk mengonsumsi data.	Tidak	STRING	initial	Nilai yang valid: initial (default): Saat startup pertama, memindai data historis lengkap lalu membaca data log biner terbaru. latest-offset: Saat startup pertama, melewati pemindaian data lengkap dan mulai membaca dari akhir log biner (posisi terbaru). Hanya membaca perubahan yang terjadi setelah konektor dimulai. earliest-offset: Melewati pemindaian data lengkap dan mulai membaca dari posisi log biner paling awal yang tersedia. specific-offset: Melewati pemindaian data lengkap dan mulai membaca dari offset log biner tertentu. Tentukan offset menggunakan scan.startup.specific-offset.file dan scan.startup.specific-offset.pos, atau gunakan scan.startup.specific-offset.gtid-set untuk memulai dari set GTID. timestamp: Melewati pemindaian data lengkap dan mulai membaca dari timestamp tertentu. Tentukan timestamp menggunakan scan.startup.timestamp-millis, dalam milidetik. Penting Saat menggunakan earliest-offset, specific-offset, atau timestamp, pastikan skema tabel tetap tidak berubah antara posisi log biner yang ditentukan dan startup pekerjaan. Perubahan skema dapat menyebabkan kesalahan.
  scan.startup.specific-offset.file	Nama file log biner untuk offset startup tertentu.	Tidak	STRING	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke specific-offset. Contoh nama file: mysql-bin.000003.
  scan.startup.specific-offset.pos	Offset dalam file log biner yang ditentukan untuk posisi startup.	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 posisi startup.	Tidak	STRING	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke specific-offset. Contoh set GTID: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	Timestamp startup dalam milidetik.	Tidak	LONG	Tidak ada	Saat menggunakan konfigurasi ini, scan.startup.mode harus diatur ke timestamp. Satuan timestamp adalah milidetik. Penting Saat menggunakan timestamp, CDC MySQL mencoba membaca event awal dari setiap file log biner untuk menentukan timestamp-nya dan menemukan file yang sesuai. Pastikan timestamp yang ditentukan sesuai dengan file log biner yang ada dan dapat dibaca di database.
  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 (zona waktu zona yang dipilih) sebagai zona waktu server database.	Misalnya, Asia/Shanghai. Parameter ini mengontrol bagaimana tipe TIMESTAMP MySQL dikonversi ke tipe STRING. Untuk informasi selengkapnya, lihat Nilai temporal Debezium.
  debezium.min.row.count.to.stream.results	Saat jumlah baris tabel melebihi nilai ini, gunakan mode pembacaan batch.	Tidak	INTEGER	1000	Flink membaca data dari tabel sumber MySQL sebagai berikut: Pembacaan lengkap: Memuat seluruh tabel ke memori. Cepat tetapi intensif memori. Tabel besar berisiko mengalami kesalahan OOM. Pembacaan batch: Membaca data dalam batch. Hemat memori tetapi lebih lambat untuk tabel besar.
  connect.timeout	Waktu maksimum menunggu sebelum mencoba koneksi ulang setelah timeout saat menghubungkan ke server database MySQL.	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 menggunakan kembali koneksi untuk mengurangi jumlah koneksi database.
  jdbc.properties.*	Parameter koneksi kustom untuk URL JDBC.	Tidak	STRING	Tidak ada	Anda dapat meneruskan parameter koneksi kustom. Misalnya, untuk menonaktifkan SSL, atur 'jdbc.properties.useSSL' = 'false'. Untuk parameter koneksi yang didukung, lihat Properti Konfigurasi MySQL.
  debezium.*	Parameter Debezium kustom untuk membaca log biner.	Tidak	STRING	Tidak ada	Anda dapat meneruskan parameter Debezium kustom. Misalnya, gunakan 'debezium.event.deserialization.failure.handling.mode'='ignore' untuk menentukan cara menangani kesalahan penguraian.
  heartbeat.interval	Interval di mana event heartbeat memajukan offset log biner di sumber.	Tidak	DURATION	30s	Event heartbeat memajukan offset log biner di sumber. Ini berguna untuk tabel dengan pembaruan jarang. Tanpa heartbeat, offset log biner mungkin macet, menyebabkan kedaluwarsa dan kegagalan pekerjaan. Heartbeat mencegah masalah ini.
  scan.incremental.snapshot.chunk.key-column	Tentukan kolom untuk digunakan dalam membagi chunk selama fase snapshot.	Lihat Catatan.	STRING	Tidak ada	Wajib untuk tabel tanpa primary key. Kolom yang dipilih harus non-null (NOT NULL). Opsional untuk tabel dengan primary key. Hanya kolom dari primary key yang didukung.
  rds.region-id	ID wilayah instans Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	Tidak ada	Untuk ID wilayah, lihat Wilayah dan zona.
  rds.access-key-id	ID AccessKey untuk akun Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan rahasia AccessKey saya?. Penting Untuk mencegah eksposur informasi AccessKey Anda, kami merekomendasikan menggunakan manajemen kunci untuk mengonfigurasi ID AccessKey. Untuk informasi selengkapnya, lihat Manajemen variabel.
  rds.access-key-secret	Rahasia AccessKey untuk akun Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan rahasia AccessKey saya?. Penting Untuk mencegah eksposur informasi AccessKey Anda, kami merekomendasikan menggunakan manajemen kunci untuk mengonfigurasi rahasia AccessKey. Untuk informasi selengkapnya, lihat Manajemen variabel.
  rds.db-instance-id	ID instans Alibaba Cloud ApsaraDB RDS for MySQL.	Wajib saat membaca log arsip dari OSS.	STRING	Tidak ada	Tidak ada.
  rds.main-db-id	ID database primer instans Alibaba Cloud ApsaraDB RDS for MySQL.	Tidak	STRING	Tidak ada	Untuk petunjuk mengambil ID database primer, lihat Cadangan log ApsaraDB RDS for MySQL. Didukung hanya di VVR 8.0.7 dan seterusnya.
  rds.download.timeout	Timeout untuk mengunduh satu log arsip dari OSS.	Tidak	DURATION	60s	Tidak ada.
  rds.endpoint	Titik akhir untuk mengakses informasi log biner OSS.	Tidak	STRING	Tidak ada	Untuk nilai yang valid, lihat Titik akhir layanan. Didukung hanya di VVR 8.0.8 dan seterusnya.
  scan.incremental.close-idle-reader.enabled	Apakah akan menutup pembaca idle setelah penyelesaian snapshot.	Tidak	BOOLEAN	false	Didukung hanya di VVR 8.0.1 dan seterusnya. Pengaturan ini memerlukan execution.checkpointing.checkpoints-after-tasks-finish.enabled diatur ke true.
  scan.read-changelog-as-append-only.enabled	Apakah akan mengonversi aliran data changelog menjadi aliran data append-only.	Tidak	BOOLEAN	false	Nilai yang valid: true: Semua jenis pesan (INSERT, DELETE, UPDATE_BEFORE, UPDATE_AFTER) dikonversi menjadi pesan INSERT. Gunakan hanya dalam kasus khusus, seperti mempertahankan pesan penghapusan tabel upstream. false (default): Semua jenis pesan diteruskan tanpa perubahan. Catatan Didukung hanya di VVR 8.0.8 dan seterusnya.
  scan.only.deserialize.captured.tables.changelog.enabled	Selama fase inkremental, apakah hanya mendeserialisasi event perubahan untuk tabel yang ditentukan.	Tidak	BOOLEAN	Default adalah false di VVR 8.x. Default adalah true di VVR 11.1 dan seterusnya.	Nilai yang valid: true: Hanya mendeserialisasi data perubahan untuk tabel target untuk mempercepat pembacaan log biner. false (default): Mendeserialisasi data perubahan untuk semua tabel. Catatan Didukung hanya di VVR 8.0.7 dan seterusnya. Di VVR 8.0.8 dan sebelumnya, gunakan debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	Selama fase inkremental, apakah mencoba mengurai event perubahan DDL lockless RDS.	Tidak	BOOLEAN	false	Nilai yang valid: true: Mengurai event perubahan DDL lockless RDS. false (default): Tidak mengurai event perubahan DDL lockless RDS. Fitur eksperimental. Sebelum melakukan perubahan lockless online, ambil snapshot pekerjaan Flink untuk pemulihan. Catatan Didukung hanya di VVR 11.1 dan seterusnya.
  scan.incremental.snapshot.backfill.skip	Apakah akan melewati backfill selama pembacaan snapshot.	Tidak	BOOLEAN	false	Nilai yang valid: true: Melewati backfill selama pembacaan snapshot. false (default): Tidak melewati backfill selama pembacaan snapshot. Jika backfill dilewati, perubahan yang terjadi selama fase snapshot dibaca dalam fase inkremental berikutnya daripada digabungkan ke dalam snapshot. Penting Melewati backfill dapat menyebabkan inkonsistensi data karena perubahan yang terjadi selama fase snapshot mungkin diputar ulang. Ini hanya memberikan semantik at-least-once. Catatan Didukung hanya di VVR 11.1 dan seterusnya.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Apakah akan mendistribusikan chunk tak terbatas terlebih dahulu selama pembacaan snapshot.	Tidak	BOOELEAN	false	Nilai yang valid: true: Mendistribusikan chunk tak terbatas terlebih dahulu selama pembacaan snapshot. false (default): Tidak mendistribusikan chunk tak terbatas terlebih dahulu selama pembacaan snapshot. Fitur eksperimental. Mengaktifkan ini mengurangi risiko kesalahan kehabisan memori (OOM) untuk TaskManager selama sinkronisasi chunk terakhir dalam fase snapshot. Kami merekomendasikan mengaktifkannya sebelum startup pekerjaan pertama. Catatan Didukung hanya di VVR 11.1 dan seterusnya.
  binlog.session.network.timeout	Timeout jaringan untuk koneksi log biner.	Tidak	DURATION	10m	Menyetel ini ke 0s menggunakan timeout default server MySQL. Catatan Didukung hanya di VVR 11.5 dan seterusnya.
  scan.rate-limit.records-per-second	Batasi jumlah maksimum catatan yang dipancarkan per detik oleh sumber.	Tidak	LONG	Tidak ada	Berguna untuk membatasi konsumsi data. Berlaku untuk fase lengkap dan inkremental. Metrik numRecordsOutPerSecond mencerminkan total catatan yang dipancarkan per detik. Sesuaikan parameter ini berdasarkan metrik tersebut. Selama pembacaan lengkap, kurangi ukuran batch untuk melengkapi batasan ini. Turunkan nilai parameter scan.incremental.snapshot.chunk.size. Catatan Didukung hanya di VVR 11.5 dan seterusnya.
  scan.binlog.tolerate.gtid-holes	Mengaktifkan parameter ini mengabaikan celah dalam urutan GTID, memungkinkan pekerjaan melewati event diskontinu dan terus berjalan.	Tidak	BOOLEAN	false	Sebelum mengaktifkan parameter ini, pastikan offset startup pekerjaan belum kedaluwarsa. Jika pekerjaan dimulai dari offset GTID yang dibersihkan atau kedaluwarsa, mesin diam-diam melewati log yang hilang, mengakibatkan kehilangan data. Catatan Didukung hanya di VVR 11.6 dan seterusnya.

• 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 operasi baca gagal.	Tidak	INTEGER	3	Didukung hanya di VVR 6.0.7 dan seterusnya.
  lookup.cache.strategy	Kebijakan cache.	Tidak	STRING	Tidak ada	Kebijakan yang didukung: None, LRU, dan ALL. Untuk deskripsi, lihat Informasi latar belakang. Catatan Saat 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	Saat kebijakan cache least recently used dipilih, ukuran cache harus diatur. Opsional saat menggunakan kebijakan cache ALL.
  lookup.cache.ttl	Waktu hidup (TTL) cache.	Tidak	DURATION	10 s	Pengaturan lookup.cache.ttl tergantung pada lookup.cache.strategy: Jika lookup.cache.strategy adalah None, lookup.cache.ttl opsional dan menunjukkan tidak ada TTL. Jika lookup.cache.strategy adalah LRU, lookup.cache.ttl adalah TTL cache. Default adalah tidak kedaluwarsa. Jika lookup.cache.strategy adalah ALL, lookup.cache.ttl adalah interval reload cache. Default adalah tidak reload. Tentukan waktu dalam format seperti 1min atau 10s.
  lookup.max-join-rows	Jumlah maksimum hasil yang dikembalikan saat menggabungkan setiap baris dari tabel utama dengan tabel dimensi.	Tidak	INTEGER	1024	Tidak ada.
  lookup.filter-push-down.enabled	Apakah akan mengaktifkan filter pushdown untuk tabel dimensi.	Tidak	BOOLEAN	false	Nilai yang valid: true: Aktifkan filter pushdown. Tabel dimensi memfilter data lebih awal berdasarkan kondisi yang ditentukan dalam pekerjaan SQL. false (default): Nonaktifkan filter pushdown. Tabel dimensi memuat semua data. Catatan Didukung hanya di VVR 8.0.7 dan seterusnya. Penting Aktifkan filter pushdown hanya saat tabel Flink digunakan sebagai tabel dimensi. Tabel sumber MySQL tidak mendukung filter pushdown. Jika tabel Flink digunakan sebagai tabel sumber dan dimensi, dan filter pushdown diaktifkan untuk tabel dimensi, atur eksplisit konfigurasi ini ke false menggunakan petunjuk SQL saat menggunakannya sebagai tabel sumber. Jika tidak, eksekusi pekerjaan mungkin gagal.

• 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 operasi tulis 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 baris yang di-cache dalam memori.	Tidak	INTEGER	10000	Parameter ini hanya berlaku saat primary key ditentukan.
  sink.buffer-flush.interval	Interval untuk flush buffer. Jika data yang di-buffer menunggu lebih lama dari interval ini tanpa memenuhi kondisi flush, sistem secara otomatis flush semua data yang di-buffer.	Tidak	DURATION	1s	Tidak ada.
  sink.ignore-delete	Apakah akan mengabaikan operasi DELETE.	Tidak	BOOLEAN	false	Saat aliran SQL Flink berisi catatan DELETE atau UPDATE_BEFORE, pembaruan simultan ke bidang berbeda dari tabel yang sama oleh beberapa tugas dapat menyebabkan inkonsistensi data. Misalnya, jika catatan dihapus dan tugas lain hanya memperbarui beberapa bidang, bidang yang tidak diperbarui menjadi null atau nilai default, menyebabkan kesalahan data. Menyetel sink.ignore-delete ke true mengabaikan operasi DELETE dan UPDATE_BEFORE upstream, menghindari masalah ini. Catatan UPDATE_BEFORE adalah bagian dari mekanisme retract Flink, digunakan untuk "menarik kembali" nilai lama selama 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, apakah akan memperbarui bidang ke null atau melewati pembaruan jika nilai input adalah null.	Tidak	BOOLEAN	false	Nilai yang valid: true: Lewati pembaruan bidang. Didukung hanya saat tabel Flink memiliki primary key. Saat diatur ke true: Di VVR 8.0.6 dan sebelumnya, penulisan batch tidak didukung untuk tabel sink. Di VVR 8.0.7 dan seterusnya, penulisan batch didukung untuk tabel sink. Penulisan batch meningkatkan efisiensi dan throughput penulisan tetapi memperkenalkan latensi dan risiko OOM. Pertimbangkan pertukaran ini berdasarkan kebutuhan bisnis Anda. false: Perbarui bidang ke null. Catatan Didukung hanya di VVR 8.0.5 dan seterusnya.

Pemetaan tipe data

• 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 diatur ke 0, tabel sumber CDC MySQL memetakan TINYINT(1) ke tipe BOOLEAN Flink secara default. Hal ini dapat menyebabkan data tidak akurat. Untuk menggunakan tipe TINYINT(1) untuk 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 mendukung catatan BLOB MySQL hingga 2.147.483.647 byte (231 − 1).
  BLOB
  MEDIUMBLOB
  LONGBLOB

Ingesti Data

Konektor MySQL dapat digunakan sebagai sumber data dalam pekerjaan ingesti data YAML.

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

Konfigurasi

Pemetaan tipe

Tabel berikut menunjukkan pemetaan tipe data untuk ingesti data.