Distribute data in real time with Flink CDC and Kafka - Realtime Compute for Apache Flink

Ce guide explique comment configurer deux pipelines Flink CDC pour transférer des données de MySQL vers un lac de données en temps réel.

Ce guide détaille la mise en place de deux pipelines Flink CDC permettant le transfert en temps réel des données de MySQL vers un lac de données :

Pipeline 1 — De MySQL vers Kafka : capturez les journaux binaires de MySQL et diffusez les événements de modification vers des topics ApsaraMQ for Kafka.
Pipeline 2 — De Kafka vers DLF : lisez ces événements de modification depuis Kafka et ingérez-les dans Data Lake Formation (DLF) via un connecteur Paimon.

MySQL (binlog) → [Pipeline 1] → ApsaraMQ for Kafka → [Pipeline 2] → DLF (Paimon)

Kafka sert de tampon durable et rejouable entre les deux tâches, découplant la capture de l'ingestion et permettant une mise à l'échelle, une auditabilité et une relecture indépendantes.

Prérequis

La journalisation binaire de MySQL est activée au format ROW.
Un cluster ApsaraMQ for Kafka est disponible et accessible depuis la tâche Flink.

Pipeline 1 : Diffuser les journaux binaires MySQL vers Kafka

Vue d'ensemble

Ce pipeline se connecte à une instance MySQL, lit les événements de modification des journaux binaires de toutes les tables de la base de données kafka_test et les écrit dans des topics ApsaraMQ for Kafka. Chaque table source correspond à son propre topic Kafka.

Configuration minimale

La configuration suivante représente le minimum requis pour exécuter le pipeline :

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: kafka_test\.\*
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  properties.enable.idempotence: false

Avertissement

Définissez properties.enable.idempotence: false dans la configuration du sink. ApsaraMQ for Kafka ne prend pas en charge les écritures idempotentes ou transactionnelles.

Configuration complète avec paramètres optionnels

La configuration ci-dessous étend la configuration minimale avec des paramètres optionnels. Ajoutez uniquement les paramètres pertinents pour votre cas d'utilisation.

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: kafka_test\.\*
  server-id: 8601-8604
  # (Optional) Sync existing and incremental data from newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Sync table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize distributing unbounded chunks during snapshot.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize changelog only for captured tables.
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # (Optional) Include the data change timestamp as a metadata column.
  metadata-column.include-list: op_ts

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  # Required: ApsaraMQ for Kafka does not support idempotent or transactional writes.
  properties.enable.idempotence: false
  # (Optional) Set the mapping between upstream tables and Kafka topics.
  sink.tableId-to-topic.mapping: kafka_test.customers:customers;kafka_test.products:products

Nommage des topics et mappage

Par défaut, chaque table source est associée à un topic Kafka nommé selon le format database.table. Par exemple, les données de kafka_test.customers sont écrites dans un topic nommé kafka_test.customers.

Pour utiliser des noms de topics personnalisés, définissez sink.tableId-to-topic.mapping. Cette option mappe les identifiants des tables sources aux noms des topics Kafka tout en conservant le nom de table d'origine dans la charge utile de l'événement pour la traçabilité. L'exemple ci-dessus associe kafka_test.customers à un topic nommé customers et kafka_test.products à un topic nommé products.

ApsaraMQ for Kafka prend en charge les formats de sortie json, canal-json et debezium-json (par défaut).

Pour obtenir la liste complète des options de configuration, consultez Connecteur Kafka.

Stratégie de partitionnement

Par défaut, tous les événements de modification sont écrits dans la partition 0 de chaque topic (all-to-zero). Cela préserve l'ordre global au sein d'un topic mais ne répartit pas la charge sur plusieurs partitions.

Pour répartir les enregistrements sur plusieurs partitions en fonction du hachage de la clé primaire, ajoutez ce qui suit à votre configuration de sink :

sink:
  partition.strategy: hash-by-key

Avec hash-by-key, tous les événements de modification pour une valeur de clé primaire donnée vont toujours vers la même partition, préservant ainsi l'ordre par clé tout en permettant une consommation parallèle.

Résultat attendu

Lorsque vous exécutez une instruction UPDATE sur la table customers, la tâche produit un message Kafka au format configuré.

Format Debezium JSON (debezium-json, par défaut) :

{
  "before": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "2222",
    "age": 12
  },
  "after": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "1234",
    "age": 12
  },
  "op": "u",
  "source": {
    "db": "kafka_test",
    "table": "customers",
    "ts_ms": 1728528674000
  }
}

Format Canal JSON (canal-json) :

{
  "old": [
    {
      "id": 4,
      "name": "John",
      "address": "New York",
      "phone_number": "2222",
      "age": 12
    }
  ],
  "data": [
    {
      "id": 4,
      "name": "John",
      "address": "New York",
      "phone_number": "1234",
      "age": 12
    }
  ],
  "type": "UPDATE",
  "database": "kafka_test",
  "table": "customers",
  "pkNames": [
    "id"
  ],
  "ts": 1728528674000,
  "es": 0
}

Remarque

Le format Debezium JSON omet les informations de clé primaire de la charge utile du message. Si un consommateur en aval doit identifier les clés primaires, utilisez le format Canal JSON ou définissez-les explicitement dans la section transform d'un pipeline en aval (comme illustré dans le Pipeline 2 ci-dessous).

Pipeline 2 : Ingérer les événements de modification Kafka dans DLF

Vue d'ensemble

Ce pipeline lit les événements de modification depuis un topic Kafka et les écrit dans Data Lake Formation (DLF) à l'aide d'un connecteur Paimon. Il est conçu pour consommer la sortie produite par le Pipeline 1.

Prenons l'exemple d'un topic Kafka inventory contenant des données de journaux binaires debezium-json provenant de deux tables, customers et products. Lorsqu'un seul topic Kafka transporte des événements de modification provenant de plusieurs tables sources, vous devez définir debezium-json.distributed-tables: true afin que le connecteur puisse acheminer correctement chaque enregistrement vers sa table de destination.

Configuration

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

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The username for commits. Set a unique 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

Détails clés de la configuration

Paramètre	Notes
`debezium-json.distributed-tables`	Définissez cette option sur `true` lorsqu'un topic Kafka contient des enregistrements provenant de plus d'une table source. Sans cela, le connecteur ne peut pas déterminer à quelle table de destination appartient chaque enregistrement. Pour le format Canal JSON, utilisez plutôt `canal-json.distributed-tables: true`.
`transform` / `primary-keys`	Le format Debezium JSON omet les informations de clé primaire de la charge utile du message. Vous devez définir les clés primaires dans la section `transform` afin que Paimon puisse effectuer correctement les upserts. L'exemple applique `id` comme clé primaire à toutes les tables correspondant à `\..\.`.
`scan.startup.mode: earliest-offset`	Lit depuis le début de chaque topic Kafka au démarrage.
`schema.inference.strategy`	Définissez une stratégie d'inférence de schéma pour la source Kafka avec l'option `schema.inference.strategy`. Pour plus d'informations, consultez Connecteur Kafka.

Paramètre

Notes

debezium-json.distributed-tables

Définissez cette option sur true lorsqu'un topic Kafka contient des enregistrements provenant de plus d'une table source. Sans cela, le connecteur ne peut pas déterminer à quelle table de destination appartient chaque enregistrement. Pour le format Canal JSON, utilisez plutôt canal-json.distributed-tables: true.

transform / primary-keys

Le format Debezium JSON omet les informations de clé primaire de la charge utile du message. Vous devez définir les clés primaires dans la section transform afin que Paimon puisse effectuer correctement les upserts. L'exemple applique id comme clé primaire à toutes les tables correspondant à \.*.\.*.

scan.startup.mode: earliest-offset

Lit depuis le début de chaque topic Kafka au démarrage.

schema.inference.strategy

Définissez une stratégie d'inférence de schéma pour la source Kafka avec l'option schema.inference.strategy. Pour plus d'informations, consultez Connecteur Kafka.

Notes d'utilisation

Remarque

Le connecteur source Kafka prend en charge la lecture des données aux formats canal-json et debezium-json (par défaut).
Définissez debezium-json.distributed-tables: true (ou canal-json.distributed-tables: true) lorsque les données de plusieurs tables sources sont réparties sur les partitions d'un même topic, ou lorsque vous devez fusionner des données distribuées.
Étant donné que Debezium JSON omet les informations de clé primaire, vous devez toujours définir explicitement les clés primaires dans le module transform lors de l'utilisation de ce format avec un sink Paimon.