🕊️ 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 payments = builder.stream("payments"); KStream shipments = builder.stream("shipments"); // 타임 윈도우를 적용한 조인 KStream 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 userActions = builder.stream("user-actions"); KTable userProfiles = builder.table("user-profiles"); KStream 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의 키를 기준으로 데이터가 업데이트될 때마다 결과 테이블도 업데이트됩니다. * 조인은 상태 기반으로 동작하며, 입력 데이터가 변경될 때마다 결과 테이블도 갱신됩니다. * 결합된 데이터는 항상 최신 상태를 유지합니다. [시나리오: 상품 정보와 카테고리 정보 결합] * product-info 테이블: 상품의 이름과 가격Key: 상품 ID, Value: 상품 이름과 가격 * category-info 테이블: 상품의 카테고리 정보Key: 상품 ID, Value: 카테고리 이름 KTable productInfo = builder.table("product-info"); KTable categoryInfo = builder.table("category-info"); KTable 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" 실무 사례: KTable - KTable 조인으로 데이터 정확도 개선 실무에서 제가 결합하고자 했던 데이터는 상품 정보 A와 상품 정보 B였습니다. 기존에는 KTable-KStream 조인을 사용했으나, 이 방식에서는 오래된 데이터 A와 최신 데이터 B가 결합되는 문제가 있었습니다. 이는 두 정보가 모두 실시간 데이터가 아니라 상태성 데이터였기 때문에 발생한 이슈였습니다.   이를 해결하기 위해 KTable-KTable 조인으로 변경하고, TTL을 설정하여 KTable 상태의 만료 시간을 관리했습니다. 이로써 과거 데이터가 결합되어 상품 정보에 반영되는 문제를 해결할 수 있었습니다.   결론 Kafka Streams의 조인 연산은 다양한 데이터 소스를 결합하여 강력한 실시간 데이터 파이프라인을 구축할 수 있는 도구입니다 * KStream-KStream: 이벤트 간 시간 관계를 분석하고 처리 * KStream-KTable: 실시간 스트림 데이터를 참조 데이터와 결합 * KTable-KTable: 상태 기반 데이터의 지속적인 동기화 및 최신 상태 유지 Kafka Streams를 활용한 조인 방식은 데이터의 특성과 요구사항에 따라 달라져야 합니다.실시간 스트림인지, 상태 기반 데이터인지 정확히 이해하고 올바른 조인 방식을 선택하는 것이 데이터 파이프라인의 정확성과 효율성을 높이는 핵심입니다. 원문: [https://ducktopia.tistory.com/141]