Kafka Streams: KStream과 KTable의 이해와 조인 활용

Kafka Streams는 실시간 데이터 처리를 위한 강력한 도구로, KStream과 KTable이라는 두 가지 데이터 구조를 제공합니다. 이번 글에서는 이 두 데이터 구조의 개념과 차이를 살펴보고, 실제 KStream-KStream, KStream-KTable, KTable-KTable 조인의 예제를 설명하겠습니다.

 

KStreams - KStreams 조인

---

[개념]

 

KStream-KStream 조인은 두 개의 실시간 데이터 스트림을 결합합니다.

이 조인은 타임 윈도우를 기준으로 이루어지며, 설정된 기간 내에 동일한 키를 가진 데이터를 결합합니다.

 

[조인의 특징]

• 타임 윈도우가 필수적입니다.
• 데이터가 실시간으로 들어오므로, 시간이 지나면 윈도우가 닫히고 더 이상 데이터를 결합할 수 없습니다.

[사용 사례]

• 두 개의 스트림에서 발생하는 동시에 일어나는 이벤트를 결합
• 예: 결제 이벤트 스트림과 배송 요청 스트림의 결합

[시나리오: 결제와 배송 요청 이벤트 결합]

• payments 스트림: 결제 정보 Key: 주문 ID, Value: 결제 상태
• shipments 스트림: 배송 요청 정보 Key: 주문 ID, Value: 배송 상태

KStream<String, String> payments = builder.stream("payments");
KStream<String, String> shipments = builder.stream("shipments");

// 타임 윈도우를 적용한 조인
KStream<String, String> joinedStream = payments.join(
    shipments,
    (payment, shipment) -> "Payment: " + payment + ", Shipment: " + shipment,
    JoinWindows.of(Duration.ofMinutes(5)) // 5분 윈도우 설정
);

joinedStream.to("joined-orders");

Key: "order123", Value: "Payment: completed, Shipment: dispatched"

 

KStreams - KTable 조인

---

[개념]

KStream-KTable 조인은 실시간 스트림(KStream)과 상태 기반 데이터(KTable)를 결합합니다.

KStream의 각 이벤트가 KTable의 현재 상태와 조인됩니다.

 

[조인의 특징]

• KStream의 데이터는 실시간으로 계속 흐르지만, KTable은 현재 상태만 유지합니다.
• KStream의 각 레코드는 KTable의 최신 상태와 결합됩니다.

[사용 사례]

• 실시간 스트림을 기존 참조 데이터와 결합
• 예: 사용자 행동 스트림과 사용자 프로필 테이블 결합

[시나리오: 사용자 행동 로그와 사용자 프로필 조인]

• user-actions 스트림: 사용자가 수행한 행동 Key: 사용자 ID, Value: 행동 타입 (click, purchase)
• user-profiles 테이블: 사용자 프로필 정보 Key: 사용자 ID, Value: 사용자 이름과 지역

KStream<String, String> userActions = builder.stream("user-actions");
KTable<String, String> userProfiles = builder.table("user-profiles");

KStream<String, String> joinedStream = userActions.leftJoin(
    userProfiles,
    (action, profile) -> {
        if (profile == null) return action + " by unknown user";
        return action + " by " + profile;
    }
);

joinedStream.to("user-actions-with-profile");

Key: "user123", Value: "click by Alice, US"
Key: "user456", Value: "purchase by unknown user"

 

KStreams - KTable 조인

---

[개념]

 

KTable-KTable 조인은 두 개의 상태 기반 데이터(KTable)를 결합합니다.

이는 KTable의 키를 기준으로 데이터가 업데이트될 때마다 결과 테이블도 업데이트됩니다.

 

[조인의 특징]

• 조인은 상태 기반으로 동작하며, 입력 데이터가 변경될 때마다 결과 테이블도 갱신됩니다.
• 결합된 데이터는 항상 최신 상태를 유지합니다.
• TTL은 KTable 상태의 만료 시간을 관리합니다.
• TTLEmitter를 사용하여 주어진 TTL 기간이 지난 데이터를 삭제.
• 이를 통해 상태 크기를 제어하고 메모리 사용량을 최적화.

[사용 사례]

• 두 개의 참조 데이터 테이블을 병합하여 확장된 참조 데이터를 생성
• 예: 상품 정보 테이블과 카테고리 정보 테이블 결합

[시나리오: 상품 정보와 카테고리 정보 결합]

• product-info 테이블: 상품의 이름과 가격Key: 상품 ID, Value: 상품 이름과 가격
• category-info 테이블: 상품의 카테고리 정보Key: 상품 ID, Value: 카테고리 이름

KTable<String, String> productInfo = builder.table("product-info");
KTable<String, String> categoryInfo = builder.table("category-info");

KTable<String, String> joinedTable = productInfo.join(
    categoryInfo,
    (product, category) -> product + " in category: " + category
);

joinedTable.toStream().to("product-with-category");

Key: "product123", Value: "Coffee Machine, $120 in category: Appliances"

 

조인의 차이점 

---

 

KStream-KStream	스트림-스트림	타임 윈도우 기반, 실시간 이벤트 간 결합	결제와 배송 요청 스트림 결합
KStream-KTable	스트림-테이블	실시간 스트림과 상태 기반 데이터 결합	사용자 행동 로그와 사용자 프로필 조인
KTable-KTable	테이블-테이블	상태 데이터 간 조인, 항상 최신 상태 유지	상품 정보와 카테고리 정보 병합

 

실무 사례: KTable - KTable 조인으로 데이터 정확도 개선

---

 

실무에서 제가 결합하고자 했던 데이터는 상품 정보 A와 상품 정보 B였습니다. 기존에는 KTable-KStream 조인을 사용했으나, 이 방식에서는 오래된 데이터 A와 최신 데이터 B가 결합되는 문제가 있었습니다. 이는 두 정보가 모두 실시간 데이터가 아니라 상태성 데이터였기 때문에 발생한 이슈였습니다.

 

이를 해결하기 위해 KTable-KTable 조인으로 변경하고, TTL을 설정하여 KTable 상태의 만료 시간을 관리했습니다. 이로써 과거 데이터가 결합되어 상품 정보에 반영되는 문제를 해결할 수 있었습니다.

 

결론

---

 

Kafka Streams의 조인 연산은 다양한 데이터 소스를 결합하여 강력한 실시간 데이터 파이프라인을 구축할 수 있는 도구입니다

• KStream-KStream: 이벤트 간 시간 관계를 분석하고 처리
• KStream-KTable: 실시간 스트림 데이터를 참조 데이터와 결합
• KTable-KTable: 상태 기반 데이터의 지속적인 동기화 및 최신 상태 유지

Kafka Streams를 활용한 조인 방식은 데이터의 특성과 요구사항에 따라 달라져야 합니다.실시간 스트림인지, 상태 기반 데이터인지 정확히 이해하고 올바른 조인 방식을 선택하는 것이 데이터 파이프라인의 정확성과 효율성을 높이는 핵심입니다.

 

참고

• https://gunju-ko.github.io/kafka/kafka-stream/2018/05/28/Stream-DSL.html