Konfigurasi evolusi skema Flink CDC - Realtime Compute for Apache Flink

Topik ini menjelaskan konfigurasi evolusi skema untuk pekerjaan ingest data Flink CDC.

Konfigurasi evolusi skema

Pekerjaan ingest data Flink CDC dapat menyinkronkan perubahan skema dari sumber ke tabel sink, seperti membuat tabel, menambah kolom, mengganti nama kolom, mengubah tipe kolom, menghapus kolom, dan menghapus tabel.

Tabel sink mungkin tidak mendukung semua perubahan skema. Anda dapat menambahkan opsi schema.change.behavior ke modul pipeline untuk mengontrol cara tabel sink menangani perubahan skema.

pipeline:
  schema.change.behavior: EVOLVE

Saat ini, framework hanya mendukung penyinkronan jenis perubahan skema berikut. Perubahan skema yang tidak tercantum dapat menyebabkan exception pada pekerjaan dan memerlukan restart tanpa status untuk pemulihan.

Buat tabel (CREATE TABLE ...)
Tambah kolom (ALTER TABLE ... ADD COLUMN ...)
Ubah tipe kolom (ALTER TABLE ... MODIFY COLUMN ...)
Hapus kolom (ALTER TABLE ... DROP COLUMN ...)
Ganti nama kolom (ALTER TABLE ... RENAME COLUMN ...)
Truncate tabel (TRUNCATE TABLE ...)
Hapus tabel (DROP TABLE ...)
Ubah urutan kolom (MODIFY COLUMN ... FIRST/AFTER atau CHANGE COLUMN ... FIRST/AFTER)

Catatan

Event ubah urutan kolom dapat dikenali dan diproses mulai dari VVR 11.9 dan versi setelahnya.

Perilaku evolusi skema

Mode	Deskripsi
LENIENT (default)	Pekerjaan ingesti data Flink CDC secara longgar mengonversi perubahan skema dan mengirimkannya ke tabel sink. Konversi mengikuti aturan berikut: Perubahan hapus tabel, truncate tabel, dan ubah urutan kolom tidak dikirim. Untuk penggantian nama kolom, event ubah tipe kolom dan event tambah kolom dikirim sebagai gantinya. Kolom asli tidak dihapus. Tipe datanya diubah menjadi nullable, dan kolom baru dengan nama baru serta tipe data nullable ditambahkan. Untuk kolom yang dihapus, event ubah tipe kolom dikirim sebagai gantinya, dan tipe field menjadi nullable. Untuk kolom baru, event tambah kolom tetap dikirim, tetapi tipe field menjadi nullable. Gunakan mode ini jika Anda ingin pekerjaan ingesti data menyinkronkan perubahan skema secara otomatis semaksimal mungkin.
IGNORE	Pekerjaan ingesti data Flink CDC mengabaikan semua perubahan skema. Perubahan skema hulu tidak diterapkan ke tabel sink. Gunakan mode ini jika tabel sink Anda tidak mendukung perubahan skema, atau Anda tidak menginginkan adanya perubahan skema, dan tetap ingin menerima data dari kolom yang tidak berubah.
EVOLVE	Pekerjaan ingesti data Flink CDC menerapkan semua perubahan skema ke tabel sink. Jika penerapan perubahan skema ke tabel sink gagal, pekerjaan akan melemparkan exception dan memicu failover. Gunakan mode ini jika Anda ingin pekerjaan ingesti data menyinkronkan perubahan skema seketat mungkin. Penting Dalam mode ini, jika sink tidak dapat menerapkan semua event perubahan skema, pekerjaan dapat mengalami failover dan tidak dapat pulih secara mandiri.
TRY_EVOLVE	Pekerjaan ingesti data Flink CDC mencoba menerapkan perubahan skema ke tabel sink. Jika tabel sink tidak dapat memproses suatu perubahan skema, pekerjaan tidak gagal atau restart. Sebagai gantinya, pekerjaan mencoba menangani perubahan tersebut dengan mentransformasi data selanjutnya. Gunakan mode ini jika Anda ingin pekerjaan ingesti data menyinkronkan perubahan skema seketat mungkin sekaligus mempertahankan tingkat toleransi kesalahan tertentu. Penting Dalam mode TRY_EVOLVE, jika penerapan perubahan skema gagal, data selanjutnya dari hulu mungkin kehilangan kolom atau terpotong agar sesuai dengan skema tabel sink.
EXCEPTION	Pekerjaan ingesti data Flink CDC tidak mengizinkan adanya perubahan skema. Pekerjaan akan melemparkan exception saat menerima event perubahan skema. Gunakan mode ini jika Anda harus memastikan bahwa tidak ada perubahan skema dalam pekerjaan ingesti data dan hanya data yang disinkronkan.

Kontrol perubahan skema di sink

Dalam skenario sinkronisasi data, Flink CDC menyediakan manajemen perubahan skema detail halus. Anda dapat menggunakan aturan untuk mengontrol jenis perubahan skema yang diterima oleh sink, sehingga mencegah kehilangan data atau gangguan layanan akibat perubahan tak terduga.

Untuk melakukannya, konfigurasikan modul sink dengan opsi include.schema.changes dan exclude.schema.changes.

Parameter	Deskripsi	Wajib	Tipe data	Nilai default	Keterangan
include.schema.changes	Perubahan skema yang dapat diterapkan.	Tidak	List<String>	Tidak ada	Semua perubahan didukung secara default.
exclude.schema.changes	Perubahan skema yang tidak dapat diterapkan.	Tidak	List<String>	Tidak ada	Memiliki prioritas lebih tinggi daripada `include.schema.changes`.

Tabel berikut mencantumkan semua jenis event perubahan skema yang dapat dikonfigurasi:

Jenis Peristiwa	Deskripsi
`add.column`	Tambah kolom
`alter.column.type`	Ubah tipe kolom
`create.table`	Buat tabel
`drop.column`	Hapus kolom
`drop.table`	Hapus tabel
`rename.column`	Ganti nama kolom
`truncate.table`	Truncate tabel
`reorder.column`	Ubah urutan kolom

Catatan

Perubahan skema mendukung pencocokan parsial. Misalnya, menentukan drop sama artinya dengan menentukan drop.column dan drop.table. Menentukan table sama artinya dengan menentukan create.table, truncate.table, dan drop.table.

Contoh

Contoh 1: Atur perilaku perubahan skema ke EVOLVE untuk menyinkronkan perubahan skema hulu ke sink.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604

sink:
  type: values
  name: Values Sink
  print.enabled: true
  sink.print.logger: true
  
pipeline:
  name: mysql to print job
  schema.change.behavior: EVOLVE

Contoh 2: Terapkan event buat-tabel dan tambah-kolom dari tabel hulu ke sink, dan abaikan event hapus-kolom.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604

sink:
  type: values
  name: Values Sink
  print.enabled: true
  sink.print.logger: true
  include.schema.changes: [create.table, column] # Mencocokkan event CreateTable, AddColumn, AlterColumnType, RenameColumn, dan DropColumn
  exclude.schema.changes: [drop.column] # Mengecualikan event DropColumn
  
pipeline:
  name: mysql to print job
  schema.change.behavior: EVOLVE