Flink Echtzeit-JOIN: Verarbeitung von Update mit delete+insert

Updates in Echtzeit-JOINs mit Apache Flink: State-Leak-Probleme lösen

In Apache-Flink-Echtzeit-JOIN-Systemen können Updates von Quelldatentabelle-Einträgen zu State-Leaks und falschen Ergebnissen führen. Wenn sich das Feld user_id in der domains-Tabelle ändert, landet das Update-Ereignis möglicherweise auf einem anderen TaskManager, ohne die alten Daten zu löschen. Ähnlich führt eine Änderung des Primärschlüssels zu Duplikaten downstream, da veraltete Einträge nicht entfernt werden.

Lassen Sie uns die Szenarien aufschlüsseln:

• Änderung von user_id bei festem domains.id: Das neue Ereignis hasht zu einem anderen Schlüssel und lässt den alten im State.
• Änderung beider domains.id und user_id: Es wird ein neuer Eintrag erstellt, der alte wird aber nicht entfernt – Deduplizierung bricht zusammen.

Flinks verteilte Architektur und lokale Zonen

Flink verteilt Daten über TaskManager mit keyBy. Der Hash des Schlüssels bestimmt den zuständigen Container und gewährleistet:

• Alle Ereignisse eines Schlüssels sind lokalisiert.
• Die Reihenfolge der Ereignisse bleibt erhalten.

Der Hash-Raum ist gleichmäßig verteilt, jeder TaskManager bearbeitet seine Zone. Ohne keyBy greift Flink auf Round-Robin zurück – ungeeignet für JOINs, da Lokalisierungs-Garantien fehlen.

Universelle Lösung: Delete + Insert

Um Updates zu handhaben, zerlegen wir Debezium CDC-Ereignisse in delete (before) und insert (after). Die Reihenfolge zählt: Zuerst delete mit altem Schlüssel, dann insert mit neuem. keyBy garantiert sequenzielle Verarbeitung.

Implementierung in der Domain-Klasse:

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)
            };
        }
    }
}

Logik:

• Delete ('d'): Erzeugt Delete-Ereignis aus before.
• Update ('u'): Delete aus before + Insert aus after.
• Insert ('c'): Nur Insert aus after.

Delete markiert den Eintrag als gelöscht im State (oder löscht ihn), Insert aktualisiert ihn. Bei veränderlichen JOIN- oder PK-Schlüsseln reinigt das Ereignis den State selbst.

Vorteile dieses Ansatzes

• Vollständige CRUD-Unterstützung in Echtzeit-Streams.
• Leak-freier, nicht-redundanter State.
• Korrekte Handhabung veränderlicher JOIN- und PK-Schlüssel.
• Reihenfolge via keyBy erhalten.

Wichtige Erkenntnisse

• Zerlegung von Updates in delete+insert behebt State-Leaks bei Schlüsseländerungen.
• Delete vor insert ist entscheidend für Konsistenz.
• Debeziums before/after liefert Zugriff auf alte Werte.
• Skaliert nahtlos in Flinks verteilter Umgebung.

— Editorial Team