Connect Flink CDC to a DLF catalog - Data Lake Formation - Alibaba Cloud Dokümantasyon Merkezi

Flink Change Data Capture (CDC), Realtime Compute for Apache Flink tarafından sunulan bir veri alımı aracıdır. Kaynaklardaki tüm veritabanlarını veri gölünüze eşitlemeyi destekler. Bu konu, Paimon REST aracılığıyla DLF kataloğuna gerçek zamanlı olarak veri almak için Flink CDC'nin nasıl kullanılacağını adım adım açıklar.

Bu konunun sonunda şunları yapmış olacaksınız:

Veri alım hedefi olarak bir DLF kataloğu oluşturma
DLF'i işaret eden Paimon havuzu ile bir Flink CDC YAML işi yapılandırma
Tam veritabanı eşitlemesi, bölümlenmiş tablolar, yalnızca ekleme yapılan tablolar ve Kafka kaynakları gibi yaygın alım senaryoları için örnekleri çalıştırma

Ön Koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

Bir Realtime Compute for Apache Flink çalışma alanı. Bkz. Çalışma alanı oluşturma.
Realtime Compute for Apache Flink çalışma alanınız ve DLF kataloglarınız aynı bölgede bulunmalıdır.
Realtime Compute for Apache Flink çalışma alanınızın VPC'si, DLF'in VPC izin listesine eklenmelidir. Bkz. VPC izin listesi yapılandırma.

Motor Gereksinimleri

Realtime Compute for Apache Flink işiniz Ververica Runtime (VVR) sürüm 11.1.0 veya daha yenisini çalıştırmalıdır.

DLF Kataloğu Oluşturma

Bkz. DLF'e giriş.

Veri Alım İşini Oluşturma ve Yapılandırma

Flink CDC kullanarak veri almak için bir YAML taslağı oluşturun. Daha fazla bilgi için bkz. Veri alımı için Flink CDC işleri geliştirme (Beta).

Havuz modülünü yapılandırın:

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Yer tutucu değerleri kendi gerçek değerlerinizle değiştirin:

Yapılandırma seçeneği	Açıklama	Zorunlu	Varsayılan	Örnek değer
`catalog.properties.metastore`	Meta deposu türü. `rest` olarak ayarlayın.	Evet	—	`rest`
`catalog.properties.token.provider`	Belirteç sağlayıcısı. `dlf` olarak ayarlayın.	Evet	—	`dlf`
`catalog.properties.uri`	DLF REST katalog sunucusuna erişim URI'si. Biçim: `http://[region-id]-vpc.dlf.aliyuncs.com`. Bölge kimlikleri için bkz. Bölgeler ve uç noktalar.	Evet	—	`http://ap-southeast-1-vpc.dlf.aliyuncs.com`
`catalog.properties.warehouse`	Paimon kataloğunun adı.	Evet	—	`dlf_test`
`commit.user`	Veri yazmaları için commit kullanıcı adı. Çakışmaları önlemek amacıyla farklı işlere benzersiz commit kullanıcı adları atayın.	Hayır	`admin`	`your_job_name`
`table.properties.deletion-vectors.enabled`	Yazma performansı üzerinde minimal etkiyle okuma performansını artırmak için silme vektörlerini etkinleştirir.	Hayır	—	`true`

Kullanım Notları

Bir işi çalıştırmadan önce aşağıdaki kısıtlamaları dikkate alın:

Commit kullanıcı çakışmaları: Varsayılan commit kullanıcı adı admin değeridir. Aynı tabloya aynı commit kullanıcı adı altında eş zamanlı veri yazma işleri çalıştırmak, commit çakışmalarına ve veri tutarsızlıklarına neden olur. Her iş için benzersiz bir commit.user değeri atayın.
Sıkıştırma veya paket seçenekleri yok: DLF otomatik dosya sıkıştırma sağlar. bucket ve num-sorted-run.compaction-trigger gibi sıkıştırma veya paket seçeneklerini yapılandırmayın.
Silme vektörleri: Okuma hızlarını önemli ölçüde artırmak ve yazma performansı üzerinde minimal etki bırakmak için table.properties.deletion-vectors.enabled: true değerini ayarlayın.

Örnekler

MySQL Veritabanının Tamamından DLF'e Veri Alma

Aşağıdaki iş, bir MySQL veritabanındaki tüm tabloları DLF'e eşitler:

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Önerilen MySQL kaynağı seçenekleri şunlardır:

Seçenek	Açıklama
`scan.binlog.newly-added-table.enabled`	Artımlı aşamada oluşturulan tablolardaki verileri eşitler.
`include-comments.enabled`	Tablo ve alan açıklamalarını eşitler.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Sınırsız parçalara öncelik vererek olası TaskManager bellek yetersizliği (OOM) hatalarını önler.
`scan.only.deserialize.captured.tables.changelog.enabled`	Okuma hızını artırmak için verileri yalnızca eşleşen tablolardan serileştirir.

Bölümlenmiş Bir Tabloya Veri Alma

Bölümlenmemiş bir kaynak tablodan DLF'teki bölümlenmiş bir tabloya veri almak için transform modülüne partition-keys seçeneğini ekleyin. Daha fazla bilgi için bkz. Flink CDC ile veri alımı.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

Yalnızca Ekleme Yapılan Bir Tabloya Veri Alma

Alım sırasında mantıksal silme (soft delete) uygulamak için transform modülünde converter-after-transform: SOFT_DELETE kullanın. Bu ayar, silme işlemlerini ekleme işlemlerine dönüştürerek aşağı akış tablosunun tüm değişiklik işlemlerini eksiksiz şekilde kaydetmesini sağlar. projection içindeki __data_event_type__ alanı, değişiklik türünü aşağı akış tablosuna yeni bir sütun olarak yazar.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE
  - source-table: mysql_test.tbl2
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

Daha fazla bilgi için bkz. Flink CDC ile veri alımı.

Kafka'dan DLF'e Gerçek Zamanlı Veri Eşitleme

Aşağıdaki iş, Kafka inventory konusundan (Debezium JSON formatındaki customers ve products tabloları) CDC verilerini okur ve bunları DLF'teki ilgili hedef tablolara eşitler. Debezium JSON mesajları birincil anahtar bilgisi içermediğinden, birincil anahtar transform modülünde açıkça belirtilir.

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

# Debezium JSON does not contain primary key info, so explicitly specify primary keys.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Kafka kaynakları için ek notlar:

Kafka kaynağı üç veri formatını destekler: canal-json, debezium-json (varsayılan) ve json.
Birden çok Kafka bölümünden DLF'teki tek bir tabloya veri alırken debezium-json.distributed-tables veya canal-json.distributed-tables değerini true olarak ayarlayın.
Kafka kaynağı, schema.inference.strategy seçeneği aracılığıyla birden çok şema çıkarım ilkesini destekler. Daha fazla bilgi için bkz. Message Queue for Apache Kafka.

Daha fazla bilgi için bkz. Flink CDC ile veri alımı.