Flink 실시간 JOIN: 테이블 4+2 TB

Apache Flink에서 대형 테이블에 대한 실시간 JOIN 구현: 기본 접근법

users 테이블(4TB)과 domains(2TB) 간의 INNER JOIN 결과에 대한 실시간 업데이트를 활성화해야 합니다. 쿼리는 user_id, firstname, lastname, domain_name을 반환합니다. 전통적인 뷰는 데이터 소스(OLTP)에 과부하를 일으키며, CDC를 통해 DWH로의 간단한 물리화는 요구되는 속도를 제공하지 않습니다.

예제 데이터:

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|

대안 분석

• OLTP 상의 직접 뷰 — OLTP 시스템에 대한 부하로 인해 받아들일 수 없습니다.
• CDC → Kafka → DWH — OLTP의 부하를 줄이지만, DWH로부터의 읽기 속도는 원본과 유사합니다. Kafka와 뷰라는 두 개의 레이어를 추가합니다.

두 접근법 모두 JOIN 결과에 대한 실시간 업데이트를 제공하지 못합니다.

Flink 아키텍처

솔루션: CDC 스트림을 Kafka로 → Flink Table API → 상태ful JOIN → 싱크(MVP를 위한 콘솔).

Maven Dependencies

<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>

Data Models

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 */ }
}

Creating Kafka Tables

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

상태ful JOIN 구현

user_id == domain.user_id를 키로 스트림을 조인하기 위해 KeyedCoProcessFunction 사용:

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는 1:N 관계로 인해 domains에 사용됩니다. 상태 작업은 O(1)입니다.

주요 포인트

• 실시간 업데이트: CDC 이벤트 도착 시 JOIN 결과가 최신 상태입니다.
• 상태ful 처리: users(1:1)에 ValueState, domains(1:N)에 MapState.
• 확장성: OLAP 싱크와 함께 테라바이트 규모에서 검증됨.
• 성능: O(1) 상태 접근, 낮은 지연.
• 다음 단계: DELETE/UPDATE 처리, RocksDB, skew 밸런싱.

프로덕션 권장 사항

• 콘솔 대신 OLAP 싱크(ClickHouse, Pinot) 사용.
• CDC의 op 플래그를 처리하여 올바른 UPDATE/DELETE 처리.
• 대용량에 대해 RocksDB를 상태 백엔드로 사용.
• 키 skew 모니터링 및 파티션 분배.
• 클러스터에서 스케일링 테스트.

— Editorial Team