Kafka - 11. kafka streams

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

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

"직접 짜면 왜 지옥인가"부터 보면 Streams의 가치가 선명해진다 — consumer poll 루프, 상태 저장(RocksDB 연동), 장애 시 상태 복구, partition 재분배 시 상태 이관, exactly-once 보장을 손수 구현하면 수천 줄의 보일러플레이트가 나온다. Streams는 이걸 groupByKey().aggregate(...) 한 줄로 압축한다. 단, 그 압축이 어떤 전제 위에 서는지 — stream-table duality, state store + changelog, application.id의 역할, EOS 옵션 — 를 정확히 알아야 "프레임워크가 알아서 한다"는 말에 속지 않는다.

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 무관 조인(참조 데이터용).

duality가 왜 실용적인가

이 dual 모델이 없으면 같은 데이터를 두 용도로 쓰기 어렵다. 예를 들어 orders topic을 "주문 이벤트 스트림"으로 보면 KStream이고, "사용자별 최신 주문 상태"로 보면 KTable이다. 같은 topic을 KStream으로 읽으면 각 주문이 이벤트로, KTable로 읽으면(또는 KStream을 groupByKey().reduce()로 바꾸면) 사용자별 최신값으로 보인다. 데이터는 그대로인데 관점이 다른 두 처리를 같은 topic에서 자연스럽게 할 수 있다 — 이게 duality의 실용적 가치다.

또한 KStream → KTable 변환(groupByKey().aggregate/reduce)과 KTable → KStream 변환(toStream())이 자유로워서, 처리 흐름 안에서 두 관점을 오간다. 예를 들어 "주문 이벤트를 모아 사용자별 합계(KTable)를 만들고, 그 합계 변화를 다시 이벤트 스트림(KStream)으로 뽑아 다른 topic으로 보낸다"는 패턴이 자연스럽다.

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로 선언적으로 정의한다.

topology의 핵심 개념:

• source node: 입력 topic에서 읽는 진입점(builder.stream("orders")).
• processor node: 변환(filter, map, aggregate, join, windowedBy).
• sink node: 출력 topic에 쓰는 출구(.to("output-topic")).

이 node들이 DAG를 이루고, 데이터가 source → processor → sink로 흐른다. 상태 저장 연산(aggregate)이 있는 processor는 state store + changelog topic을 자동으로 갖는다. topology를 Topology.build()로 컴파일한 뒤 KafkaStreams로 실행하면, 각 partition이 인스턴스에 분산分配되고(장 05의 consumer group과 같은 방식), 각 인스턴스가 자기 partition에 해당하는 state store를 로컬에 유지한다.

예 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 복구가 어떻게 "정확히" 되는가

state store의 모든 갱신이 changelog에 기록되므로, changelog를 처음부터 다시 replay하면 같은 상태가 재구축된다. 이게 "정확히 복구"의 원리다 — 상태를 직접 백업하는 게 아니라, 상태를 만든 갱신 이력을 백업해서 replay로 복원한다. 이벤트 소싱의 핵심 아이디어와 같다.

EOS(exactly_once_v2)를 켜면 changelog 쓰기도 트랜잭션으로 원자화돼서, 복구 시에도 "중복 없이 정확히"가 보장된다. EOS가 없으면 at-least-once라 복구 중 일시적 중복이 생길 수 있지만, 최종 상태는 같다(aggregate 연산이 멱등하면).

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 변경 금지(상태 손상).

changelog topic 이름은 {application.id}-{store-name}-changelog 형식이다. 그래서 application.id를 바꾸면 새 changelog topic이 만들어져서 상태를 처음부터 다시 구축한다 — "application.id를 함부로 바꾸면 안 된다"는 규칙의 이유. 같은 앱의 재배포는 같은 application.id를 써야 기존 상태를 이어받는다.

windowing과 시간 — event time vs processing time

스트림 처리에서 "시간"은 두 가지가 있다:

• event time: 이벤트가 실제로 발생한 시각(record의 timestamp). "언제 주문이 일어났나".
• processing time: Streams가 이벤트를 처리하는 시각. "Streams가 언제 이 메시지를 봤나".

이 둘이 다르면 집계 결과가 달라진다. 예를 들어 "1시간 창으로 주문 합산"을 할 때, event time 기준이면 "오후 12시에 발생한 주문"이 한 창에 들어가지만, processing time 기준이면 "오후 12시에 Streams가 처리한 주문"이 한 창에 들어간다. consumer 지연·재시작이 있으면 두 시간이 크게 벌어진다 — 늦게 도착한 event time 1:30의 주문이 processing time 2:30에 처리될 수 있다.

Streams는 event time 기반 windowing을 권장한다 (streams/developer-guide). 단, 늦게 도_arrive하는 이벤트(late event)를 어떻게 처리할지가 문제된다 — grace period(TimeWindows.ofSizeWithNoGrace vs ofSizeAndGrace)로 "창이 닫힌 뒤 얼마나 늦은 이벤트까지 받을지"를 정한다. grace가 0이면 창 닫힌 뒤 도착한 이벤트는 버려지고, 길면 늦은 이벤지를 흡수하지만 창 닫힘이 늦어진다.

windowing을 쓸 때 "event time vs processing time", "grace period", "late event 처리" 세 가지를 항상 함께 고려해야 한다. 하나라도 놓치면 집계 결과가 예상과 다르다 — 특히 consumer 지연·재처리 시나리오에서.

interactive queries — state store를 직접 조회

상태 저장 처리(state store)의 결과를 애플리케이션에서 직접 읽을 수 있다. 예를 들어 "사용자 A의 현재 주문 합계"를 조회하려면, 별도 DB에 동기화하지 않고 state store에서 바로 읽을 수 있다.

// 상태 저장 앱에서 store를 직접 조회
ReadOnlyKeyValueStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType(
        "user-totals-store",
        QueryableStoreTypes.keyValueStore()
    )
);
Long userATotal = store.get("userA");  // "userA"의 현재 합계

단, state store는 partition별로 분산되므로, 내 인스턴스가 갖지 않은 partition의 key는 직접 조회할 수 없다. 이때는 metadataForKey()로 "이 key는 어느 인스턴스에 있나"를 찾은 뒤, 그 인스턴스에 RPC로 물어봐야 한다. 이 "분산 store 조회" 패턴을 쓰면 별도 DB 없이 Streams 앱 자체를 쿼리 가능한 상태 저장소로 쓸 수 있다.

DSL vs Processor API

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

초보자는 DSL로 시작하라 — 대부분의 집계·필터·조인이 DSL 한 줄로 된다. DSL의 aggregate/join이 내부적으로 Processor API를 쓴다. DSL이 부족할 때(예: 복잡한 상태 머신, 외부 시스템과의 동기 호출)만 Processor API로 내려간다. 두 API는 같은 topology를 표현하는 다른 수준일 뿐, 서로 배타적이지 않다 — 한 topology 안에서 섞어 쓸 수도 있다.

Kafka 4.3 실습 — 직접 확인하기

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

앱 빌드·실행

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

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

# 3. 다른 터미널에서 입력 topic에 메시지 produce (console-producer 입력 예)
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이 자동 생성.

repartition topic 관찰 (key 변경 시)

topology 안에서 selectKey나 map으로 key를 바꾸면 자동 repartition topic이 생성된다:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# order-aggregation-app-repartition-XXX 같은 내부 topic

확인할 것: key를 바꾸는 topology에선 repartition topic이 생기고, 데이터가 한 번 더 Kafka를 거친다 — 네트워크·디스크 비용이 든다. key 변경이 빈번하면 성능 영향이 있으므로, 가능하면 원래 key를 유지하는 설계가 좋다.

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

오해	정정
"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은 삭제해도 된다"	금지. 상태 손상
"application.id는 아무 값이나 해도 된다"	안 된다. changelog topic 이름의 prefix라, 바꾸면 상태를 처음부터 재구축
"KStream-KTable join은 무조건 된다"	co-partitioning(같은 partition 수 + 같은 key partitioning)이 필요. 안 맞으면 join 안 됨
"늦게 도착한 이벤트는 그래도 창에 들어간다"	grace period 내이면 들어감. grace 0이거나 창이 닫혔으면 버려짐

더 깊이

• 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).
• co-partitioning: 두 KStream/KTable을 join하려면 같은 수의 partition과 같은 key partitioning이 필요. 이게 안 맞으면 join이 의도대로 안 됨 — topic 설계 단계에서 partition 수와 partitioner를 맞춰야 한다.
• repartition topic: topology 안에서 key를 바꾸면(selectKey, map 등) downstream 처리를 위해 자동 repartition topic이 생성된다. 이 과정은 자동이지만 네트워크·디스크 비용이 들므로, key 변경이 빈번한 topology는 성능 영향이 있다.
• rockDB tuning: state store의 기본 백엔드. 대량 상태를 쓸 때 rockDB 옵션(num_THREADS, write_buffer_size, block_cache_size)을 튜닝해야 할 수 있다.

요약 — 이 글의 결론

• Kafka Streams = 클라이언트 라이브러리(별도 클러스터 아님). 앱에 내장.
• stream-table duality: stream = 이벤트 흐름, table = key별 최신 상태. KStream ↔ KTable 변환 자유로움.
• KStream(이벤트) / KTable(상태) / GlobalKTable(전체 복제).
• Maven 의존성 + Properties 설정(application.id 핵심) + DSL topology로 앱 구성.
• state store(RocksDB) + changelog topic = 내결함 상태 처리. 별도 DB 불필요. 복구는 changelog replay.
• DSL(고수준) / Processor API(저수준). 한 topology 안에서 섞어 쓸 수 있음.
• windowing: event time 기준 권장. grace period로 늦은 이벤트 처리.
• co-partitioning: KStream-KTable join은 같은 partition 수 + 같은 key partitioning이 필요.
• EOS: processing.guarantee=exactly_once_v2.
• interactive queries: state store를 앱에서 직접 조회. partition 분산이라 다른 인스턴스 key는 RPC 필요.
• 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의 역할 차이는? 둘을 같이 쓰는 시나리오를 상상해 보라.
7. event time과 processing time의 차이가 집계 결과에 어떤 영향을 미치는가? grace period는 왜 필요한가?
8. co-partitioning이란 무엇이고, KStream-KTable join에서 왜 필요한가?
9. interactive queries로 다른 인스턴스가 갖고 있는 key의 값을 조회하려면 어떻게 해야 하는가?

---

참고

• Streams — Core Concepts - 접근 2026-07-09 (stream-table duality, KStream/KTable, topology)
• Streams Developer Guide - 접근 2026-07-09 (DSL, Processor API, windowing)
• Streams config (EOS, application.id) - 접근 2026-07-09 (application.id, processing.guarantee)
• Streams code examples - 접근 2026-07-09 (pipeline 예제)
• Interactive Queries (developer-guide) - 접근 2026-07-09 (state store 조회, metadataForKey)
• KIP-447(exactly_once_v2), KIP-345(static membership)