Connect Flink CDC to a DLF catalog - Data Lake Formation - Centre de documentation Alibaba Cloud

Flink Change Data Capture (CDC) est un outil d'ingestion de données proposé par Realtime Compute for Apache Flink. Il permet de synchroniser des bases de données entières depuis des sources vers votre data lakehouse. Cette rubrique vous guide dans l'utilisation de Flink CDC pour ingérer des données en temps réel dans un catalogue DLF via Paimon REST.

À la fin de cette rubrique, vous aurez :

Créé un catalogue DLF en tant que destination d'ingestion
Configuré une tâche YAML Flink CDC avec un sink Paimon pointant vers DLF
Exécuté des exemples pour les scénarios d'ingestion courants : synchronisation complète de base de données, tables partitionnées, tables append-only et sources Kafka

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un espace de travail Realtime Compute for Apache Flink. Consultez Créer un espace de travail.
L'espace de travail Realtime Compute for Apache Flink et les catalogues DLF doivent se trouver dans la même région.
Le VPC de votre espace de travail Realtime Compute for Apache Flink doit être ajouté à la liste d'autorisation VPC de DLF. Consultez Configurer une liste d'autorisation VPC.

Exigences du moteur

Votre tâche Realtime Compute for Apache Flink doit s'exécuter sur la version 11.1.0 ou ultérieure de Ververica Runtime (VVR).

Créer un catalogue DLF

Consultez Prise en main de DLF.

Créer et configurer une tâche d'ingestion de données

Créez un brouillon YAML pour ingérer des données à l'aide de Flink CDC. Pour plus d'informations, consultez Développer des tâches Flink CDC pour l'ingestion de données (Bêta).

Configurez le module sink :

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

Remplacez les valeurs d'espace réservé par vos valeurs réelles :

Option de configuration	Description	Obligatoire	Par défaut	Exemple de valeur
`catalog.properties.metastore`	Le type de metastore. Définissez-le sur `rest`.	Oui	—	`rest`
`catalog.properties.token.provider`	Le fournisseur de jetons. Définissez-le sur `dlf`.	Oui	—	`dlf`
`catalog.properties.uri`	L'URI pour accéder au serveur de catalogue REST DLF. Format : `http://[region-id]-vpc.dlf.aliyuncs.com`. Pour les ID de région, consultez Régions et endpoints.	Oui	—	`http://ap-southeast-1-vpc.dlf.aliyuncs.com`
`catalog.properties.warehouse`	Le nom du catalogue Paimon.	Oui	—	`dlf_test`
`commit.user`	L'utilisateur de validation pour les écritures de données. Attribuez des utilisateurs de validation uniques à des tâches distinctes pour éviter les conflits.	Non	`admin`	`your_job_name`
`table.properties.deletion-vectors.enabled`	Active les vecteurs de suppression pour améliorer les performances de lecture avec un impact minimal sur les écritures.	Non	—	`true`

Notes d'utilisation

Avant d'exécuter une tâche, tenez compte des contraintes suivantes :

Conflits d'utilisateurs de validation : L'utilisateur de validation par défaut est admin. L'exécution simultanée de tâches d'écriture de données vers la même table sous le même utilisateur de validation entraîne des conflits de validation et des incohérences de données. Attribuez un commit.user unique à chaque tâche.
Aucune option de compactage ou de bucket : DLF fournit un compactage automatique des fichiers. Ne configurez pas d'options de compactage ou de bucket telles que bucket et num-sorted-run.compaction-trigger.
Vecteurs de suppression : Définissez table.properties.deletion-vectors.enabled: true pour accélérer considérablement les lectures avec un impact minimal sur les écritures.

Exemples

Ingérer des données depuis une base de données MySQL entière vers DLF

La tâche suivante synchronise toutes les tables d'une base de données MySQL vers DLF :

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

Les options recommandées pour la source MySQL sont les suivantes :

Option	Description
`scan.binlog.newly-added-table.enabled`	Synchronise les données des tables créées pendant la phase incrémentielle.
`include-comments.enabled`	Synchronise les commentaires de table et de champ.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Empêche les erreurs potentielles d'épuisement de la mémoire (OOM) du TaskManager en donnant la priorité aux shards non bornés.
`scan.only.deserialize.captured.tables.changelog.enabled`	Désérialise les données uniquement à partir des tables correspondantes pour accélérer les lectures.

Ingérer des données dans une table partitionnée

Pour ingérer des données depuis une table source non partitionnée vers une table partitionnée dans DLF, ajoutez l'option partition-keys au module transform. Pour plus d'informations, consultez Ingestion de données avec Flink CDC.

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

Ingérer des données dans une table append-only

Pour implémenter des suppressions logiques (soft delete) lors de l'ingestion, utilisez converter-after-transform: SOFT_DELETE dans le module transform. Cela convertit les opérations de suppression en opérations d'insertion, de sorte que la table en aval enregistre toutes les opérations de modification de manière complète. Le champ __data_event_type__ dans projection écrit le type de modification sous forme de nouvelle colonne dans la table en aval.

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

Pour plus d'informations, consultez Ingestion de données avec Flink CDC.

Synchroniser des données depuis Kafka vers DLF en temps réel

La tâche suivante lit les données CDC depuis le topic Kafka inventory (tables customers et products au format JSON Debezium) et les synchronise vers les tables de destination correspondantes dans DLF. Étant donné que les messages JSON Debezium ne contiennent pas d'informations sur la clé primaire, celle-ci est explicitement spécifiée dans le module transform.

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

Notes supplémentaires pour les sources Kafka :

La source Kafka prend en charge trois formats de données : canal-json, debezium-json (par défaut) et json.
Lors de l'ingestion depuis plusieurs partitions Kafka vers une seule table dans DLF, définissez debezium-json.distributed-tables ou canal-json.distributed-tables sur true.
La source Kafka prend en charge plusieurs stratégies d'inférence de schéma via l'option schema.inference.strategy. Pour plus d'informations, consultez Message Queue for Apache Kafka.

Pour plus d'informations, consultez Ingestion de données avec Flink CDC.