Modul Transform Flink CDC - Realtime Compute for Apache Flink

Topik ini menjelaskan aturan sintaksis dan fungsi bawaan modul Transform untuk pekerjaan ingesti data Flink CDC.

Parameter aturan transform

Modul Transform mendukung manipulasi langsung terhadap kolom data. Anda dapat menghapus atau menambahkan kolom yang sudah ada serta menyaring data yang tidak diinginkan selama sinkronisasi data. Gunakan parameter berikut untuk menentukan aturan transform:

Parameter	Deskripsi	Wajib	Catatan
`source-table`	Menentukan tabel hulu yang akan ditransformasikan.	Ya	Mendukung ekspresi reguler.
`projection`	Menentukan aturan proyeksi untuk tabel hulu. Ini menentukan semua kolom data setelah transformasi.	Tidak	Sintaksisnya mirip dengan pernyataan SELECT SQL. Jika parameter ini tidak ditentukan, tidak ada kolom yang ditambahkan atau dihapus. Untuk informasi lebih lanjut, lihat Penyaringan data. Untuk informasi tentang fungsi bawaan yang tersedia, lihat Fungsi bawaan Flink CDC.
`filter`	Aturan untuk penyaringan baris.	Tidak	Sintaksisnya mirip dengan pernyataan WHERE SQL. Jika parameter ini tidak ditentukan, tidak ada baris yang disaring.
`primary-keys`	Menentukan kolom kunci primer setelah transformasi.	Tidak	Jika parameter ini tidak ditentukan, definisi kunci primer dari skema asli dipertahankan. Kunci primer dipisahkan dengan koma (`,`). Penting Secara default, Anda tidak dapat menghapus kolom kunci primer dari hulu. Tambahkan parameter `transform.allow.trimming.pk-columns` ke konfigurasi `pipeline:` untuk mengurangi jumlah kolom kunci primer hulu. Saat Anda menentukan kolom kunci primer kustom, pastikan data masuk dari tabel hulu mematuhi kendala PRIMARY KEY. Kami menyarankan agar Anda menyertakan kolom kunci primer tabel hulu dalam kolom kunci primer kustom untuk mencegah masalah ketidakteraturan data selama penulisan lintas partisi.
`partition-keys`	Menentukan daftar kolom kunci partisi setelah transformasi.	Tidak	Jika parameter ini tidak ditentukan, definisi kunci partisi dari skema asli dipertahankan. Kunci partisi dipisahkan dengan koma (`,`). Penting Saat Anda menentukan kolom kunci partisi kustom, pastikan data masuk dari tabel hulu mematuhi kendala PRIMARY KEY untuk mencegah masalah ketidakteraturan data selama penulisan lintas partisi.
`table-options`	Meneruskan informasi konfigurasi tambahan ke sink.	Tidak	Daftar properti opsional, seperti jumlah bucket untuk sink Paimon atau komentar. Item konfigurasi dipisahkan dengan koma (`,`). Kunci dan nilai setiap item dipisahkan dengan tanda sama dengan (`=`). Contoh konfigurasi: `key1=value1,key2=value2`
`description`	Deskripsi aturan transform.	Tidak	Tidak ada.
`converter-after-transform`	Konverter yang melakukan pemrosesan tambahan terhadap data setelah transformasi selesai.	Tidak	Untuk informasi lebih lanjut, lihat Konverter setelah Transform.

Catatan

Setelah mengubah pernyataan dalam modul transform, Anda harus me-restart pekerjaan dari keadaan awal tanpa status.
Umumnya, pernyataan projection dan filter tidak perlu diapit tanda kutip.
```
transform:
  - projection: a, b, c
    # Setara dengan
  - projection: "a, b, c"
```
Namun, jika ekspresi proyeksi dimulai dengan karakter khusus, seperti * atau ', seluruh ekspresi mungkin tidak dapat diurai sebagai literal string YAML yang valid. Dalam kasus ini, Anda harus secara manual mengapit seluruh ekspresi dengan tanda kutip tunggal (') atau ganda ("), atau menggunakan backslash (\) untuk meng-escape karakter tersebut.
```
transform:
  - projection: *, 42      # Bukan YAML yang valid
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```
Mulai dari Ververica Runtime (VVR) 11.6, Anda dapat purge pengaturan kunci primer yang diinferensi dari hulu dan mengonversi tabel menjadi tabel tanpa kunci primer. Sintaksisnya sebagai berikut:
```
transform:
  - source-table: db.tbl
    primary-keys: '' # Menghapus pengaturan PRIMARY KEY untuk tabel db.tbl
```

Penyaringan data

Modul Transform menggunakan sintaksis mirip SQL untuk menentukan aturan penyaringan data (proyeksi). Aturan ini memungkinkan Anda memilih kolom tertentu, menambahkan kolom terhitung, dan menambahkan kolom metadata.

Pemangkasan kolom

Untuk mengekstrak kolom tertentu dari tabel sumber dan menyinkronkannya ke hilir, Anda dapat menentukan kolom yang diinginkan dalam aturan proyeksi. Kolom yang tidak ditentukan tidak dikirim ke hilir.

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 dipangkas

Penting

Pemangkasan kolom dapat menyebabkan skema tabel hulu dan hilir menjadi tidak konsisten jika skema tabel hulu berubah.

Karakter wildcard

Untuk mengirim semua kolom dari tabel sumber, termasuk kolom baru yang ditambahkan nanti, ke hilir tanpa modifikasi, Anda dapat menggunakan karakter wildcard asterisk (*) dalam aturan proyeksi.

Catatan

Jika aturan proyeksi tidak menggunakan karakter wildcard (*), skema hasilnya bersifat tetap. Skema tetap konsisten dengan versi yang ditentukan dalam aturan proyeksi, dan evolusi skema tidak dapat disinkronkan.

Sebagai contoh, *, 'extras' AS extras menunjukkan bahwa kolom tambahan ditambahkan setelah kolom dari skema hulu. Evolusi skema hulu tetap disinkronkan ke hilir.

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

Kolom terhitung

Anda dapat menggunakan sintaksis <Expression> AS <ColName> dalam aturan proyeksi untuk menambahkan kolom terhitung. Ekspresi dievaluasi untuk setiap catatan dari hulu, dan hasilnya mengisi kolom yang sesuai.

Catatan

Ekspresi untuk kolom terhitung tidak dapat mereferensi nilai kolom terhitung lain, bahkan jika kolom yang direferensikan muncul sebelumnya. Sebagai contoh, a, b AS c, c AS d bukanlah ekspresi yang valid.

Sebagai contoh, ketika tabel hulu db.tbl mengirim catatan data [+I, id = 1], catatan tersebut ditransformasikan menjadi baris data [+I, id = 1, inc_id = 2] lalu dikirim ke hilir.

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

Kolom metadata

Saat menulis aturan proyeksi, Anda dapat menggunakan kolom metadata pradefinisi berikut dengan cara yang sama seperti kolom data biasa:

Penting

Jangan mendefinisikan kolom data biasa dengan nama yang sama seperti kolom metadata.

Kolom metadata berikut berlaku untuk semua konektor.

Nama kolom metadata	Tipe data	Deskripsi
`__namespace_name__`	String	Nama namespace tabel sumber untuk catatan perubahan ini.
`__schema_name__`	String	Nama skema tabel sumber untuk catatan perubahan ini.
`__table_name__`	String	Nama tabel tabel sumber untuk catatan perubahan ini.
`__data_event_type__`	String	Jenis operasi untuk catatan perubahan ini (`+I`, `-U`, `+U`, atau `-D`). Penting Event CDC selalu mengemas catatan Update Before dan Update After untuk pembaruan menjadi satu event. Oleh karena itu, isi `__data_event_type__` adalah `-U` dan `+U` masing-masing dalam satu event pembaruan yang sama. Jangan menggunakannya sebagai kunci primer.

Sebagai contoh, Anda dapat menulis nama lengkap tabel hulu ke dalam kolom terhitung dan mengirimkannya ke hilir.

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

Tabel berikut menunjukkan hubungan pemetaan antara konektor database dan nama Namespace, Skema, serta Tabel.

Jenis database	Nama namespace	Nama skema	Nama tabel
MySQL	-	Database	Tabel
Kafka	-	-	Topic
SLS	-	Proyek	LogStore
MongoDB	-	Database	Collection
Paimon	-	Database	Tabel
Hologres	-	Skema	Tabel
StarRocks	-	Database	Tabel
Doris	-	Database	Tabel
Postgres	Database Catatan Ini hanya berlaku saat parameter `table-id.include-database` diaktifkan.	Skema	Tabel

Penyaringan data

Modul Transform menggunakan sintaksis mirip SQL untuk menentukan aturan penyaringan baris.

Aturan filter harus berupa ekspresi yang mengevaluasi ke nilai BOOLEAN. Ekspresi tersebut dapat mereferensi kolom apa pun dari tabel sumber dan kolom terhitung apa pun.

Jika catatan perubahan cocok dengan aturan transform yang memiliki filter, dan ekspresi filter mengevaluasi ke FALSE, baris data tersebut tidak dikirim ke hilir.

Catatan

Jika Anda menggunakan kolom terhitung dalam aturan proyeksi untuk menimpa kolom hulu yang sudah ada, ekspresi filter mereferensi kolom terhitung tersebut.

Sebagai contoh, aturan transform berikut valid:

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

id yang direferensikan dalam ekspresi filter adalah kolom terhitung yang telah dikonversi ke tipe VARCHAR.

Aturan konfigurasi lanjutan

Referensi kolom yang tidak dipangkas dan metadata

Contoh 1: Filter berdasarkan kolom yang dipangkas

Asumsikan skema tabel hulu adalah [INT a, INT b, BOOLEAN c]. Untuk mengeluarkan kolom a dan b serta hanya menyimpan baris di mana c bernilai true, Anda dapat menggunakan aturan konfigurasi berikut:

transform:
  - source-table: ...
    projection: a, b
    filter: c

Contoh 2: Filter berdasarkan kolom metadata

Anda dapat menggunakan kolom metadata langsung sebagai kondisi filter tanpa harus mendefinisikannya secara eksplisit dalam projection. Sebagai contoh, Anda dapat menyaring data perubahan jenis DELETE.

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

Timpa kolom yang sudah ada

Dalam projection, Anda dapat mendefinisikan bidang dengan nama yang sama seperti kolom hulu untuk menimpa nilai atau tipe kolom tersebut. Metode ini memastikan evolusi skema disinkronkan dengan benar.

Contoh: Paksa konversi tipe

Asumsikan skema tabel hulu adalah [INT a, INT b, BOOLEAN c]. Untuk mempertahankan nama kolom tetapi memaksa konversi kolom a ke tipe string, Anda dapat menggunakan konfigurasi berikut:

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

Skema tabel hilir menjadi [STRING a, INT b, BOOLEAN c]. Kolom asli a ditimpa oleh definisi baru.

Gunakan kembali kolom terhitung dalam kondisi filter

filter dapat langsung mereferensi alias kolom terhitung yang didefinisikan dalam projection.

Contoh: Referensi hasil terhitung

Anda dapat mendefinisikan kolom baru d dalam projection dan menggunakannya langsung dalam filter.

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

Konfigurasi ini memiliki efek yang sama dengan konfigurasi berikut tetapi lebih mudah dibaca:

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Konverter setelah Transform

converter-after-transform digunakan untuk memproses perubahan data setelah semua aturan transform diterapkan. Anda dapat menentukan beberapa konverter dengan memisahkannya menggunakan koma (,). Konverter diterapkan sesuai urutan yang tercantum. Nilai konfigurasi berikut didukung.

Nama konverter	Fitur	Versi yang didukung
SOFT_DELETE	Mengonversi perubahan delete menjadi insert.	VVR 8.0.11 dan yang lebih baru.
FIELD_NAME_LOWER_CASE	Mengonversi semua nama bidang tabel menjadi huruf kecil.	VVR 11.1 dan yang lebih baru.

Soft delete

Konverter SOFT_DELETE bekerja dengan kolom metadata __data_event_type__ untuk mengimplementasikan penghapusan logis. Artinya, data tidak dihapus secara fisik di sistem hilir. Sebagai contoh, konfigurasi transform berikut mengimplementasikan penghapusan logis. Operasi delete dikonversi menjadi insert, dan op_type untuk data yang sesuai diperbarui menjadi -D untuk menandakan penghapusan.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE