Realtime Compute for Apache Flink:Paimon connector

LENIENT (varsayılan)

LENIENT modunda aşağıdaki şema değişiklikleri desteklenir:

• Null yapılabilir bir sütun ekleme: Sütun sonuç tablosunun şemasına otomatik olarak eklenir ve yeni sütunun verileri otomatik olarak senkronize edilir.

• Null yapılabilir bir sütunu kaldırma: Sütun sonuç tablosundan kaldırılmaz. Bunun yerine sütunun verileri otomatik olarak NULL değerleriyle değiştirilir.

• NOT NULL bir sütun ekleme: Sütun sonuç tablosunun şemasına otomatik olarak eklenir ve yeni sütunun verileri otomatik olarak senkronize edilir. Yeni sütun varsayılan olarak null yapılabilir olup sütun eklenmeden önce üretilen veriler NULL olarak ayarlanır.

• Sütunu yeniden adlandırma: İşlem bir sütun ekleme ve bir sütun kaldırma olarak değerlendirilir. Yeniden adlandırılan sütun sonuç tablosunun sonuna eklenir ve özgün sütunun verileri otomatik olarak NULL değerleriyle değiştirilir. Örneğin col_a, col_b olarak yeniden adlandırılırsa col_b sonuç tablosunun sonuna eklenir ve col_a verileri NULL değerleriyle değiştirilir.

• Sütun türünü değiştirme: Sonuç tablosundaki ilgili sütunun veri türünü değiştirir.

• Aşağıdaki şema değişiklikleri desteklenmez:

  • Birincil anahtar türlerinde yapılan değişiklikler.

  • NULLABLE'dan NOT NULL'a yapılan değişiklikler.

EVOLVE

EVOLVE modunda aşağıdaki şema değişiklikleri desteklenir:

• Null yapılabilir bir sütun ekleme: Desteklenir.

• Null yapılabilir bir sütunu kaldırma: Desteklenmez.

• NOT NULL bir sütun ekleme: Sonuç tablosuna null yapılabilir bir sütun eklenir.

• Sütunu yeniden adlandırma: Desteklenir. Özgün sütun sonuç tablosunda yeniden adlandırılır.

• Sütun türünü değiştirme: Sonuç tablosundaki ilgili sütunun veri türünü değiştirir.

• Aşağıdaki şema değişiklikleri desteklenmez:

  • Birincil anahtar türlerinde yapılan değişiklikler.

  • NULLABLE'dan NOT NULL'a yapılan değişiklikler.

Aşağıdaki örnekte Paimon kataloğu rest türündeyken Data Lake Formation (DLF) üzerine veri yazma işlemi gösterilmektedir:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

catalog.properties ön ekine sahip parametreler hakkında bilgi için bkz. Flink CDC Katalog yapılandırma parametreleri.

filesystem Paimon kataloğuyla Object Storage Service (OSS) üzerine yazmaya ilişkin örnek yapılandırma:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

catalog.properties ön ekine sahip parametreler hakkında bilgi için bkz. Paimon FileSystem Kataloğu Oluşturma.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job 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: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # Synchronize all tables in the MySQL test_db database to the Paimon test_db2 database and retain the table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Paimon test_db.user table.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On job restart, check the new tables captured by the tables parameter and take a snapshot.
  # Note: Must be used together with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Tables that match this regular expression are not synchronized.
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  # Specify the file format of output files as lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline
  schema.change.behavior: evolve