Flink CDC job structure - Realtime Compute for Apache Flink - Centre de documentation Alibaba Cloud

Aperçu de la structure du job

Un fichier de job complet respecte la structure suivante :

source:         # Required: defines the data source
  ...

sink:           # Required: defines the destination
  ...

transform:      # Optional: filters and reshapes data before writing
  - ...

route:          # Optional: maps source tables to different sink tables
  - ...

pipeline:       # Optional: sets global job parameters
  ...

Configuration minimale

L’exemple suivant synchronise toutes les tables correspondantes de MySQL vers Paimon en utilisant uniquement les modules obligatoires :

# Source module
source:
  type: mysql
  name: MySQL Source
  host: localhost
  port: 3306
  username: admin
  password: <yourPassword>
  tables: adb.*

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

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

Configuration complète

L’exemple suivant ajoute les modules Transform et Route pour filtrer les champs, appliquer des expressions et rediriger les tables :

# 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.\\.*

# 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

Module Source

Le module Source définit l’origine des données lues par Flink CDC. Connecteurs pris en charge :

source:
  type: mysql      # Connector type
  name: mysql source
  # Connector-specific parameters follow
  xxx: ...

Pour la liste complète des paramètres pris en charge, consultez la documentation du connecteur correspondant.

Module Sink

Le module Sink définit la destination d’écriture des données par Flink CDC. Connecteurs pris en charge :

sink:
  type: hologres   # Connector type
  name: hologres sink
  # Connector-specific parameters follow
  xxx: ...

Pour la liste complète des paramètres pris en charge, consultez la documentation du connecteur correspondant.

Module Transform

Le module Transform applique des règles de projection, de calcul et de filtrage aux données des tables sources avant leur écriture dans le sink. Définissez une règle par modèle de table source.

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

Paramètre	Description	Obligatoire
`source-table`	Identifiant de la table source ; prend en charge les expressions régulières	Obligatoire
`projection`	Sélection de colonnes et expressions, similaires à la clause SQL `SELECT`	Facultatif
`filter`	Condition de filtrage des lignes, similaire à la clause SQL `WHERE`	Facultatif
`primary-keys`	Clés primaires de la table sink, séparées par des virgules	Facultatif
`partition-keys`	Clés de partition de la table sink, séparées par des virgules	Facultatif
`table-options`	Options de création de table appliquées lors de la création automatique de la table sink	Facultatif
`converter-after-transform`	Convertisseur appliqué aux événements de modification après l’exécution de la règle de transformation	Facultatif
`description`	Description de la règle	Facultatif

Pour plus de détails sur la syntaxe des expressions, consultez la rubrique Module Flink CDC Transform.

Module Route

Le module Route mappe les tables sources aux tables sinks. Utilisez-le pour fusionner des tables fragmentées (sharded), ajouter des préfixes aux noms de tables ou rediriger des tables vers différentes destinations.

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

Paramètre	Description	Obligatoire
`source-table`	Identifiant de la table source ; prend en charge les expressions régulières	Obligatoire
`sink-table`	Identifiant de la table sink cible	Obligatoire
`description`	Description de la règle de routage	Facultatif

Pour plus de détails sur la syntaxe des expressions, consultez la rubrique Module Flink CDC Route.

Module Pipeline

Le module Pipeline définit les paramètres globaux applicables à l’ensemble du job.

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

Paramètre	Description	Obligatoire
`name`	Nom du job affiché dans le tableau de bord Flink	Facultatif
`schema.change.behavior`	Mode de gestion des modifications de schéma en amont par le job. Valeurs acceptées : `evolve`, `LENIENT` et autres.	Facultatif

Pour la liste complète des paramètres configurables, consultez la rubrique Module Flink CDC Pipeline.

Realtime Compute for Apache Flink:Flink CDC job structure