Flink CDC schema evolution configurations - Realtime Compute for Apache Flink

Şema evrimi yapılandırmaları

Flink CDC veri alımı işleri; tablo oluşturma, sütun ekleme, sütun yeniden adlandırma, sütun türü değiştirme, sütun silme ve tablo silme gibi şema değişikliklerini kaynaktan hedef tabloya senkronize edebilir.

Hedef tablo tüm şema değişikliklerini desteklemeyebilir. Hedef tablonun şema değişikliklerini nasıl işleyeceğini kontrol etmek için işlem hattı modülüne schema.change.behavior seçeneğini ekleyebilirsiniz.

pipeline:
  schema.change.behavior: EVOLVE

Çerçeve şu anda yalnızca aşağıdaki şema değişikliği türlerinin senkronizasyonunu destekler. Listede yer almayan şema değişiklikleri iş istisnalarına neden olabilir; kurtarma için durum bilgisi içermeyen bir yeniden başlatma gerekebilir.

Tablo oluşturma (CREATE TABLE ...)
Sütun ekleme (ALTER TABLE ... ADD COLUMN ...)
Sütun türünü değiştirme (ALTER TABLE ... MODIFY COLUMN ...)
Sütun silme (ALTER TABLE ... DROP COLUMN ...)
Sütun yeniden adlandırma (ALTER TABLE ... RENAME COLUMN ...)
Tabloyu boşaltma (TRUNCATE TABLE ...)
Tablo silme (DROP TABLE ...)
Sütun sırasını değiştirme (MODIFY COLUMN ... FIRST/AFTER veya CHANGE COLUMN ... FIRST/AFTER)

Not

Sütun sırası değiştirme olayı VVR 11.9 ve sonraki sürümlerde tanınır ve işlenir.

Şema evrimi davranışı

Mod	Açıklama
LENIENT (varsayılan)	Flink CDC veri alımı işi, şema değişikliklerini esnek biçimde dönüştürür ve hedef tabloya gönderir. Dönüşüm aşağıdaki kuralları izler: Tablo silme, tabloyu boşaltma ve sütun sırası değiştirme değişiklikleri gönderilmez. Sütun yeniden adlandırma durumunda bunun yerine bir sütun türü değiştirme olayı ve bir sütun ekleme olayı gönderilir. Orijinal sütun silinmez. Türü null yapılabilir (nullable) olarak değiştirilir ve yeni ada sahip, null yapılabilir veri türünde yeni bir sütun eklenir. Silinen bir sütun için bunun yerine bir sütun türü değiştirme olayı gönderilir ve alan türü null yapılabilir hale gelir. Yeni bir sütun için sütun ekleme olayı yine gönderilir ancak alan türü null yapılabilir hale gelir. Veri alımı işinin şema değişikliklerini mümkün olduğunca otomatik olarak senkronize etmesini istediğinizde bu modu kullanın.
IGNORE	Flink CDC veri alımı işi tüm şema değişikliklerini yok sayar. Kaynak tarafındaki şema değişiklikleri hedef tabloya uygulanmaz. Hedef tablonuz şema değişikliklerini desteklemiyorsa, şema değişikliklerinin gerçekleşmesini istemiyorsanız ve değişmemiş sütunlardan veri almaya devam etmek istiyorsanız bu modu kullanın.
EVOLVE	Flink CDC veri alımı işi tüm şema değişikliklerini hedef tabloya uygular. Bir şema değişikliği hedef tabloya uygulanamadığında iş bir istisna oluşturur ve yük devretmeyi tetikler. Veri alımı işinin şema değişikliklerini mümkün olduğunca sıkı biçimde senkronize etmesini istediğinizde bu modu kullanın. Önemli Bu modda, hedef tüm şema değişikliği olaylarını uygulayamazsa iş yük devredebilir ve kendi kendine kurtaramaz.
TRY_EVOLVE	Flink CDC veri alımı işi, şema değişikliklerini hedef tabloya uygulamayı dener. Hedef tablo bir şema değişikliğini işleyemezse iş başarısız olmaz veya yeniden başlatılmaz. Bunun yerine, değişikliği sonraki verileri dönüştürerek işlemeye çalışır. Veri alımı işinin şema değişikliklerini mümkün olduğunca sıkı biçimde senkronize ederken belirli bir hata toleransı düzeyini korumasını istediğinizde bu modu kullanın. Önemli TRY_EVOLVE modunda, bir şema değişikliği uygulanamadığında kaynaktan gelen sonraki veriler hedef tablo şemasına uymak için sütunları kaybedebilir veya kesilebilir.
EXCEPTION	Flink CDC veri alımı işi herhangi bir şema değişikliğine izin vermez. İş, bir şema değişikliği olayı aldığında bir istisna oluşturur. Veri alımı işinde hiçbir şema değişikliği olmamasını ve yalnızca verilerin senkronize edilmesini sağlamanız gerektiğinde bu modu kullanın.

Hedefte şema değişikliklerini kontrol etme

Veri senkronizasyonu senaryolarında Flink CDC, ayrıntılı şema değişikliği yönetimi sağlar. Beklenmeyen değişikliklerden kaynaklanan veri kaybını veya hizmet kesintisini önlemek için hedefin alacağı şema değişikliği türlerini kurallarla kontrol edebilirsiniz.

Bunu yapmak için sink modülünü include.schema.changes ve exclude.schema.changes seçenekleriyle yapılandırın.

Parametre	Açıklama	Gerekli mi?	Veri türü	Varsayılan değer	Açıklamalar
include.schema.changes	Uygulanabilecek şema değişiklikleri.	Hayır	List<String>	Yok	Tüm değişiklikler varsayılan olarak desteklenir.
exclude.schema.changes	Uygulanamayacak şema değişiklikleri.	Hayır	List<String>	Yok	`include.schema.changes` seçeneğine göre önceliklidir.

Aşağıdaki tabloda yapılandırılabilir tüm şema değişikliği olay türleri listelenmiştir:

Olay türü	Açıklama
`add.column`	Sütun ekleme
`alter.column.type`	Sütun türünü değiştirme
`create.table`	Tablo oluşturma
`drop.column`	Sütun silme
`drop.table`	Tablo silme
`rename.column`	Sütun yeniden adlandırma
`truncate.table`	Tabloyu boşaltma
`reorder.column`	Sütun sırasını değiştirme

Not

Şema değişiklikleri kısmi eşleşmeyi destekler. Örneğin, drop belirtmek, hem drop.column hem de drop.table belirtmeye eşdeğerdir. table belirtmek ise create.table, truncate.table ve drop.table belirtmeye eşdeğerdir.

Örnekler

Örnek 1: Kaynak tarafındaki şema değişikliklerini hedefe senkronize etmek için şema değişikliği davranışını EVOLVE olarak ayarlayın.

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

Örnek 2: Kaynak tablodan gelen tablo oluşturma ve sütun olaylarını hedefe uygulayın, sütun silme olaylarını ise yok sayın.

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] # Matches the CreateTable, AddColumn, AlterColumnType, RenameColumn, and DropColumn events
  exclude.schema.changes: [drop.column] # Excludes the DropColumn event
  
pipeline:
  name: mysql to print job
  schema.change.behavior: EVOLVE