Inner join v Flink s Debezium CRUD

Implementace inner join pro tabulky v Apache Flink s podporou CRUD operací

V systémech real-time datových vitrín na Apache Flink vyžaduje inner join mezi proudy zohlednění všech CRUD operací z Debezium. Předchozí přístup ignoroval mazání a aktualizace, což vedlo k nesprávným výsledkům v one-to-many relacích. Nová implementace parsuje operace a správně řídí stav.

Struktura zpráv Debezium určuje logiku: pole op určuje typ operace, before a after obsahují data před/po změně.

Parsování zpráv Debezium

Debezium generuje události v Kafka se fixní strukturou:

{
	"op": "(c|r|u|d)",
	"source": { ... },
	"ts_ms": "...",
	"before": [Data, null],
	"after": [Data, null]
}

Sémantika polí:

• c (create): before=null, after=Data
• r (read/snapshot): before=null, after=Data
• u (update): before=Data, after=Data
• d (delete): before=Data, after=null

Mapovací funkce extrahuje aktuální data v závislosti na op.

Mapování modelů dat

Třída Domain deserializuje Row z Flink s ohledem na typ operace:

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

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

        Row domain = (Row) row.getField(op == 'd' ? 0 : 2);

        Integer id = (Integer) domain.getField(0);
        Integer user_id = (Integer) domain.getField(1);
        String domain_name = (String) domain.getField(2);  // ispravleno on String

        return new Domain(id, user_id, domain_name, op == 'd');
    }
}

Logika výběru dat:

• Pro c/r — vezmeme after.
• Pro d — vždy before.
• Pro u — after (nuance v následující části).

Stejně pro model User.

KeyedCoProcessFunction pro inner join

Hlavní logika v InnerJoinFunction používá state pro ukládání souvisejících záznamů podle klíče (user_id):

• ValueState<User> pro aktuálního uživatele.
• MapState<Integer, Domain> pro domény uživatele.

Při příchodu User:

• Aktualizujeme state.
• Iterujeme všechny domény a generujeme Output pro každou.

Při příchodu Domain:

• Uložíme do MapState.
• Pokud uživatel existuje, generujeme Output.

Úplný kód:

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;

public class InnerJoinFunction extends KeyedCoProcessFunction<Integer, User, Domain, InnerJoinFunction.Output> {
    private MapState<Integer, Domain> domainsState;
    private ValueState<User> usersState;

    @Override
    public void processElement1(final User user, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        usersState.update(user);
      
        for (final Domain domain : domainsState.values()) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void processElement2(final Domain domain, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        domainsState.put(domain.id, domain);

        final User user = usersState.value();

        if (user != null) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void open(OpenContext openContext) throws Exception {
        var usersStateDescriptor = new ValueStateDescriptor<>(
                "users",
                User.class
        );
        var domainsStateDescriptor = new MapStateDescriptor<>(
                "domains",
                Integer.class,
                Domain.class
        );
        usersState = getRuntimeContext().getState(usersStateDescriptor);
        domainsState = getRuntimeContext().getMapState(domainsStateDescriptor);

        super.open(openContext);
    }

    public static class Output implements Serializable {
        public Integer user_id;
        public String firstname;
        public String lastname;
        public String domain_name;
        public boolean delete;
        
        // getters and setters omitted
    }
}

Omezení implementace

Aktuální přístup správně zpracovává CRUD, ale selhává při změně join-klíče (user_id) v update. State je vázán na klíč lokálně, retransmise do nové klíčové partition neprobíhá.

Řešení — v následující části.

Co je důležité:

• Parsování op z Debezium určuje výběr before/after.
• delete-vlajka v modelech zajišťuje mazání z vitríny.
• One-to-many: iterace po MapState generuje všechny kombinace.
• State-backend je kritický pro výkon při velkých objemech.
• Změna join-klíče vyžaduje speciální zpracování.

— Editorial Team