Flink Echtzeit-JOIN: Tabellen 4+2 TB

Implementierung von Echtzeit-JOIN für große Tabellen auf Apache Flink: Grundlegender Ansatz

Wir müssen Echtzeit-Updates für das Ergebnis des INNER JOIN zwischen der users-Tabelle (4 TB) und domains (2 TB) ermöglichen. Die Abfrage liefert user_id, firstname, lastname und domain_name. Traditionelle Views belasten die Datenquelle (OLTP), während eine einfache Materialisierung über CDC in DWH nicht die geforderte Geschwindigkeit bietet.

Beispieldaten:

users:

| id | firstname | lastname |

|----|-----------|----------|

| 1 | Egor | Myasnik |

| 2 | Pavel | Hvastun |

| 3 | Mitya | Volk |

domains:

| id | user_id | domain_name |

|----|---------|-------------|

| 1 | 1 | Approval |

| 2 | 1 | Rejection |

| 3 | 1 | Stoppage |

| 4 | 3 | Cancellation|

Analyse der Alternativen

• Direkte View auf OLTP — nicht akzeptabel wegen der Belastung des OLTP-Systems.
• CDC → Kafka → DWH — entlastet OLTP, aber die Lesegeschwindigkeit aus DWH ist ähnlich wie beim Original. Fügt zwei Schichten hinzu: Kafka und die View.

Beide Ansätze liefern keine Echtzeit-Updates für das JOIN-Ergebnis.

Flink-Architektur

Lösung: CDC-Streams nach Kafka → Flink Table API → stateful JOIN → Sink (Konsole für MVP).

Maven-Abhängigkeiten

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>3.2.0-1.19</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-sql-client</artifactId>
    <version>1.19</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-loader</artifactId>
    <version>${flink.version}</version>
</dependency>

Datenmodelle

public class User implements Serializable {
    public Integer id;
    public String firstname;
    public String lastname;
    // getters/setters
    public static User fromRow(Row row) { /* mapper */ }
}

public class Domain implements Serializable {
    public Integer id;
    public Integer user_id;
    public String domain_name;
    // getters/setters
    public static Domain fromRow(Row row) { /* mapper */ }
}

Kafka-Tabellen erstellen

tableEnv.executeSql("CREATE TABLE users (" +
    "`before` ROW<id: INT, firstname: STRING, lastname: STRING>," +
    "`op` STRING," +
    "`after` ROW<id: INT, firstname: STRING, lastname: STRING>," +
    ") WITH (" +
    "'connector' = 'kafka'," +
    "'topic' = 'users_topic'," +
    "'properties.bootstrap.servers' = 'kafka-brokers'," +
    "'properties.group.id' = 'users_consumer_group'," +
    "'scan.startup.mode' = 'earliest'");

DataStream<User> users = tableEnv.toDataStream(tableEnv.from("users")).map(User::fromRow);

// Similarly for domains
tableEnv.executeSql("CREATE TABLE domains (...)");
DataStream<Domain> domains = tableEnv.toDataStream(tableEnv.from("domains")).map(Domain::fromRow);

Implementierung des stateful JOIN

Streams nach dem Schlüssel user_id == domain.user_id mit KeyedCoProcessFunction joinen:

users
    .connect(domains)
    .keyBy(
        user -> user.id,
        domain -> domain.user_id
    )
    .process(new Join1())
    .print();

KeyedCoProcessFunction

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

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

    @Override
    public void processElement2(Domain domain, Context ctx, Collector<Output> out) throws Exception {
        domainsState.put(domain.id, domain);
        User user = usersState.value();
        if (user != null) {
            out.collect(new Output(user.id, user.firstname, user.lastname, domain.domain_name));
        }
    }

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

    public static class Output implements Serializable {
        public Integer user_id;
        public String firstname, lastname, domain_name;
        // constructors/getters/setters
    }
}

MapState wird für domains verwendet wegen der 1:n-Beziehung. State-Operationen sind O(1).

Wichtige Punkte

• Echtzeit-Updates: JOIN-Ergebnis ist bei Ankunft eines CDC-Events aktuell.
• Stateful-Verarbeitung: ValueState für users (1:1), MapState für domains (1:N).
• Skalierbarkeit: Bewährt bei Terabyte-Volumen mit OLAP-Sink.
• Leistung: O(1)-State-Zugriff, niedrige Latenz.
• Nächste Schritte: DELETE/UPDATE handhaben, RocksDB, Skew-Balancing.

Empfehlungen für den Produktionsbetrieb

• OLAP-Sink (ClickHouse, Pinot) statt Konsole verwenden.
• op-Flags aus CDC für korrekte UPDATE/DELETE-Behandlung verarbeiten.
• RocksDB als State Backend bei großen Volumen einsetzen.
• Key-Skew überwachen und Partitionen verteilen.
• Skalierung im Cluster testen.

— Editorial Team