Kafka Streams

개념

토픽에 적재된 데이터를 실시간으로 변환하여 다른 토픽에 적재하는 라이브러리이다.
소스 토픽과 싱크 토픽의 카프카 클러스터가 서로 다르다면 스트림즈를 사용할 수 없으므로 컨슈머와 프로듀서 조합으로 처리해야 한다.
내부적으로 스레드를 1개 이상 생성 가능하고, 스레드는 1개 이상의 태스크를 가진다.
태스크는 데이터 처리 최소 단위로, 하나의 파티션과 매핑된다.
병렬 처리를 위해 파티션과 스트림즈 스레드 혹은 프로세스 개수를 늘려 처리량을 늘릴 수 있다.
보통 장애가 나도 안정적으로 운영할 수 있도록 스트림즈 프로세스를 여러 개 둔다.

토폴로지

프로세서
- 토폴로지를 이루는 노드 하나를 의미하며, 소스 프로세서, 스트림 프로세서, 싱크 프로세서 세 종류가 존재한다.
- 소스 프로세서는 데이터를 처리하기 위해 최초로 선언해야 하며 하나 이상의 토픽에서 데이터를 가져온다.
- 스트림 프로세서는 다른 프로세서가 반환한 데이터를 처리한다.
- 싱크 프로세서는 데이터를 특정 카프카 토픽으로 저장한다.

스트림
- 노드와 노드를 이은 선을 지칭하며, 토픽의 데이터인 레코드를 의미한다.

스트림즈 DSL

스트림즈DSL은 스트림 프로세싱에 쓰일 만한 다양한 기능들을 API로 만들어 제공한다.
예를 들어 메시지 값을 기반으로 토픽 분기 처리, 지난 n분간 들어온 데이터 개수 집계 등의 작업을 처리할 수 있다.

내부 개념

스트림즈 DSL에서는 레코드의 흐름을 추상화한 KStream, KTable, GlobalKTable 개념이 존재한다.

KStream

레코드의 흐름을 표현하며, 메시지 키와 값으로 구성된다.
KStream은 컨슈머로 토픽을 구독하는 것과 동일한 선상에서 사용된다.
토픽 혹은 KStream의 모든 레코드를 조회할 수 있다.
각 파티션은 1개 태스크에 할당되어 사용된다.

KTable

KStream과 달리 메시지 키를 기준으로 묶어 사용한다.
유니크한 메시지 키를 기준으로 가장 최신 레코드를 조회할 수 있다.
동일한 메시지 키를 가진 레코드가 추가되면 최신 레코드로 덮어쓰여진다.
각 파티션은 1개 태스크에 할당되어 사용된다.

GlobalKTable

KTable과 동일하게 메시지 키를 기준으로 묶어 사용하지만, 모든 파티션이 각 태스크에 할당되어 사용된다.
따라서 어떤 태스크이더라도 파티션에 관계 없이 전역적으로 데이터에 접근 가능하므로, 코파티셔닝 되지 않은 KStream과 조인이 가능하다.

코파티셔닝
토픽 A의 데이터와 토픽 B의 데이터를 조인하고 싶을 때 두 토픽의 파티션 개수와 파티셔닝 전략이 같아야(코파티셔닝) 동일한 메시지 키를 가진 데이터가 동일한 태스크에 들어간다.
만약 코파티셔닝 되어있지 않다면 한쪽 토픽을 리파티셔닝하여 새로운 토픽에 새로운 메시지 키를 갖도록 재배열해주어 코파티셔닝되도록 할 수 있다.

각 태스크가 GlobalKTable의 모든 데이터를 저장하고 사용하기 때문에 프로세스의 메모리 사용량이 증가하고 네트워크, 브로커에 부하가 생긴다. 따라서 작은 용량의 데이터일 경우에만 사용하는 것이 좋다.

주요 옵션

필수 옵션

bootstrap.server
- 카프카 브로커 클러스터 주소
application.id
- 스트림즈 애플리케이션 구분을 위한 고유한 아이디를 설정한다.
- 이 값을 기준으로 병렬 처리를 한다.

선택 옵션

default.key.serde
- 레코드의 메시지 키를 직렬화/역직렬화하는 클래스 이름을 지정한다.
- 기본값은 바이트 직렬화/역직렬화 클래스인 Serdes.ByteArray().getClass().getName()이다.
default.value.serde
- 레코드의 메시지 값을 직렬화/역직렬화하는 클래스 이름을 지정한다.
- 기본값은 바이트 직렬화/역직렬화 클래스인 Serdes.ByteArray().getClass().getName()이다.
num.stream.threads
- 스트림 프로세싱 실행 시 사용될 스레드 개수를 지정한다.
- 기본값은 1이다.
state.dir
- 상태 기반 데이터 처리를 할 때 데이터 저장할 디렉토리를 지정한다.
- 기본값은 /tmp/kafka-streams이다.

제공하는 기능

stream() / to()

특정 토픽을 KStream 형태로 가져오는 stream 메서드와 KStream 데이터를 특정 토픽에 저장하는 to 메서드를 제공한다.

아래와 같이 기존에 존재하는 토픽의 데이터를 실시간으로 다른 토픽으로 복제하는 애플리케이션을 작성할 수 있다.

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_NAME);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("STREAM_LOG");

stream.to("STREAM_LOG_COPY");

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

filter()

메시지 키 또는 값을 필터링해 특정 조건에 맞는 데이터만 골라낼 수 있다.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("STREAM_LOG");
KStream<String, String> stream = stream.filter((key, value) -> value.length() > 5);
stream.to("STREAM_LOG_FILTER");

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

join()

데이터를 DB에 저장하지 않고도 메시지 키를 기준으로 조인하여 스트리밍 처리할 수 있다.
예를 들어 이름을 메시지 키, 주소를 메시지 값으로 갖는 KTable이 있고, 이름을 메시지 키, 주문 내역을 메시지 값으로 갖는 KStream이 있다면 이를 조인하여 물품과 주소가 결합된 데이터를 만들어낼 수 있다.
조인 시에는 반드시 코파티셔닝이 되어 있어야 하며 만약 되어있지 않은 상태라면 TopologyException이 발생한다.

StreamsBuilder builder = new StreamsBuilder();
KTable<String, String> table = builder.table("address");
KStream<String, String> stream = builder.stream("order");
// 동일한 메시지 키를 가진 데이터를 찾았다면 주문 내역과 주소를 합쳐 새로운 토픽에 저장한다.
KStream<String, String> stream = stream
    .join(table, (order, address) -> order + "send to " + address)
    .to("order_with_address");

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

코파티셔닝되어 있지 않은 토픽을 조인하려면 리파티셔닝하거나 GlobalKTable과 KStream 형태로 조인해야 한다.
GlobalKTable은 KTable의 조인과 달리 레코드 매칭 시 KStream의 메시지 키와 메시지 값 둘 다 사용할 수 있으므로 join 메서드의 두번째 인자에 어떤 값을 사용해 매칭할 지 전달해주어야 한다.

StreamsBuilder builder = new StreamsBuilder();
GlobalKTable<String, String> table = builder.globalTable("address");
KStream<String, String> stream = builder.stream("order");

KStream<String, String> stream = stream
    .join(table, (orderKey, orderValue) -> orderKey, (order, address) -> order + "send to " + address)
    .to("order_with_address");

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

프로세서 API

스트림즈DSL보다 투박하지만 토폴로지를 기준으로 데이터를 처리하는 같은 목적을 가진다.
스트림즈DSL에서 사용하는 KStream, KTable, GlobalKTable 개념이 존재하지 않는다.
Processor 인터페이스를 상속받는 스트림 프로세서 클래스를 만들고 토폴로지 객체를 생성해 프로세서를 사용하도록 코드를 구현해야 한다.
ProcessorContext 클래스를 통해 현재 스트림 처리 중인 토폴로지의 토픽 정보, 애플리케이션 아이디를 조회하거나 프로세싱 처리에 필요한 다양한 메서드를 사용할 수 있다.

public class FilterProcessor implements Processor<String, String> {
    private ProcessorContext context;
    
    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }
    
    @Override
    public void process(String key, String value) {
        if (value.length() > 5) {
            context.forward(key, value);
        }
        context.commit();
    }
    
    @Override
    public void close() {}
}

addSource 메서드의 첫 번째 인자로는 소스 프로세서 이름을 입력하고, 두 번째 인자로는 토픽 이름을 입력해야 한다.
addProcessor 메서드에는 스트림 프로세서 이름, 프로세서 객체, 부모 노드 이름을 입력한다.
addSink 메서드에는 싱크 프로세서의 이름, 저장할 토픽 이름, 부모 노드 이름을 입력한다.
생성 완료된 토폴로지 객체를 KafkaStreams 객체에 넣고 start 메서드를 통해 스트림즈 애플리케이션을 실행시킬 수 있다.

Topology topology = new Topology();
topology.addSource("Source", STREAM_LOG)
        .addProcessor("Process", () -> new FilterProcessor(), "Source")
        .addSink("Sink", STREAM_LOG_FILTER, "Process");

KafkaStreams streaming = new KafkaStreams(topology, props);
streaming.start();

PreviousConsumer NextKafka Connect

Last updated 4 days ago