Flink CDC job structure - Realtime Compute for Apache Flink

Bu konuda, Flink CDC tabanlı bir veri alımı işinin temel yapısı ve anahtar parametreleri açıklanmaktadır.

İş yapısı örneği

Tipik bir Flink CDC veri alımı işi aşağıdaki modüllerden oluşur:

Örneğin, MySQL'den Paimon'a veri yazan bir işin yapısı şu şekildedir:

# MySQL Source module
source:
  type: mysql
  name: MySQL Source
  host: localhost
  port: 3306
  username: admin
  password: <yourPassword>
  tables: adb.*, bdb.user_table_[0-9]+, [app|web]_order_.*, mydb.\\.*

# Paimon Sink module
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

# Transform module
transform:
  - source-table: mydb.app_order_.*
    projection: id, order_id, TO_UPPER(product_name)
    filter: id > 10 AND order_id > 100
    primary-keys: id
    partition-keys: product_name
    table-options: comment=app order
    description: project fields from source table
    converter-after-transform: SOFT_DELETE
  - source-table: mydb.web_order_.*
    projection: CONCAT(id, order_id) as uniq_id, *
    filter: uniq_id > 10
    description: add new uniq_id for each row

# Route module
route:
  - source-table: mydb.default.app_order_.*
    sink-table: odsdb.default.app_order
    description: sync all table shards to one
  - source-table: mydb.default.web_order
    sink-table: odsdb.default.ods_web_order
    description: sync table with given prefix ods_

# Pipeline module
pipeline:
  name: source-database-sync-pipe
  schema.change.behavior: evolve

Kaynak modülü

Kaynak modülü, bir Flink CDC veri alımı işinin verileri nereden okuyacağını tanımlar. Desteklenen bağlayıcılar arasında Kafka SQL bağlayıcısı, MySQL SQL bağlayıcısı, Postgres CDC (genel önizleme), MongoDB ve Simple Log Service (SLS) bulunur.

Söz dizimi

source:
  type: mysql
  name: mysql source
  xxx: ...

Desteklenen tüm parametreler için ilgili bağlayıcının belgelerine bakın.

Hedef modülü

Hedef modülü, bir Flink CDC veri alımı işinin verileri nereye yazacağını tanımlar. Desteklenen bağlayıcılar arasında Fluss, Kafka SQL bağlayıcısı, Upsert Kafka, Hologres, Paimon, Iceberg, StarRocks SQL bağlayıcısı, SelectDB, MaxCompute ve Print yer alır.

Söz dizimi

sink:
  type: hologres
  name: hologres sink
  xxx: ...

Desteklenen tüm parametreler için ilgili bağlayıcının belgelerine bakın.

Dönüşüm modülü

Bir Flink CDC veri alımı işinin Dönüşüm modülünde, kaynak tablolardaki verileri seçmek, hesaplamak ve filtrelemek için kurallar tanımlayabilirsiniz.

Söz dizimi

transform:
  - source-table: db.tbl1
    projection: ...
    filter: ...
  - source-table: db.tbl2
    projection: ...
    filter: ...

İfade söz dizimi ayrıntıları için Flink CDC Dönüşüm modülü konusuna bakın.

Yönlendirme modülü

Bir Flink CDC veri alımı işinin Yönlendirme modülünde, parçalanmış kaynak veritabanlarını ve tablolarını birleştirmek veya çoğaltmak için kurallar tanımlayabilirsiniz.

Söz dizimi

route:
  - source-table: db.tbl1
    sink-table: sinkdb.tbl1
  - source-table: db.tbl2
    sink-table: sinkdb.tbl2

İfade söz dizimi ayrıntıları için Flink CDC Yönlendirme modülü konusuna bakın.

İşlem hattı modülü

Bir Flink CDC veri alımı işinin İşlem hattı modülünde, genel parametreleri yapılandırabilirsiniz.

Söz dizimi

pipeline:
  name: CDC YAML job
  schema.change.behavior: LENIENT

Yapılandırılabilecek tüm seçenekler için Flink CDC İşlem hattı modülü konusuna bakın.