Kafka Streams를 활용하여 실시간 피드 제공 기능을 개발하는 도중 KStream-KTable 간 조인 시 발생했던 문제 상황을 공유하고자 합니다.글에서 제공되는 예시 코드는 실제로 동작하지 않는 pseudo code입니다.  현재 상황코스피/코스닥에 상장된 기업들의 주가 정보(시가/고가/종가/현재가 등등)을 5초마다 호출하여 파티션이 5개인 토픽에 저장합니다.오전 6시에 이전 영업일의 코스피/코스닥에 상장된 기업들의 valuation 값(per, pbr, psr 등등)들이 저장된 엑셀 데이터를 파싱하여 DB 저장 및 파티션이 1개인 토픽에 저장합니다. 두 토픽에 저장된 데이터를 이용하여 각 기업의 실시간 주가 정보를 이용하여 valuation을 계산하여 오전 6시에 엑셀 데이터를 파싱하여 토픽에..

Kafka Streams API를 이용하여 실시간 데이터 파이프라인을 구축하는 중 직면한 문제를 기록했다. 우선 실시간 파이프라인의 대략적인 플로우를 설명하면 A라는 토픽에 데이터가 특정 시간 동안 얼마나 들어오는지 계산하여 B라는 토픽에 데이터를 적재하는 토폴로지를 구성하고 있다. 여기서 핵심은 특정 시간 동안 내가 원하는 데이터가 몇 개가 들어오는지 파악하는 것이다. 현 상황우선 A라는 토픽에서 데이터를 읽어와 조건에 맞춰 filter() 메서드를 이용해 필터링 처리 한 이후 groupBy() 메서드로 특정 값 기준으로 데이터를 그룹화시킨 다음 count 메서드로 개수를 산정했다. 이렇게 하면 A라는 토픽에 들어오는 데이터 중 filter 조건에 맞는 데이터의 개수를 전부 카운트하게 된다.이때, 특정..