Aller au contenu
Ingestion

Capture des changements (CDC)

La CDC lit le journal de transactions de la base source pour émettre chaque insertion, modification et suppression. On l'utilise pour répliquer en quasi temps réel sans requêter les tables.

  • change data capture
  • debezium
  • wal
  • binlog
  • journal
  • réplication
  • op

Syntaxe

Modèle
Événement de changement (forme courante, inspirée de Debezium) :{  "op": "c" | "u" | "d" | "r",   création, mise à jour, suppression, lecture du snapshot  "before": { … } | null,        état avant (null pour une création)  "after":  { … } | null,        état après (null pour une suppression)  "lsn": 104,                    position dans le journal de la base  "ts_ms": 1793520000000         horodatage du changement}Application sur la cible, dans l'ordre du journal et par clé :  c, r → INSERT     u → UPDATE (ou upsert)     d → DELETE

Exemple

Exemple concret
# Rejouer un flux CDC sur une table cible (ici un dict indexé par id)evenements = [    {"op": "c", "lsn": 101, "after": {"id": 7, "ville": "Lille"}},    {"op": "u", "lsn": 102, "before": {"id": 7, "ville": "Lille"},                            "after": {"id": 7, "ville": "Rennes"}},    {"op": "c", "lsn": 103, "after": {"id": 8, "ville": "Nice"}},    {"op": "d", "lsn": 104, "before": {"id": 8, "ville": "Nice"}},    # après une reprise, le connecteur renvoie un événement déjà traité    {"op": "u", "lsn": 102, "before": {"id": 7, "ville": "Lille"},                            "after": {"id": 7, "ville": "Rennes"}},]cible = {}dernier_lsn = 0for e in evenements:    if e["lsn"] <= dernier_lsn:        continue                      # déjà appliqué : on l'ignore    if e["op"] in ("c", "r", "u"):        cible[e["after"]["id"]] = e["after"]    elif e["op"] == "d":        cible.pop(e["before"]["id"], None)    dernier_lsn = e["lsn"]print(cible)# {7: {'id': 7, 'ville': 'Rennes'}}# Le client 8, créé puis supprimé, n'existe plus dans la cible.

À retenir

  1. Il existe trois familles : lecture du journal (WAL de PostgreSQL, binlog de MySQL), interrogation périodique d'une colonne de date, et déclencheurs. Seule la lecture du journal voit les suppressions et les états intermédiaires sans charger la source.
  2. Une CDC commence par un instantané initial (événements r) de l'existant, puis enchaîne sur le journal à partir de la position où l'instantané s'est arrêté.
  3. Publiez les événements dans Kafka avec la clé primaire comme clé de message : tous les changements d'une même ligne tombent dans la même partition et gardent leur ordre.
  4. Piège : supposer que chaque événement arrive une seule fois. Après une panne, le connecteur reprend au dernier point confirmé et renvoie des événements ; comparez la position (lsn) ou appliquez des upserts pour que le rejeu ne change rien.

Voir aussi