Modul sumber dan sink Flink CDC - Realtime Compute for Apache Flink

Topik ini menjelaskan modul sumber dan sink untuk Pekerjaan Ingesti Data Flink Change Data Capture (CDC) serta mencantumkan konektor yang didukung.

Konektor yang didukung

Connector	Supported type
Connector	Source	Sink
MySQL Catatan Mendukung koneksi ke ApsaraDB RDS for MySQL, PolarDB for MySQL, dan self-managed MySQL.	√	×
Paimon	×	√
Kafka	√ Catatan Hanya didukung di VVR 8.0.10 dan versi lebih baru.	√
Upsert Kafka	×	√
StarRocks	×	√
Hologres	×	√
Simple Log Service	√ Catatan Hanya didukung di VVR 11.1 dan versi lebih baru.	×
MongoDB	√ Catatan Hanya didukung di VVR 11.2 dan versi lebih baru.	×
MaxCompute	×	√ Catatan Hanya didukung di VVR 11.1 dan versi lebih baru.
SelectDB	×	√ Catatan Hanya didukung di VVR 11.1 dan versi lebih baru.
PostgreSQL CDC (Public Preview)	√ Catatan Hanya didukung di VVR 11.4 dan versi lebih baru.	×
Print	×	√
Iceberg	×	√ Catatan Hanya didukung di VVR 11.6 dan versi lebih baru.

Konfigurasi konektor

Anda dapat mengonfigurasi parameter untuk konektor sumber dan sink dalam Pekerjaan Ingesti Data Flink CDC. Untuk detail mengenai konektor yang didukung beserta parameter masing-masing, lihat bagian berikut.

# Modul sumber
source:
  type: mysql # Atau pengenal konektor lainnya
  name: MySQL Source
  # Parameter lainnya. Gunakan pasangan key: value.

# Modul sink
sink:
  type: paimon # Atau pengenal konektor lainnya
  name: Paimon Sink
  # Parameter lainnya. Gunakan pasangan key: value.

Konfigurasi umum

Parameter	Description	Required	Type	Default	Remarks
type	Jenis konektor untuk sumber atau sink.	Yes	String	None	None
name	Nama node.	No	String	None	None
using.built-in-catalog	Gunakan kembali katalog bawaan.	No	String	None	Tersedia di VVR 11.5 dan versi lebih baru. Parameter ini hanya didukung oleh konektor tertentu. Untuk daftar konektor yang didukung, lihat Gunakan kembali informasi koneksi dari katalog.

Gunakan kembali informasi koneksi dari katalog

Mulai dari VVR 11.5, Anda dapat langsung mereferensikan katalog bawaan yang dibuat pada halaman Data Management untuk Pekerjaan Ingesti Data Flink CDC Anda. Pendekatan ini menggunakan kembali properti koneksi seperti URL, username, dan password sehingga menyederhanakan konfigurasi manual.

Sintaksis

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  
sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog

Anda dapat menggunakan parameter using.built-in-catalog dalam modul source dan sink untuk mereferensikan katalog bawaan yang telah tersedia.

Sebagai contoh, dalam contoh di atas, metadata katalog untuk mysql_rds_catalog telah mencakup parameter yang diperlukan seperti hostname, username, dan password. Dengan demikian, Anda tidak perlu menentukan parameter tersebut lagi dalam pekerjaan YAML.

Catatan penggunaan

Konektor berikut mendukung penggunaan kembali informasi koneksi dari katalog:

MySQL (sumber)
Kafka (sumber)
Upsert Kafka (sink)
StarRocks (sink)
Hologres (sink)
Paimon (sink)
Simple Log Service (sumber)
Iceberg (sink)

Catatan

Parameter katalog yang tidak kompatibel dengan konfigurasi CDC YAML akan diabaikan. Untuk informasi lebih lanjut, lihat daftar parameter untuk setiap konektor.

Konfigurasi sumber

Parameter	Description	Required	Type	Default	Remarks
source-expand	Menentukan strategi distribusi untuk data yang dipancarkan dari sumber.	No	Lihat bagian sintaksis untuk detail konfigurasi.	None	Tersedia di VVR 11.6 dan versi lebih baru.

source-expand

Parameter source-expand mendistribusikan atau mereplikasi data sebelum modul Transform dan Route memprosesnya.

Sintaksis

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # Replikasi mydb.orders ke tiga tabel, masing-masing diproses secara berbeda sebelum ditulis ke sink.
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# Terapkan transformasi berbeda pada masing-masing tiga tabel yang diperluas
transform:
  # Lapisan DWD: Pertahankan semua bidang, tambahkan kolom terhitung
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # Lapisan DWS: Pertahankan hanya bidang utama, filter pesanan kecil
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # Lapisan ADS: Pertahankan hanya bidang yang diperlukan untuk pelaporan, filter pesanan yang dibatalkan
  - source-table: ads.orders_report
    projection: order_id, user_id, amount, TO_UPPER(order_status) as status
    filter: order_status <> 'CANCELLED'
    primary-keys: order_id

# Arahkan tiga tabel yang diperluas ke tabel sink yang berbeda
route:
  - source-table: dwd.orders_full
    sink-table: starrocks_dwd.orders_full_detail
  - source-table: dws.orders_summary
    sink-table: starrocks_dws.orders_summary
  - source-table: ads.orders_report
    sink-table: starrocks_ads.orders_report

sink:
  type: starrocks
  name: sink-starrocks
  jdbc-url: jdbc:mysql://localhost:9030
  load-url: localhost:8030
  username: root
  password: pass

Seperti yang ditunjukkan dalam contoh, konfigurasi ini mereplikasi data dari tabel mydb.orders ke tiga tabel logis: dwd.orders_full, dws.orders_summary, dan ads.orders_report. Masing-masing tabel tersebut kemudian diproses secara berbeda sebelum ditulis ke tabel sink terpisah.

Catatan penggunaan

Tersedia di VVR 11.6 dan versi lebih baru.
Secara default, tabel asli tidak dipertahankan setelah distribusi data. Dalam contoh sintaksis, tabel mydb.orders dihapus dari aliran data. Untuk mempertahankan tabel asli, tambahkan ke daftar output-table.
```
source-expand:
  - input-table: mydb.orders
    output-table: [ mydb.orders, db1.orders, db2.orders ]
```
Parameter input-table dan output-table tidak mendukung ekspresi reguler.

Konfigurasi sink

Parameter	Description	Required	Type	Default	Remarks
include.schema.changes	Menentukan jenis perubahan skema yang akan diterapkan.	No	List<String>	None	Secara default, semua perubahan skema disinkronkan.
exclude.schema.changes	Menentukan jenis perubahan skema yang akan dikecualikan.	No	List<String>	None	Memiliki prioritas lebih tinggi daripada `include.schema.changes`.

Untuk informasi lebih lanjut tentang cara menggunakan include.schema.changes dan exclude.schema.changes, lihat Konfigurasi Sinkronisasi Perubahan Skema.