kafka streams

[yeomko22]

kafka streams

• 토픽에 적재된 데이터를 실시간으로 변환하여 다른 토픽에 적재하는 라이브러리
• 카프카 스트림 처리를 위해 Spark, Flink, Storm, Fluentd와 같은 오픈 소스들도 존재
• 그럼에도 카프카 스트림의 장점은 일단 공식 지원 라이브러리. 카프카 버전이 오를 때마다 함께 릴리지 됨
• 신기능 추가가 활발함. 안정성이 매우 뛰어남

• 스트림즈 애플리케이션은 JVM 위에서 하나의 프로세스로 실행됨

카프카 스트림즈 애플리케이션

• 내부적으로 1개 이상의 스레드 생성이 가능
• 스레드는 1개 이상의 테스크를 가짐
• 테스크는 데이터 처리 최소 단위

topology

• 카프카에서는 토폴로지를 이루는 노드를 하나의 프로세서라고 부르고 노드와 노드를 이은 선을 스트림이라 부른다.
• 스트림은 토픽의 데이터를 뜻하며 이는 레코드와 동이랗다.
• 프로세서: 소스 프로세서, 스트림 프로세서, 싱크 프로세서 3가지가 있다.
  • 소스 프로세서: 토픽에서 데이터를 가져오는 역할
  • 스트림 프로세서: 다른 프로세서가 반환한 데이터를 처리하는 역할
  • 싱크 프로세서: 데이터를 특정 카프카 토픽으로 저장하는 역할을 수행, 최종 종착지
• 스트림즈DSL과 프로세서 API 2가지 방법으로 개발이 가능

스트림즈 DSL 데이터 처리 예시

• 메시지 값을 기반으로 한 토픽 분기 처리
• 지난 10분간 들어온 데이터의 개수 집계
• 토픽과 다른 토픽의 결합으로 새로운 데이터 생성

프로세서 API로 구현하는 데이터 처리 예시

• 메시지 값의 종류에 따라 토픽을 가변적으로 전송
• 일정한 시간 간격으로 데이터 처리

[yeomko22]

StreamsDSL

• streamsDSL에서 3가지 새로운 개념인 등장, KStream, KTable, GlobalKTable

KStream

• 레코드의 흐름을 표현한 것, 메세지 키와 메세지 값으로 구성
• KStream으로 데이터 조회하면 토픽에 존재하는 모든 레코드를 출력

KTable

• key를 기준으로 묶어서 사용, key value 1:1 매칭

GlobelKTable

• KTable로 선언된 토픽은 1개 파티션이 1개 테스크에 할당
• GlobalKTable로 선언도니 토픽은 모든 파티션 데이터가 각 태스크에 할당되어 사용된다는 차이점
• Kstraem과 KTable이 데이터 조인을 수행할 때 GlobalKTable사용

[yeomko22]

streams application example

public class SimpleStreamApplication {
    private static String APPLICATION_NAME = "streams-application";
    private static String BOOTSTRAP_SERVER = "my-kafka:9092";
    private static String STREAM_LOG = "stream_log";
    private static String STREAM_LOG_COPY = "stream_log_copy";
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_NAME);
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        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> streamLog = builder.stream(STREAM_LOG);
        streamLog.to(STREAM_LOG_COPY);
        
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

• application name을 지정해주어야 함. 이를 기준으로 병렬 처리 수행. 컨슈머 그룹과 비슷한 개념
• StreamBuilder는 스트림 토폴로지를 정의하기 위한 용도로 사용
• stream_log 토픽으로부터 KStream 객체를 만들기 위해 stream() 메서드를 사용. 만일 table() 호출시 KTable, globalTable() 호출 시 GlobalKTable을 만들어준다.
• 이들 메서드들은 최초의 토픽 데이터를 가져오는 소스 프로세서들이다.
• to 메서드를 사용하여 데이터를 다른 토픽으로 전송한다. 즉, to 메서드는 싱크 프로세서이다.
• builder에 저장된 정보를 바탕으로 KafkaStreams 객체를 만들어 준 뒤, start를 호출해준다.
• 위 예시는 소스 프로세서, 싱크 프로세서로 구성된 토폴로지로 데이터 처리를 담당하는 스트림 프로세서가 없다.

[yeomko22]

streamsDSL filter

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

• filteredStream을 중간에 추가, 이는 데이터 처리를 담당하는 스트림 프로세서이다.

[yeomko22]

streamsDSL KTable, KStream join

        StreamsBuilder builder = new StreamsBuilder();
        KTable<String, String> addressTable = builder.table(ADDRESS_TABLE);
        KStream<String, String> orderStream = builder.stream(ORDER_STREAM);
        orderStream.join(addressTable,
                (order, address) -> order + " send to " + address)
                .to(ORDER_JOIN_STREAM);

• KTable과 KStream은 메세지 키를 기준으로 조인 가능
• 즉, 카프카는 실시간으로 들어오는 데이터들을 조인할 수 있다. 이를 이용해 사용자의 이벤트 데이터를 데이터 베이스에 저장하지 않고도 조인해서 스트리밍 처리를 할 수 있는 장점이 있다.
• KTable과 Kstream 조인을 구현할 때에는 반드시 코파티셔닝이 되어있는지 확인할 것.
• 코파티셔닝이란 동일한 파티션 개수, 동일한 파티셔닝을 사용하는 것을 말한다.
• KTable의 경우 동일한 메세지 키 일 경우 마지막 레코드를 유효한 것으로 친다. 그러므로 KTable에 값이 업데이트 되면 이후에 조인되는 메세지들은 최신 값을 기반으로 조인된다.

[yeomko22]

streamsDSL - GlobalKTable과 KStream join

코파티셔닝이 되어 있지 않은 토픽을 조인해야할 때 가능한 두 가지 방법

• 리파티셔닝을 수행하여 코파티셔닝 해준 상태에서 조인처리
• KTable로 사용하는 토픽을 GlobalKTable로 선언하여 사용하는 것

GlobalKTable join example

        StreamsBuilder builder = new StreamsBuilder();
        GlobalKTable<String, String> addressGlobalTable = builder.globalTable(ADDRESS_GLOBAL_TABLE);
        KStream<String, String> orderStream = builder.stream(ORDER_STREAM);
        orderStream.join(addressGlobalTable,
                        (orderKey, orderValue) -> orderKey,
                        (order, address) -> order + " send to " + address)
                        .to(ORDER_JOIN_STREAM);

• KTable과 그닥 달라보이지 않지만 GlobalKTable로 선언한 토픽은 토픽에 존재하는 모든 데이터를 테스크마다 저장, 조인 처리를 수행한다.
• 조인 수행 시엔 KStream의 메시지 키 뿐만 아니라 메시지 값을 기준으로도 매칭하여 조인할 수 있다.

[yeomko22]

ProcessorAPI

• streamsDSL로 구현하지 못하는 상세 로직 구현이 필요하다면 processor API 활용이 가능하다.
• processorAPI에서는 KStream, KTable, GLobalKTable 개념이 없다는 것에 주의하라.

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