Flink CDC source and sink modules - Realtime Compute for Apache Flink

Bu konu, Flink Change Data Capture (CDC) veri alımı işlerinde kullanılan kaynak ve hedef modüllerini açıklar ve desteklenen bağlayıcıları listeler.

Desteklenen bağlayıcılar

Bağlayıcı	Desteklenen tür
Bağlayıcı	Kaynak	Hedef
MySQL Not ApsaraDB RDS for MySQL, PolarDB for MySQL ve kendi kendine yönetilen MySQL bağlantılarını destekler.	√	×
Paimon	×	√
Kafka	√ Not Yalnızca VVR 8.0.10 ve sonraki sürümlerde desteklenir.	√
Upsert Kafka	×	√
StarRocks	×	√
Hologres	×	√
Simple Log Service	√ Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.	×
MongoDB	√ Not Yalnızca VVR 11.2 ve sonraki sürümlerde desteklenir.	×
MaxCompute	×	√ Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
SelectDB	×	√ Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
PostgreSQL CDC (Public Preview)	√ Not Yalnızca VVR 11.4 ve sonraki sürümlerde desteklenir.	×
Print	×	√
Iceberg	×	√ Not Yalnızca VVR 11.6 ve sonraki sürümlerde desteklenir.

Bağlayıcı yapılandırması

Flink CDC veri alımı işlerinde kaynak ve hedef bağlayıcıları için parametreleri yapılandırabilirsiniz. Desteklenen bağlayıcılar ve bunların parametreleriyle ilgili ayrıntılar için aşağıdaki bölümlere bakın.

# Source module
source:
  type: mysql # Or another connector identifier
  name: MySQL Source
  # Other parameters. Use key: value pairs.

# Sink module
sink:
  type: paimon # Or another connector identifier
  name: Paimon Sink
  # Other parameters. Use key: value pairs.

Genel yapılandırma

Parametre	Açıklama	Gerekli	Tür	Varsayılan	Açıklamalar
type	Kaynak veya hedef için bağlayıcı türü.	Evet	String	Yok	Yok
name	Düğümün adı.	Hayır	String	Yok	Yok
using.built-in-catalog	Yerleşik bir kataloğu yeniden kullanır.	Hayır	String	Yok	VVR 11.5 ve sonraki sürümlerde kullanılabilir. Bu parametre yalnızca belirli bağlayıcılar tarafından desteklenir. Desteklenen bağlayıcıların listesi için bkz. Katalogdan bağlantı bilgilerini yeniden kullanma.

Katalogdan bağlantı bilgilerini yeniden kullanma

VVR 11.5 ile başlayarak, Flink CDC veri alımı işiniz için Data Management sayfasında oluşturulan yerleşik bir kataloğu doğrudan referans alabilirsiniz. Bu yaklaşım, URL, kullanıcı adı ve parola gibi bağlantı özelliklerini yeniden kullanarak manuel yapılandırmayı basitleştirir.

Sözdizimi

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

Mevcut bir yerleşik kataloğa referans vermek için source ve sink modüllerinde using.built-in-catalog parametresini kullanabilirsiniz.

Örneğin, yukarıdaki örnekte mysql_rds_catalog katalog meta verileri; hostname, username ve password gibi gerekli parametreleri zaten içerir. Bu nedenle, bu parametreleri YAML işinde tekrar belirtmeniz gerekmez.

Kullanım notları

Aşağıdaki bağlayıcılar, katalogdan bağlantı bilgilerinin yeniden kullanılmasını destekler:

MySQL (kaynak)
Kafka (kaynak)
Upsert Kafka (hedef)
StarRocks (hedef)
Hologres (hedef)
Paimon (hedef)
Simple Log Service (kaynak)
Iceberg (hedef)

Not

CDC YAML yapılandırmasıyla uyumsuz olan katalog parametreleri yoksayılır. Daha fazla bilgi için her bağlayıcının parametre listesine bakın.

Kaynak yapılandırması

Parametre	Açıklama	Gerekli	Tür	Varsayılan	Açıklamalar
source-expand	Kaynaktan yayılan veriler için dağıtım stratejisini belirtir.	Hayır	Yapılandırma ayrıntıları için sözdizimi bölümüne bakın.	Yok	VVR 11.6 ve sonraki sürümlerde kullanılabilir.

source-expand

source-expand parametresi, Transform ve Route modülleri işlemeye başlamadan önce verileri dağıtır veya çoğaltır.

Sözdizimi

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # Replicate mydb.orders into three tables, each processed differently before writing to the sink.
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# Apply different transformations to each of the three expanded tables
transform:
  # DWD layer: Retain all fields, add a calculated column
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # DWS layer: Retain only key fields, filter out small orders
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # ADS layer: Retain only fields needed for reporting, filter out canceled orders
  - 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

# Route the three expanded tables to different sink tables
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

Örnekte gösterildiği gibi, bu yapılandırma mydb.orders tablosundaki verileri üç mantıksal tabloya çoğaltır: dwd.orders_full, dws.orders_summary ve ads.orders_report. Bu tabloların her biri, ayrı bir hedef tabloya yazılmadan önce farklı şekilde işlenir.

Kullanım notları

VVR 11.6 ve sonraki sürümlerde kullanılabilir.
Varsayılan olarak, veri dağıtımından sonra orijinal tablo korunmaz. Sözdizimi örneğinde, mydb.orders tablosu veri akışından kaldırılır. Orijinal tabloyu korumak için bunu output-table listesine ekleyin.
```
source-expand:
  - input-table: mydb.orders
    output-table: [ mydb.orders, db1.orders, db2.orders ]
```
input-table ve output-table parametreleri normal ifadeleri desteklemez.

Hedef yapılandırması

Parametre	Açıklama	Gerekli	Tür	Varsayılan	Açıklamalar
include.schema.changes	Uygulanacak şema değişikliği türlerini belirtir.	Hayır	List<String>	Yok	Varsayılan olarak tüm şema değişiklikleri senkronize edilir.
exclude.schema.changes	Dışlanacak şema değişikliği türlerini belirtir.	Hayır	List<String>	Yok	`include.schema.changes` parametresinden daha yüksek önceliğe sahiptir.

include.schema.changes ve exclude.schema.changes parametrelerinin nasıl kullanılacağı hakkında daha fazla bilgi için bkz. Şema Değişikliği Senkronizasyonu Yapılandırması.