Flink real-time JOIN: zpracování update s delete+insert

Zpracování aktualizací v real-time JOIN na Apache Flink: Řešení problémů se stavy

V systémech real-time JOIN na Apache Flink aktualizace záznamů v zdrojových tabulkách vytvářejí rizika úniků stavů a nesprávných výsledků. Pokud se změní pole user_id v tabulce domains, událost update může skončit na jiném TaskManageru, aniž by se vyčistila stará data. Stejně tak změna primárního klíče vede k duplicitám v downstream bez odstranění zastaralých záznamů.

Projdeme scénáře:

• Změna user_id při pevném domains.id: nová událost se hashuje podle nového klíče, starý záznam zůstává ve stavu.
• Změna domains.id a user_id: vytvoří se nový záznam, předchozí se neodstraní, což narušuje deduplikaci.

Rozložená architektura Flink a lokální zóny

Flink distribuuje data mezi TaskManagery prostřednictvím keyBy. Hash klíče určuje vlastnický kontejner, což zajišťuje:

• Lokalizaci všech událostí jednoho klíče.
• Zachování pořadí událostí.

Hashový prostor je rovnoměrně rozdělen, každý TaskManager odpovídá za svou zónu. Bez keyBy se používá round-robin, ale pro JOIN to nejde – ztrácí se záruka lokalizace.

Univerzální řešení: delete + insert

Pro zpracování update rozdělujeme CDC události Debezium na delete (before) a insert (after). Pořadí je klíčové: nejdříve delete podle starého klíče, pak insert podle nového. Záruky keyBy zajišťují sekvenční zpracování.

Implementace v třídě 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)
            };
        }
    }
}

Logika:

• Delete ('d'): generuje delete událost podle before.
• Update ('u'): delete podle before + insert podle after.
• Insert ('c'): pouze insert podle after.

Delete označí záznam ve stavu jako smazaný (nebo ho odstraní), insert aktualizuje. Pro měnící se klíč JOIN nebo PK událost sama vyčistí stav.

Výhody přístupu

• Plná podpora CRUD v real-time proudu.
• Stav bez úniků a nadbytečnosti.
• Správná práce s měnícími se klíči JOIN a PK.
• Zachování pořadí prostřednictvím keyBy.

Co je důležité

• Rozdělení update na delete+insert řeší úniky stavů při změně klíčů.
• Pořadí delete před insert je klíčové pro sekvenčnost.
• Debezium before/after zajišťuje přístup ke starým hodnotám.
• Přístup je škálovatelný na rozloženou architekturu Flink.

— Editorial Team