Kafka - 11. kafka streams

Kafka Streams — topic의 데이터를 가공(집계·조인·창)하는 라이브러리

orders topic에 주문 이벤트가 계속 쌓인다. "5분 단위로 주문 금액을 합산하고 싶다" — 직접 consumer + window 로직 + 상태 저장을 짜면, 장애 시 상태 복구·재처리가 지옥이다. Kafka Streams는 집계·조인·창(windowing) + 상태 저장(state store) + 장애 복구(changelog)를 프레임워크가 알아서 처리한다.

이 글은 Streams가 뭔지부터 시작해, 실제로 앱을 어떻게 만들고 실행하는지(Maven 설정, topology 코드, 실행)까지 다룬다. 읽고 나면 간단한 스트림 처리 앱을 만들 수 있다.

Streams란 — 앱에 내장되는 스트림 처리 라이브러리

Kafka Streams는 별도 처리 클러스터가 아니라, 애플리케이션에 내장되는 클라이언트 라이브러리다. Spark/Flink처럼 별도 클러스터를 세울 필요 없이, 당신의 앱 프로세스 안에서 동작한다.

왜 라이브러리인 게 강점인가:

• 배포 단순 — 그냥 앱(JAR)을 띄우면 됨. 별도 클러스터 운영 오버헤드 없음.
• 확장 — 인스턴스를 더 띄우면 consumer group처럼 partition 재분배로 부하 분산(장 05).
• 장애 복구 — 인스턴스가 죽으면 다른 인스턴스가 partition + state를 인계.

Connect(장 10)가 "데이터를 옮기는" 거라면, Streams는 "데이터를 가공하는" 도구다.

stream-table duality — 같은 데이터의 두 모습

Streams의 핵심 통찰: "stream과 table은 같은 것의 두 모습"이다.

• stream = 이벤트의 흐름(예: 주문이 발생하는 기록).
• table = key별 최신 상태(예: 각 사용자의 현재 주문 합계).

거래 내역을 시간순으로 적은 장부(stream)에서, 각 계좌별 마지막 잔액만 모으면 잔액장(table)이 된다. 반대로 잔액장의 변경 이력을 펼치면 다시 거래 내역(stream)이 된다.

Streams는 이를 두 타입으로 모델링한다:

• KStream — record 스트림. 각 이벤트가 독립.
• KTable — key→값의 최신 상태 테이블. 같은 key 갱신은 이전값 대체.
• GlobalKTable — 모든 인스턴스가 전체 복제. partition 무관 조인(참조 데이터용).

Streams 앱 만들기 — Maven 설정부터

의존성 (pom.xml)

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>4.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>4.3.0</version>
</dependency>
<!-- JSON serde (예시) -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>4.3.0</version>
    <scope>test</scope>
</dependency>

앱 설정 (Properties)

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-aggregation-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
// EOS 켜기 (선택)
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

• application.id — Streams 앱 식별자. changelog topic 이름의 prefix가 됨. 같은 id의 인스턴스들이 consumer group 형성.
• PROCESSING_GUARANTEE — at_least_once(기본) 또는 exactly_once_v2(EOS, 장 08).

topology 코드 — "처리 흐름" 정의

topology는 데이터 처리 흐름의 DAG(유향 비순환 그래프)다. DSL API로 선언적으로 정의한다.

예 1: 사용자별 주문 금액 합산

// 1. 입력 topic → KStream
KStream<String, String> orders = builder.stream("orders");

// 2. key별 그룹화 + 금액 합산 → KTable
KTable<String, Long> userTotals = orders
    .groupByKey()
    .aggregate(
        () -> 0L,                          // 초기값
        (key, order, total) -> total + parseAmount(order),  // 집계 로직
        Materialized.as("user-totals-store")  // state store 이름
    );

// 3. 결과를 topic으로 내보내기
userTotals.toStream().to("user-totals", Produced.with(Serdes.String(), Serdes.Long()));

이 코드가 하는 일:

• orders topic의 메시지를 key(사용자 ID)별로 그룹화.
• 금액을 누적 합산(state store에 저장).
• 결과를 user-totals topic으로 내보냄.
• 상태 저장(aggregate)이므로 changelog topic이 자동 생성되어 장애 복구 지원.

예 2: 필터링 (무상태)

KStream<String, String> orders = builder.stream("orders");
KStream<String, String> bigOrders = orders.filter((key, value) -> parseAmount(value) > 1000);
bigOrders.to("big-orders");

무상태 연산(filter/map)은 state store나 changelog가 필요 없다 — 빠르고 단순.

예 3: 시간 창 집계 (windowing)

KTable<Windowed<String>, Long> hourlyTotals = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .count();

1시간 창으로 key별 개수 집계. event time 기준.

topology 시작

Topology topology = builder.build();
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();

// 종료 훅
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

상태 저장 처리가 장애 후에도 정확히 복구되는 원리 — changelog

상태 저장 처리(집계 등)는 "별도 DB가 필요하다"고 생각하기 쉽다. Streams는 state store를 로컬(RocksDB)에 두고, 그 백업을 Kafka 내부 topic(changelog)에 영속한다.

flowchart LR
    IN[orders topic] --> S["KStream"]
    S --> AGG["aggregate<br/>(state store: RocksDB)"]
    AGG --> CH["changelog topic<br/>(백업/복구)"]
    AGG --> OUT[user-totals topic]

• 집계 상태는 로컬 state store(RocksDB)에 유지.
• store의 모든 갱신은 changelog topic에 기록.
• 장애 시 changelog에서 state store 재구축 → 별도 DB 없이 내결함 상태 처리.
• state store는 partition 단위로 분산 → 확장성.

changelog topic 확인

Streams 앱 기동 후 자동 생성되는 내부 topic:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# order-aggregation-app-user-totals-store-changelog 같은 topic 생성 확인

이 topic이 state store의 복구 원천이다 — 삭제/retention 변경 금지(상태 손상).

DSL vs Processor API

API	수준	특징
DSL	고수준	map/filter/aggregate/join/windowedBy. 선언적. 대부분 이걸로 충분.
Processor API	저수준	커스텀 processor + state store 직접 제어. 최대 유연성.

초보자는 DSL로 시작하라 — 대부분의 집계·필터·조인이 DSL 한 줄로 된다.

Kafka 4.3 실습 — 직접 확인하기

사전: KRaft 단일 노드 기동. Streams는 Java 라이브러리 → Maven 프로젝트 필요.

앱 빌드·실행

# 1. Maven 프로젝트 빌드
mvn clean package

# 2. 앱 실행
java -jar target/my-streams-app.jar

# 3. 다른 터미널에서 입력 topic에 메시지 produce
bin/kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092
# userA:{"amount": 5000}
# userA:{"amount": 3000}
# userB:{"amount": 7000}

# 4. 출력 topic 확인
bin/kafka-console-consumer.sh --topic user-totals --from-beginning --bootstrap-server localhost:9092
# userA: 8000
# userB: 7000

changelog topic 관찰

bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# order-aggregation-app-user-totals-store-changelog 생성 확인

확인할 것: 상태 저장 연산을 쓰면 changelog topic이 자동 생성.

흔히 묻는 것, 흔히 틀리는 것

오해	정정
"Streams는 별도 클러스터다"	라이브러리. 앱에 내장
"KStream과 KTable은 같다"	다르다. stream=이벤트 흐름, table=최신 상태
"상태 저장 처리엔 외부 DB가 필요"	로컬 state store(RocksDB) + changelog로 자체 해결
"Connect와 Streams는 같은 역할"	다르다. Connect=데이터 이동, Streams=데이터 가공
"EOS는 자동이다"	processing.guarantee=exactly_once_v2 설정 시. 기본은 at-least-once
"windowing은 processing time 기준"	event time 기준 권장 (지연 이벤트 처리)
"changelog topic은 삭제해도 된다"	금지. 상태 손상

더 깊이

• interactive queries: state store를 앱이 직접 조회(별도 DB 동기화 없이 "현재 key X의 값" 반환). 단, partition 분산이라 다른 인스턴스의 key 조회엔 RPC 필요.
• GlobalKTable vs KTable 조인: KTable-KStream join은 같은 key + 같은 partition 전제(co-partitioning). GlobalKTable은 partition 무관(참조 데이터용).
• rebalance와 static membership: Streams 앱 재시작 시 rebalance 비용. group.instance.id(장 05)로 최소화.
• EOS: processing.guarantee=exactly_once_v2로 consume-process-produce 자동 원자화(장 08).

요약 — 이 글의 결론

• Kafka Streams = 클라이언트 라이브러리(별도 클러스터 아님). 앱에 내장.
• stream-table duality: stream = 이벤트 흐름, table = key별 최신 상태.
• KStream(이벤트) / KTable(상태) / GlobalKTable(전체 복제).
• Maven 의존성 + Properties 설정(application.id 핵심) + DSL topology로 앱 구성.
• state store(RocksDB) + changelog topic = 내결함 상태 처리. 별도 DB 불필요.
• DSL(고수준) / Processor API(저수준).
• EOS: processing.guarantee=exactly_once_v2.
• Connect ≠ Streams: 이동 vs 가공.

생각해 볼 문제

1. Streams가 "라이브러리"라는 것이 운영상 어떤 의미인가?
2. stream-table duality를 KStream/KTable로 설명하라.
3. Maven 프로젝트에서 Streams 앱을 빌드하려면 어떤 의존성이 필요한가?
4. application.id가 changelog topic 이름에 어떤 영향을 주는가?
5. state store가 장애 후 정확히 복구되는 원리(changelog)는?
6. Connect와 Streams의 역할 차이는? 둘을 같이 쓰는 시나리오를 상상해 보라.

---

참고

• Streams — Core Concepts - 접근 2026-07-09
• Streams Developer Guide - 접근 2026-07-09
• Streams config (EOS, application.id) - 접근 2026-07-09
• Streams code examples - 접근 2026-07-09
• KIP-447(exactly_once_v2)