Module Transform de Flink CDC - Realtime Compute for Apache Flink

Paramètres des règles de transformation

Le module Transform permet de manipuler directement les colonnes de données. Supprimez ou étendez les colonnes existantes et filtrez les données indésirables lors de la synchronisation. Utilisez les paramètres suivants pour définir les règles de transformation :

Paramètre	Description	Obligatoire	Notes
`source-table`	Spécifie la table amont à transformer.	Oui	Prend en charge les expressions régulières.
`projection`	Spécifie la règle de projection pour la table amont. Cela détermine toutes les colonnes de données après la transformation.	Non	La syntaxe est similaire à l'instruction SQL SELECT. Si ce paramètre n'est pas spécifié, aucune colonne n'est ajoutée ni supprimée. Pour plus d'informations, consultez Filtrage des données. Pour obtenir des informations sur les fonctions intégrées disponibles, consultez Fonctions intégrées de Flink CDC.
`filter`	Règle de filtrage des lignes.	Non	La syntaxe est similaire à l'instruction SQL WHERE. Si ce paramètre n'est pas spécifié, aucune ligne n'est filtrée.
`primary-keys`	Spécifie les colonnes de clé primaire après la transformation.	Non	Si ce paramètre n'est pas spécifié, la définition de clé primaire du schéma d'origine est conservée. Les clés primaires sont séparées par des virgules (`,`). Important Par défaut, vous ne pouvez pas supprimer les colonnes de clé primaire de la source amont. Ajoutez le paramètre `transform.allow.trimming.pk-columns` à la configuration `pipeline:` pour réduire le nombre de colonnes de clé primaire amont. Lorsque vous définissez des colonnes de clé primaire personnalisées, assurez-vous que les données entrantes de la table amont respectent la contrainte PRIMARY KEY. Nous vous recommandons d'inclure les colonnes de clé primaire de la table amont dans les colonnes de clé primaire personnalisées afin d'éviter les problèmes de désordre des données lors des écritures inter-partitions.
`partition-keys`	Spécifie la liste des colonnes de clé de partition après la transformation.	Non	Si ce paramètre n'est pas spécifié, la définition de clé de partition du schéma d'origine est conservée. Les clés de partition sont séparées par des virgules (`,`). Important Lorsque vous définissez des colonnes de clé de partition personnalisées, assurez-vous que les données entrantes de la table amont respectent la contrainte PRIMARY KEY afin d'éviter les problèmes de désordre des données lors des écritures inter-partitions.
`table-options`	Transmet des informations de configuration supplémentaires au puits (sink).	Non	Une liste de propriétés facultatives, telles que le nombre de compartiments pour un puits Paimon ou des commentaires. Les éléments de configuration sont séparés par des virgules (`,`). La clé et la valeur de chaque élément sont séparées par un signe égal (`=`). Exemple de configuration : `key1=value1,key2=value2`
`description`	Description de la règle de transformation.	Non	Aucune.
`converter-after-transform`	Convertisseur qui effectue un traitement supplémentaire sur les données une fois la transformation terminée.	Non	Pour plus d'informations, consultez Convertisseur après transformation.

Remarques

Après avoir modifié une instruction dans le module Transform, redémarrez la tâche à partir d'un état initial sans état.
Généralement, les instructions projection et filter n'ont pas besoin d'être entourées de guillemets.
```
transform:
  - projection: a, b, c
    # Equivalent to
  - projection: "a, b, c"
```
Toutefois, si une expression de projection commence par un caractère spécial, tel que * ou ', l'expression entière risque de ne pas être analysée comme un littéral de chaîne YAML valide. Dans ce cas, entourez manuellement l'expression entière de guillemets simples (') ou doubles ("), ou utilisez une barre oblique inverse (\) pour échapper le caractère.
```
transform:
  - projection: *, 42      # Not valid YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```
À partir de Ververica Runtime (VVR) 11,6, vous pouvez purger les paramètres de clé primaire déduits de la source amont et convertir la table en une table sans clé primaire. La syntaxe est la suivante :
```
transform:
  - source-table: db.tbl
    primary-keys: '' # Clears the PRIMARY KEY setting for the db.tbl table
```

Filtrage des données

Le module Transform utilise une syntaxe de type SQL pour définir les règles de filtrage des données (projection). Ces règles vous permettent de sélectionner des colonnes spécifiques, d'ajouter des colonnes calculées et d'ajouter des colonnes de métadonnées.

Élagage des colonnes

Pour extraire des colonnes spécifiques de la table source et les synchroniser vers l'aval, spécifiez les colonnes souhaitées dans la règle de projection. Les colonnes non spécifiées ne sont pas envoyées vers l'aval.

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 is pruned

Important

L'élagage des colonnes peut entraîner une incohérence entre les schémas des tables amont et aval si le schéma de la table amont change.

Caractères génériques

Pour envoyer toutes les colonnes de la table source, y compris les nouvelles colonnes ajoutées ultérieurement, vers l'aval sans modification, utilisez le caractère générique astérisque (*) dans la règle de projection.

Remarque

Si une règle de projection n'utilise pas de caractère générique (*), le schéma résultant est fixe. Le schéma reste cohérent avec la version spécifiée dans la règle de projection et l'évolution du schéma ne peut pas être synchronisée.

Par exemple, *, 'extras' AS extras indique qu'une colonne supplémentaire est ajoutée après les colonnes du schéma amont. L'évolution du schéma amont est toujours synchronisée vers l'aval.

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

Colonnes calculées

Utilisez la syntaxe <Expression> AS <ColName> dans une règle de projection pour ajouter une colonne calculée. L'expression est évaluée pour chaque enregistrement provenant de la source amont et le résultat remplit la colonne correspondante.

Remarque

L'expression d'une colonne calculée ne peut pas référencer la valeur d'une autre colonne calculée, même si la colonne référencée apparaît avant elle. Par exemple, a, b AS c, c AS d n'est pas une expression valide.

Par exemple, lorsque la table amont db.tbl envoie l'enregistrement de données [+I, id = 1], l'enregistrement est transformé en la ligne de données [+I, id = 1, inc_id = 2] puis envoyé vers l'aval.

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

Colonnes de métadonnées

Lorsque vous écrivez des règles de projection, vous pouvez utiliser les colonnes de métadonnées prédéfinies suivantes de la même manière que les colonnes de données classiques :

Important

Ne définissez pas de colonnes de données classiques portant les mêmes noms que les colonnes de métadonnées.

Les colonnes de métadonnées suivantes s'appliquent à tous les connecteurs.

Nom de la colonne de métadonnées	Type de données	Description
`__namespace_name__`	String	Nom de l'espace de noms de la table source pour cet enregistrement de modification.
`__schema_name__`	String	Nom du schéma de la table source pour cet enregistrement de modification.
`__table_name__`	String	Nom de la table de la table source pour cet enregistrement de modification.
`__data_event_type__`	String	Type d'opération pour cet enregistrement de modification (`+I`, `-U`, `+U` ou `-D`). Important Un événement CDC regroupe toujours les enregistrements Update Before et Update After d'une mise à jour en un seul événement. Par conséquent, le contenu de `__data_event_type__` est respectivement `-U` et `+U` au sein du même événement de mise à jour. Ne l'utilisez pas comme clé primaire.

Par exemple, vous pouvez écrire le nom qualifié complet de la table amont dans une colonne calculée et l'envoyer vers l'aval.

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

Le tableau suivant présente les relations de mappage entre les connecteurs de base de données et les noms Namespace, Schema et Table.

Type de base de données	Nom de l'espace de noms	Nom du schéma	Nom de la table
MySQL	-	Base de données	Table
Kafka	-	-	Topic
SLS	-	Projet	LogStore
MongoDB	-	Base de données	Collection
Paimon	-	Base de données	Table
Hologres	-	Schéma	Table
StarRocks	-	Base de données	Table
Doris	-	Base de données	Table
Postgres	Base de données Remarque Cela n'est effectif que lorsque le paramètre `table-id.include-database` est activé.	Schéma	Table

Filtrage des données

Le module Transform utilise une syntaxe de type SQL pour définir les règles de filtrage des lignes.

La règle de filtre doit être une expression qui renvoie une valeur BOOLEAN. Elle peut référencer n'importe quelle colonne de la table source et n'importe quelle colonne calculée.

Si un enregistrement de modification correspond à une règle de transformation avec un filtre et que l'expression de filtre est évaluée à FALSE, la ligne de données n'est pas envoyée vers l'aval.

Remarque

Si vous utilisez une colonne calculée dans une règle de projection pour remplacer une colonne amont existante, l'expression de filtre référence la colonne calculée.

Par exemple, la règle de transformation suivante est valide :

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

Le id référencé dans l'expression filter est la colonne calculée convertie au type VARCHAR.

Correspondance de motifs de chaîne

Utilisez LIKE et NOT LIKE dans les expressions de filtre pour effectuer une correspondance de motifs de chaîne. L'implémentation utilise la sémantique find() des expressions régulières Java, qui diffère de la norme SQL LIKE de deux manières importantes :

• Le motif est une expression régulière Java. % et _ sont traités comme des littéraux regex plutôt que comme des caractères génériques.

• La correspondance utilise la sémantique de sous-chaîne. Le motif doit seulement apparaître à une certaine position dans la chaîne pour renvoyer TRUE ; une correspondance sur toute la chaîne n'est pas requise.

Exemple : Filtrer les enregistrements dont la colonne code commence par HSRS.

Le motif suivant ne correspond pas correctement — ABC_HSRS123 est également mis en correspondance :

transform:
  - source-table: db.orders
    filter: code LIKE 'HSRS.*'

Le motif correct utilise l'ancre de début ^ :

transform:
  - source-table: db.orders
    filter: code LIKE '^HSRS'

Les équivalents de motifs courants sont les suivants :

Syntaxe SQL	Regex CDC équivalente	Signification
`col LIKE 'HSRS%'`	`col LIKE '^HSRS'`	Commence par HSRS
`col LIKE '%HSRS'`	`col LIKE 'HSRS$'`	Se termine par HSRS
`col LIKE '%HSRS%'`	`col LIKE 'HSRS'`	Contient HSRS
`col LIKE 'HSRS'`	`col LIKE '^HSRS$'`	Égal exactement à HSRS

Pour la sémantique complète de LIKE et d'autres équivalents regex, consultez Fonctions intégrées de Flink CDC.

Règles de configuration avancées

Référence des colonnes non élaguées et des métadonnées

Exemple 1 : Filtrer en fonction d'une colonne élaguée

Supposons que le schéma de la table amont soit [INT a, INT b, BOOLEAN c]. Pour produire les colonnes a et b et ne conserver que les lignes où c est true, utilisez la règle de configuration suivante :

transform:
  - source-table: ...
    projection: a, b
    filter: c

Exemple 2 : Filtrer en fonction des colonnes de métadonnées

Vous pouvez utiliser directement les colonnes de métadonnées comme conditions de filtre sans les définir explicitement dans la projection. Par exemple, vous pouvez exclure les données de modification de type DELETE.

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

Remplacement des colonnes existantes

Dans la projection, vous pouvez définir un champ portant le même nom qu'une colonne amont pour remplacer la valeur ou le type de cette colonne. Cette méthode garantit que l'évolution du schéma est correctement synchronisée.

Exemple : Forcer la conversion de type

Supposons que le schéma de la table amont soit [INT a, INT b, BOOLEAN c]. Pour conserver les noms de colonnes inchangés mais forcer la conversion de la colonne a en type chaîne, utilisez la configuration suivante :

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

Le schéma de la table aval devient [STRING a, INT b, BOOLEAN c]. La colonne d'origine a est remplacée par la nouvelle définition.

Réutilisation des colonnes calculées dans les conditions de filtre

Un filter peut référencer directement l'alias d'une colonne calculée définie dans une projection.

Exemple : Référence d'un résultat calculé

Vous pouvez définir une nouvelle colonne d dans la projection et l'utiliser directement dans le filter.

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

Cette configuration a le même effet que la suivante, mais est plus lisible :

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Convertisseur après transformation

converter-after-transform est utilisé pour traiter les modifications de données après l'application de toutes les règles de transformation. Spécifiez plusieurs convertisseurs en les séparant par des virgules (,). Les convertisseurs sont appliqués dans l'ordre où ils sont répertoriés. Les valeurs de configuration suivantes sont prises en charge.

Nom du convertisseur

Fonctionnalité

Versions prises en charge

SOFT_DELETE

Convertit les modifications de suppression en insertions.

VVR 8.0.11 et versions ultérieures.

FIELD_NAME_LOWER_CASE

Convertit tous les noms de champs de table en minuscules.

VVR 11.1 et versions ultérieures.

Suppression logique (Soft delete)

Le convertisseur SOFT_DELETE fonctionne avec la colonne de métadonnées __data_event_type__ pour implémenter des suppressions logiques. Cela signifie que les données ne sont pas physiquement supprimées dans le système aval. Par exemple, la configuration de transformation suivante implémente une suppression logique. Une opération de suppression est convertie en insertion et le op_type des données correspondantes est mis à jour à -D pour indiquer la suppression.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

Le convertisseur SOFT_DELETE transforme uniquement les données dont __data_event_type__ est -D. Pour ignorer également les données dont __data_event_type__ est -U, ajoutez la configuration filter: op_type <> '-U'.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    filter: op_type <> '-U'
    converter-after-transform: SOFT_DELETE