Flink JOIN en temps réel : traitement de la mise à jour avec delete+insert

Gestion des mises à jour dans les JOIN en temps réel sur Apache Flink : Résoudre les fuites d'état

Dans les systèmes de JOIN en temps réel sur Apache Flink, les mises à jour des enregistrements de la table source peuvent provoquer des fuites d'état et des résultats incorrects. Si le champ user_id dans la table domains change, l'événement de mise à jour peut atterrir sur un TaskManager différent sans effacer les anciennes données. De même, modifier la clé primaire entraîne des doublons en aval sans supprimer les enregistrements obsolètes.

Décomposons les scénarios :

• Modification de user_id avec un domains.id fixe : Le nouvel événement est haché sur une clé différente, laissant l'ancienne en état.
• Modification de domains.id et user_id : Un nouvel enregistrement est créé, mais l'ancien n'est pas supprimé, ce qui casse la déduplication.

Architecture distribuée de Flink et zones locales

Flink distribue les données entre les TaskManagers via keyBy. Le hachage de la clé détermine le conteneur propriétaire, garantissant :

• Que tous les événements pour une clé sont localisés.
• Que l'ordre des événements est préservé.

L'espace de hachage est réparti uniformément, chaque TaskManager gérant sa zone. Sans keyBy, on retombe sur un round-robin, mais cela ne convient pas aux JOIN — cela perd les garanties de localité.

Solution universelle : Suppression + Insertion

Pour gérer les mises à jour, on décompose les événements CDC Debezium en suppression (avant) et insertion (après). L'ordre compte : suppression par ancienne clé d'abord, puis insertion par nouvelle clé. keyBy garantit un traitement séquentiel.

Implémentation dans la classe Domain :

public class Domain implements Serializable {
  public Integer id;
  public Integer user_id;
  public String domain_name;
  public boolean delete;

  // getters and setters omitted

  public static Domain[] fromRow(Row row) {
        Character op = (Character) row.getField(1);

        if (op == null) {
            throw new IllegalStateException("Never should happen, if Debezium feels fine");
        }

        if (op == 'd') {
            Row before = (Row) row.getField(0);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true)            };
        } else if (op == 'u') {
            Row before = (Row) row.getField(0);
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true),
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        } else {
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        }
    }
}

Logique :

• Suppression ('d') : Génère un événement de suppression à partir de l'avant.
• Mise à jour ('u') : Suppression de l'avant + insertion de l'après.
• Insertion ('c') : Insertion de l'après uniquement.

La suppression marque l'enregistrement comme retiré dans l'état (ou le supprime), tandis que l'insertion le met à jour. Pour les clés JOIN ou PK modifiables, l'événement nettoie lui-même l'état.

Avantages de cette approche

• Support complet CRUD dans les flux en temps réel.
• État sans fuites ni redondance.
• Gestion correcte des clés JOIN et PK modifiables.
• Ordre préservé via keyBy.

Points clés à retenir

• Décomposer les mises à jour en suppression+insertion corrige les fuites d'état lors des changements de clé.
• Supprimer avant d'insérer est crucial pour la cohérence.
• Les before/after de Debezium donnent accès aux anciennes valeurs.
• Scalable sans accroc dans l'architecture distribuée de Flink.

— Editorial Team