Kafka

Kafka Streams로 이벤트 처리하기 — WordCount 예제

데이터가 Kafka에 이벤트로 저장되고 나면, Java/Scala용 클라이언트 라이브러리인 Kafka Streams로 그 데이터를 처리할 수 있다. 입력·출력 데이터가 Kafka 토픽에 저장되는 미션 크리티컬한 실시간 애플리케이션과 마이크로서비스를 구현할 수 있게 해주는데, 클라이언트 쪽에서 표준 Java·Scala 애플리케이션을 작성·배포하는 단순함과, Kafka의 서버 쪽 클러스터 기술이 주는 이점(고도의 확장성·탄력성·내결함성·분산)을 결합한 것이 핵심이다. 이 라이브러리는 정확히 한 번(exactly-once) 처리, 상태 유지(stateful) 연산과 집계, 윈도잉, 조인, 이벤트-타임 기반 처리 등을 지원한다.

대표적인 첫 예제로 널리 알려진 WordCount 알고리즘을 Kafka Streams로 구현한 코드는 다음과 같다.

KStream<String, String> textLines = builder.stream("quickstart-events");

KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();

wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

코드를 단계별로 보면, builder.stream("quickstart-events")는 quickstart-events 토픽을 읽어 KStream<String, String> textLines를 만든다 — 각 레코드가 하나의 텍스트 줄에 해당하는, 끝없이 이어지는 이벤트 스트림이다. 이어서 flatMapValues로 각 줄의 값을 소문자로 바꾸고 공백 기준으로 단어들로 쪼개, 레코드 하나(한 줄)를 여러 개의 단어 레코드로 펼친다(flat-map). groupBy((keyIgnored, word) -> word)는 원래 키는 무시하고 단어 자체를 새 키로 삼아 레코드를 재그룹화하며, count()는 그룹별(=단어별)로 몇 번 등장했는지 세어 KTable<String, Long> wordCounts를 만든다 — KTable은 각 단어 키에 대한 '현재까지의 누적 카운트'를 나타내는, 계속 갱신되는 테이블이다. 마지막으로 wordCounts.toStream()으로 이 테이블의 변경 이력을 다시 이벤트 스트림으로 바꾸고, .to("output-topic", Produced.with(Serdes.String(), Serdes.Long()))로 output-topic이라는 토픽에 (단어, 누적 카운트) 쌍을 문자열 키·Long 값의 Serde로 직렬화해 게시한다.

Kafka 환경 종료하기

Quick Start를 마쳤다면 환경을 정리해도 되고, 계속 실습을 이어가도 된다. 정리하려면 먼저 아직 켜 둔 프로듀서·컨슈머 클라이언트를 Ctrl-C로 멈추고, 이어서 Kafka 브로커도 Ctrl-C로 멈춘다. 만약 지금까지 만든 이벤트를 포함해 로컬 Kafka 환경의 데이터까지 모두 지우고 싶다면 다음 명령을 실행한다.

$ rm -rf /tmp/kafka-logs /tmp/kraft-combined-logs

Ecosystem — 메인 배포판 밖의 통합 도구들

공식 문서의 Ecosystem 섹션은 매우 짧다. Kafka 메인 배포판 바깥에서 Kafka와 통합되는 다양한 도구가 많이 있으며, ecosystem 페이지가 스트림 처리 시스템, Hadoop 연동, 모니터링, 배포 도구 등 이런 도구 다수를 목록으로 정리해 안내한다는 내용만 담겨 있다.