Flink real-time JOIN: przetwarzanie update za pomocą delete+insert

Obsługa aktualizacji w real-time JOIN na Apache Flink: rozwiązanie problemów z state

W systemach real-time JOIN na Apache Flink aktualizacje rekordów w źródłowych tabelach niosą ryzyko wycieków state i niepoprawnych wyników. Jeśli pole user_id w tabeli domains ulegnie zmianie, zdarzenie update może trafić na inny TaskManager, nie usuwając starych danych. Podobnie zmiana klucza głównego powoduje duplikaty w downstream bez usunięcia przestarzałych rekordów.

Rozważmy scenariusze:

• Zmiana user_id przy stałym domains.id: nowe zdarzenie jest haszowane po nowym kluczu, stare pozostaje w state.
• Zmiana domains.id i user_id: tworzony jest nowy rekord, poprzedni nie jest usuwany, co łamie deduplikację.

Rozproszona architektura Flink i lokalne strefy

Flink rozdziela dane po TaskManagerach za pomocą keyBy. Hash klucza określa właściciela-kontenera, zapewniając:

• Lokalizację wszystkich zdarzeń tego samego klucza.
• Zachowanie kolejności zdarzeń.

Przestrzeń hash jest równomiernie rozdzielana, każdy TaskManager odpowiada za swoją strefę. Bez keyBy stosowany jest round-robin, ale dla JOIN to nie działa — tracona jest gwarancja lokalności.

Uniwersalne rozwiązanie: delete + insert

Do obsługi update rozbijamy zdarzenia CDC z Debezium na delete (before) i insert (after). Kolejność jest kluczowa: najpierw delete po starym kluczu, potem insert po nowym. Gwarancje keyBy zapewniają sekwencyjną obróbkę.

Implementacja w klasie 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 zdarzenie delete na podstawie before.
• Update ('u'): delete na podstawie before + insert na podstawie after.
• Insert ('c'): tylko insert na podstawie after.

Delete oznacza rekord w state jako usunięty (lub go usuwa), insert aktualizuje. Dla zmieniającego się klucza JOIN lub PK zdarzenie samo czyści state.

Zalety podejścia

• Pełne wsparcie CRUD w strumieniu real-time.
• State bez wycieków i nadmiarowości.
• Poprawna praca z zmieniającymi się kluczami JOIN i PK.
• Zachowanie kolejności dzięki keyBy.

Co ważne

• Rozbicie update na delete+insert rozwiązuje wycieki state przy zmianie kluczy.
• Kolejność delete przed insert jest kluczowa dla spójności.
• Debezium before/after zapewnia dostęp do starych wartości.
• Podejście skaluje się w rozproszonej architekturze Flink.

— Editorial Team