Chapter 14. 스트림 처리

hubtwork commented 2 weeks ago

Chapter Ownership : @iamzin

iamzin commented 1 week ago

Chapter 14. 스트림 처리

카프카를 활용하면서 내용상으로 중요하고 완벽한 순서를 유지하면서 오랫동안 저장되어 있는 데이터 스트림 처리가 필요했다. 카프카 스트림즈를 사용하면 외부 처리 프레임워크에 의존할 필요 없이 애플리케이션에 이벤트를 읽고, 처리하고, 쓰는 기능을 구현할 수 있다.

14.1 스트림 처리란 무엇인가?

데이터 스트림

무한히 늘어나는 데이터세트(unbounded dataset)을 추상화한 것
시간이 흐름에 따라 새로운 레코드가 계속해서 추가되기 대문에 데이터세트가 무한해지는 것

이벤트 스트림

우리가 분석하고자 하는 모든 비즈니스 활동을 나타낼 수 있음

이벤트 스트림은 순서가 있다
- 테이블의 레코드는 순서가 없는 것으로 간주되나 이벤트는 다른 이벤트 전/후에 발생했다는 의미를 가짐
데이터 레코드는 불변(immutable)하다
- 이벤트는 발생한 뒤 고칠 수 없으며, 이벤트 스트림은 모든 트랜잭션을 포함하기 때문에 작업 내역들이 모두 기록됨
이벤트 스트림은 재생(replay)가 가능하다
- 예전에 발생한 raw stream을 그대로 재생할 수 있음(새로운 분석 방법을 시도하거나, 감사를 수행하는 등에 활용됨)

프로그래밍 패러다임

요청-응답
- 보통 블로킹(blocking) 방식으로 요청 후 으압을 보내 줄 때까지 대기하는 것이 보통이며, 데이터베이스에서는 이 패러다임이 OLTP(Online Transcation Processing)으로 알려져 있음
배치 처리
- 처리량이 크고 지연이 큰 편임
- 최근의 비즈니스는 보다 시기적절하고 효율적인 의사 결정을 위해 더 짧은 시간 간격 안에 사용 가능한 데이터를 필요로 함
스트림 처리
- 연속적이며 논블로킹(nonblocking)하게 동작하는 방식
- 비즈니스 리포트가 지속적으로 업데이트되고 비즈니스 애플리케이션들이 계속해서 응답할 수 있기 때문에 응답을 기다릴 필요 없이 처리를 진행할 수 있음

14.2 스트림 처리 개념

스트림 처리는 일반적인 데이터 처리와 매우 비슷하다. 데이터를 읽고, 처리를 한 뒤, 결과물을 기록한다.

14.2.1 토폴로지(topology)

소스 스트림, 스트림 프로세서의 그래프, 싱크 스트림이 서로 연결된 것

14.2.2 시간

스트림 애플리케이션이 시간 윈도우(time window)에 대해 작업을 수행하는 만큼 시간에 대해 공통적인 개념을 가져야 함

이벤트 시간
- 다루고자 하는 이벤트가 발생하여 레코드가 생성된 시점
- 레코드에 이벤트 시간 필드를 추가하여 데이터가 추가된 시간과 함께 활용
로그 추가 시간(접수 시간)
- 이벤트가 카프카 브로커에 전달되어 저장된 시점
- 이벤트 시간이 기록되지 않는 경우 레코드가 생성된 다음부터 변하지 않는 값인 만큼 이벤트 시간에 대한 합리적인 근사값으로 볼 수 있음
처리 시간
- 스트림 처리 애플리케이션이 연산을 수행하기 위해 이벤트를 받은 시간
- 이벤트가 발생한 뒤 한 참 뒤 일 수 있음
- 따라서 신뢰성이 떨어지는 데이터

카프카 스트림즈는 TimestampExtractor 인터페이스를 사용해 각 이벤트에 시간을 부여한다. 이 인터페이스의 서로 다른 구현체를 사용하여 위 3가지 시간 개념 중 하나를 사용하거나 타임스탬프를 결정하여 완전히 다른 시간 개념을 사용할 수도 있다.

[!NOTE]

카프카 스트림즈가 결과물을 카프카 토픽에 쓸 때 '이벤트에 타임스탬프를 부여하는 규칙'

결과 레코드가 입력으로 주어진 레코드에 직접적으로 대응될 경우: 결과 레코드는 입력 레코드와 동일한 타임스탬프를 사용

결과 레코드가 집계(aggregation) 연산의 결과물일 경우: 집계에 사용된 레코드 타임스탬프의 최대값을 결과 레코드의 타임스탬프로 사용

결과 레코드가 두 스트림을 조인한 결과일 경우: 조인된 두 레코드 타임스탬프 중 큰 쪽의 타임스탬프를 결과 레코드의 타임스탬프로 사용

결과 레코드가 스트림과 테이블을 조인한 경우: 스트림 레코드 쪽의 타임스탬프 사용

입력과 상관없이 특정한 스케줄에 따라 데이터를 생성한 결과 레코드의 경우: 스트림 처리 애플리케이션의 현재 내부 시각에 따라 결정 (ex: 카프카 스트림즈의 punctuate())

[!NOTE] Timezone 주의 전체 데이터 파이프라인이 표준화된 시간대 하나만 쓰거나, 서로 다른 시간대의 데이터 스트림을 다뤄야 한다면 윈도우에 작업을 수행하기 전에 이벤트 시각을 하나의 시간대로 변환해두어야 한다. 혹은 레코드에 시간대 정보를 저장해 넣는 경우도 있다.

14.2.3 상태

로컬 혹은 내부 상태
- 스트림 처리 애플리케이션의 특정 인스턴스에서만 사용할 수 있는 상태
- 내장형 인메모리 데이터베이스를 사용해서 유지 관리
- 장점: 엄청나게 빠르다 / 단점: 사용 가능한 메모리 크기의 제한
- 스트림 처리의 많은 디자인 패턴들은 데이터 분할 후 한정된 크기의 상태로 서브스트림으로 만드는 데 이 상태를 활용
외부 상태
- NoSQL 시스템을 사용해 저장(ex: 카산드라)
- 장점: 크기에 제한이 없으며 여러 다른 애플리케이션에서도 접근이 가능 / 단점: 다른 시스템을 추가하는 데 지연 증가, 복잡도 증가, 가용성 문제

※ 애플리케이션 내부 상태와 외부 상태를 일관적으로 유지해야하는 점을 고려해야 함

14.2.4 스트림-테이블 이원성

스트림은 변경 내역을 저장한다. 변경을 유발하는 이벤트의 연속이다. 예를 들어 재고 변경 스트림을 통해 재고 변경을 구체화할 수 있다.

14.2.5 시간 윈도우

스트림 작업은 시간을 윈도우라 불리는 구간 단위로 잘라서 처리한다. 두 스트림을 조인하는 작업 역시 윈도우 작업이다.

윈도우 크기
- 윈도우 크기가 커질 수록 랙(lag)이 커짐
- 카프카 스트림은 윈도우의 크기가 inactivity 길이에 따라 결정되는 세션 윈도우를 지원
- 세션 간격을 정의하면, 세션 간격보다 작은 시간 간격을 두고 연속적으로 도착한 이벤트들은 하나의 세션에 속하게 됨
- 당연히 세션 간격 이상으로 이벤트가 도착하지 않으면 새로운 세션이 생성됨
- 시간 윈도우의 진행 간격
- 호핑 윈도우: 윈도우의 크기와 윈도우 사이의 고정된 시간 간격이 같은 경우
  - ex: 5분마다 이동하는 5분 길이의 윈도우
- 텀블링 윈도우: 진행 간격과 윈도우 크기가 같은 경우
  - ex: 1분마다 이동하는 5분 길이의 윈도우 (윈도우끼리 서로 겹치기 때문에 이벤트는 다수의 윈도우에 속함)
*윈도우를 업데이트할 수 있는 시간(grace period)
- 이벤트가 이벤트에 해당하는 윈도우에 추가될 수 있는 기한을 정의할 수 있는 것이 이상적

14.2.6 처리 보장

장애가 발생했을 경우에도 각각의 레코드를 한 번만 처리할 수 있어야 한다. 즉, '정확히 한 번' 보장이 없는 경우 스트림 처리는 정ㅇ확한 결과가 요구되는 상황에서 쓸모가 없다.

카프카 스트림즈는 카프카의 트랜잭션 기능을 사용해서 스트림 처리 애플리케이션에서 '정확히 한 번' 보장을 지원한다.

카프카 스트림즈를 사용하면 processing.guarantee=exactly_once 로 설정하여 '정확히 한 번'을 활성화
processing.guarantee=exactly_once_beta 를 설정하여 '더 효율적인 정확히 한 번' 구현체를 활성화

[!NOTE] 버전 3.0 이후의 처리 보장 processing.guarantee=exactly_once_v2 사용을 권장 (processing.guarantee=exactly_once와 processing.guarantee=exactly_once_beta는 버전 4.0부터 제거될 예정)

14.3 스트림 처리 디자인 패턴

14.3.1 단일 이벤트 처리

가장 단순한 스트림 처리 패턴으로 이벤트를 개별적으로 처리하는 것이다. map/filter 패턴이라고도 한다.

불필요한 이벤트를 스트림에서 걸러내거나 각 이벤트를 변환하기 위해 사용되는 패턴
스트림의 이벤트를 읽어와 각각 처리한 뒤, 처리된(수정된) 이벤트를 다른 스트림에 기록
ex: 에러 로그 레벨이 ERROR이면 높은 우선순위 스트림에, 나머지를 낮은 우선순위 스트림에 기록 / JSON 이벤트를 읽어와 AVRO 형식으로 사용
각 이벤트가 독립적으로 처리될 수 있기 때문에 상태를 유지할 필요도, 복구할 필요가 없으며, 이 때문에 장애 복구나 부하 분산이 매우 쉬운 편

14.3.2 로컬 상태와 스트림 처리 (그룹별 집계)

대부분 윈도우 집계와 같이 정보의 집계에 초점을 맞춘다. 집계를 위해 스트림의 상태(state)를 유지해야 한다.

로컬 상태를 사용하는 경우
- 전체 주식이 아닌 주식 종목별로 분류해야 하는 경우
  
  카프카 파티셔너를 통해 동일한 주식에 대한 모든 이벤트가 동일한 파티션에 쓰여지도록 처리 -> 각각의 애플리케이션 인스턴스는 자신에게 할당된 파티션에 저장된 모든 이벤트를 컨슘 -> 각 애플리케이션 인스턴스는 자신에게 할당된 파티션에 쓰여진 전체 주식 종목의 부분집합에 대한 상태 유지가 가능
메모리 사용
- 로컬 상태는 애플리케이션 인스턴스가 사용 가능한 메모리 안에 들어갈 수 있는 것이 가장 이상적 (디스크에 저장하는 기능은 성능에 상당한 영향을 미치므로 주의)
영속성
- 애플리케이션 인스턴스가 종료되거나 재실행되거나 다른 인스턴스에 의해 대체되었을 때 복구되어야 함
- 카프카 스트림즈는 이를 내장된 RocksDB를 사용하여 로컬 상태를 인메모리 방식으로 저장함과 동시에 위와 같은 상황들에서 빠르게 복구가 가능하도록 디스크에 데이터를 영속적으로 저장
- 이 때, 로컬 상태에 대한 모든 변경 사항은 카프카 토픽에도 보내지므로 스트림 처리를 담당하고 있는 노드에 장애가 발생하더라도 토픽으로 부터 이벤트를 읽어올 수 있어 로컬 상태가 유실되지 않음
- 추후 로컬 캐시를 재구성할 때 활용
- 토픽이 과도하게 쌓이는 것을 방지하고 상태 복구를 언제든지 실행 가능하게 하기 위해 로그 압착을 사용
리밸런싱
- 파티션은 서로 다른 컨슈머에게 다시 할당될 수 있으므로, 재할당이 발생하면 파티션을 상실한 애플리케이션 인스턴스는 마지막 상태를 저장하여 해당 파티션을 할당받은 인스턴스가 재할당 이전 상태를 복구시킬 수 있도록 해야 함

14.3.3 다단계 처리/리파티셔닝 (전체 집계)

전체 집계가 필요한 경우 정보가 서로 다른 인스턴스에 할당된 파티션에 분산되어있을 수 있다.

각 프로세서가 맡고 있는 파티션에 대한 결과물 산출 -> 이 결과물을 별도 토픽의 단일 파티션에 적재 -> 또 다른 프로세서가 결과물을 종합하여 최종 결과물을 산출

14.3.4 외부 검색을 상요하는 처리: 스트림-테이블 조인

데이터베이스에 저장된 데이터를 사용해 검증하거나 정보를 조합해 확장할 수 있다.

이벤트 토픽 발생 -> 데이터베이스의 데이터를 통한 확장(조합) -> 확장(조합)된 새 이벤트를 다른 토픽에 적재

외부 검색이 각각의 레코드를 처리하는 데 있어 지연 발생
따라서 스트림 처리 애플리케이션 안에 데이터베이스에 저장된 데이터를 캐시
캐시의 정보가 만료되지 않도록 혹은 항상 최신으로 유지할 수 있도록, 데이터베이스 테이블의 모든 변경점을 이벤트 스트림에 담아 스트림 처리 작업이 이 스트림을 받아와 캐시를 업데이트 하는데 사용

[!NOTE] CDC(Change Data Capture) 데이터베이스 변경 내역을 이벤트 스트림으로 받아오는 것 (카프카 커넥트의 커넥터들은 CDC를 수행하며 데이터베이스 테이블을 변경 이벤트 스트림으로 변환할 수 있음)

14.3.5 테이블-테이블 조인

두 테이블을 조인하는 것은 윈도우 처리되지 않는 연산이기 때문에 작업이 실행되는 시점의 데이터 상태를 조인한다. 카프카 스트림에서는 동일한 방식으로 파티션된 동일한 키를 가지는 두 테이블에 대해 동등 조인을 수행할 수 있다.

카프카 스트림즈는 두 테이블에 대해 외래 키 조인을 지원
한 스트림 혹은 테이블의 키와 다른 스트림 혹은 테이블의 필드를 조인
Kafka Summit 2020: Crossing the Streams

14.3.6 스트리밍 조인

스트림과 스트림을 조인해야할 경우가 있다. 이 조인에는 항상 이동하는 시간 윈도우가 수반되므로 스트리밍 조인을 윈도우 조인이라고도 부른다.

내용A를 담은 스트림과 내용B를 담은 스트림을 조인 -> 내용A와 B를 조합한 결과물 산출을 위해 내용A를 특정 기간을 기준으로 하되 특정 시간 윈도우 범위 안에 있는 내용B 하고 조합 -> 즉, 각 스트림에 대해 같은 길이를 가지는 윈도우를 유지하면서 각 윈도우에 속한 이벤트끼리 맞춰야 함

카프카 스트림즈는 조인할 두 스트림이 똑같이 조인 키에 대해 파티셔닝 되어 있을 경우 동등 조인을 지원
카프카 스트림즈는 두 토픽에 대한 조인 윈도우를 내장된 RocksDB 상태 저장소에 유지함으로써 조인을 수행

14.3.7 비순차 이벤트

네트워크 등 여러 이유에 의해 잘못된 시간에 스트림에 도착한 이벤트를 처리하는 Out of sequence 이벤트는 자주 발생할 수 있다.

대응 방법

이벤트가 Out of sequence 상태임을 인지
- 애플리케이션이 이벤트 시간을 확인해 현재 시각보다 이전인지 확인
Out of sequence 이벤트의 순서(sequence)를 복구할 수 있는 시간 영역을 정의
- ex: 3시간 차이의 경우는 복구하지만, 3주 이상 오래된 경우는 포기
순서를 복구하기 위해 이벤트를 그룹핑
- 계속해서 돌아가는 동일한 프로세스가 현시점을 기준으로 오래된 이벤트와 새로운 이벤트를 모두 처리
결과 변경이 가능하도록 설정
- 로컬 상태에 다수의 집계 윈도우를 변경 가능한 형태로 유지하며, 이 윈도우를 얼마나 오랫동안 유지할지를 설정 (집계 윈도우를 더 오랫동안 변경 가능한 형태로 유지할수록 로컬 상태를 유지하기 위한 메모리가 더 많이 필요해짐)
- 보통 구글의 Dataflow나 카프카 스트림즈 등의 스트림 처리 프레임워크가 처리 시간과 독립적인 이벤트 시간의 개념을 자체적으로 지원하며, 현재 처리 시간 이전 혹은 이후의 이벤트 시간을 가지는 이벤트를 다룰 수 있는 기능을 제공
- 카프카 스트림즈는 집계 결과를 결과 토픽에 적재하며, 이 토픽들은 대부분 로그 압착이 설정되어 있음

14.3.8 재처리하기

방법1

사례: 구번전에서 사용하던 이벤트 스트림을 신버전 애플리케이션에서 읽어와 산출된 새로운 결과 스트림을 쓴다. 구버전의 결과를 교체하는 것이 아닌 일정 기간 동안 두 버전의 결과를 비교한 뒤, 특정 시점에 구버전 대신 신버전 결과를 사용하도록 한다.

신버전을 새 컨슈머 그룹으로 실행

신버전이 입력 토픽의 첫 오프셋부터 처리를 시작하도록 설정하여 입력 스트림의 모든 이벤트에 대한 복사본을 적재

신버전이 처리를 계속하고, 신버전이 처리 작업을 따라잡았을 때 새로운 결과 스트림으로 전환

방법2

사례: 기존 스트림 처리 애플리케이션에 버그가 많아서 버그를 고친 뒤 이벤트 스트림을 재처리하여 결과를 재산출 한다.

이미 존재하는 애플리케이션을 초기화하여 입력 스트림의 맨 처음부터 다시 처리하도록 처리

로컬 상태를 초기화하고 기존 출력 스트림 내용을 삭제

카프카 스트림즈가 스트림 처리 상태를 초기화하기 위한 툴을 제공하지만, 용량이 충분한 경우 방법1이 버전 간 전환이 가능하고, 버전 간 결과물 비교가 가능하며, 정리(clean up) 과정에서 중요한 데이터가 유실되거나 에러가 발생할 위험이 낮아서 더 권장된다.

14.4 예제로 보는 카프카 스트림즈

14.4.1 단어 개수 세기

단일 이벤트 처리 패턴(맵/필터 패턴)

filter(): 기준에 맞게 필터링
groupByKey(): 키값 기준으로 그룹화하여 각 단어별 이벤트 집합 생성

count(): 결과 산출을 위한 메서드 실행

public class WordCountExample {
public static void main(String[] args) throws Exception{
    ...
    StreamsBuilder builder = new StreamsBuilder();

    KStream<String, String> source = builder.stream("wordcount-input");

    final Pattern pattern = Pattern.compile("\\W+");
    KStream counts  = source.flatMapValues(value-> Arrays.asList(pattern.split(value.toLowerCase())))
            .map((key, value) -> new KeyValue<Object, Object>(value, value))
            .filter((key, value) -> (!value.equals("the")))
            .groupByKey()
            .count().mapValues(value->Long.toString(value)).toStream();
    counts.to("wordcount-output");

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

    streams.start();
}
}

14.4.2 주식 시장 통계

집계

groupByKey(): 이벤트 스트림이 레코드 키 기준으로 파티셔닝 되도록 설정 (groupByKey()를 실행하기 전 작업 없이, 토픽에 데이터를 쓸 때 키값을 가지는 데이터를 키값 기준으로 파티셔닝될 뿐)
windowedBy(): 윈도우 정의 (windowSize만큼 매초 전진)
aggregate(): 스트림을 서로 중첩되는 윈도우들로 나눈 뒤 각 윈도우에 배정된 모든 이벤트에 대해 집계 연산을 적용 (집계 결과를 저장할 새 객체를 aggregate() 와 같이 설정
(k, v, tradestats) -> tradestats.add(v): 실제 집계를 수행하는 메서드
Materialized(): 상태 저장소 설정 (withValueSerde로 집계 결과를 직렬화/역직렬화 하기 위한 Serde 객체 지정)
toStream(): 이벤트 스트림화

mapValues(): 집계 결과를 사용해 최종 결과를 계싼하여 출력 스트림으로 내보냄

public class StockStatsExample {
public static void main(String[] args) throws Exception {
    ...
    StreamsBuilder builder = new StreamsBuilder();

    KStream<String, Trade> source = builder.stream(Constants.STOCK_TOPIC);

    KStream<Windowed<String>, TradeStats> stats = source
            .groupByKey()
            .windowedBy(TimeWindows.of(Duration.ofMillis(windowSize)).advanceBy(Duration.ofSeconds(1)))
            .<TradeStats>aggregate(() -> new TradeStats(),
                    (k, v, tradestats) -> tradestats.add(v),
                    Materialized.<String, TradeStats, WindowStore<Bytes, byte[]>>as("trade-aggregates")
                            .withValueSerde(new TradeStatsSerde()))
            .toStream()
            .mapValues((trade) -> trade.computeAvgPrice());

    stats.to("stockstats-output", Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class, windowSize)));

    Topology topology = builder.build();

    KafkaStreams streams = new KafkaStreams(topology, props);

    streams.start();
}
}

14.4.3 클릭 스트림 확장

스트리밍 조인

leftJoin(): 스트림-테이블 조인으로 스트림의 각 이벤트는 캐시된 사본에서 정보를 받아오거나, 스트림-스트림 조인에 사용
JoinWindows.of(): 스트림-스트림 조인은 시간 윈도우를 사용하는 조인이기에 모든 스트림을 조인하는 것이 아닌 특정 길이의 조인 윈도우를 정의

StreamJoined.with(key, value1, value2): 조인 결과에 대한 Serde를 정의

public class ClickstreamEnrichment {
public static void main(String[] args) throws Exception {
    ...
    StreamsBuilder builder = new StreamsBuilder();

    KStream<Integer, PageView> views = builder.stream(Constants.PAGE_VIEW_TOPIC, Consumed.with(Serdes.Integer(), new PageViewSerde()));
    KTable<Integer, UserProfile> profiles = builder.table(Constants.USER_PROFILE_TOPIC, Consumed.with(Serdes.Integer(), new ProfileSerde()));
    KStream<Integer, Search> searches = builder.stream(Constants.SEARCH_TOPIC, Consumed.with( Serdes.Integer(), new SearchSerde()));

    // 스트림-테이블 조인: 정보 확장
    KStream<Integer, UserActivity> viewsWithProfile = views.leftJoin(profiles,
                (page, profile) -> {
                    if (profile != null)
                        return new UserActivity(profile.getUserID(), profile.getUserName(), profile.getZipcode(), profile.getInterests(), "", page.getPage());
                    else
                       return new UserActivity(-1, "", "", null, "", page.getPage());
    });

    // 스트림-스트림 조인: 시간 윈도우 기준
    KStream<Integer, UserActivity> userActivityKStream = viewsWithProfile.leftJoin(searches,
            (userActivity, search) -> {
                if (search != null)
                    userActivity.updateSearch(search.getSearchTerms());
                else
                    userActivity.updateSearch("");
                return userActivity;
            },
            JoinWindows.of(Duration.ofSeconds(1)),
            StreamJoined.with(Serdes.Integer(), new UserActivitySerde(), new SearchSerde()));

    userActivityKStream.to(Constants.USER_ACTIVITY_TOPIC, Produced.with(Serdes.Integer(), new UserActivitySerde()));

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

    streams.start();
}
}

14.5 카프카 스트림즈: 아키텍처 개요

14.5.1 토폴로지 생성하기

모든 이벤트가 입력에서 출력으로 이동하는 동안 수행되는 작업과 변환 처리의 집합
위 14.4의 코드들이 토폴로지

14.5.2 토폴로지 최적화하기

카프카 스트림즈 실행 단계

KStream, KTable 객체를 생성하고 여기에 필터, 조인과 같은 DSL 작업을 수행함으로써 논리적 토폴로지 정의
StreamBuilder.build() 메서드가 논리적 토폴로지로부터 물리적 토폴로지를 생성
KafkaStreams.start()가 토폴로지 실행하여 데이터 읽고, 처리, 쓰기

최적화는 2단계에 적용
대부분 가능한 한 토픽을 재활용하도록 StreamsConfig.TOPOLOGY_OPTIMIZATION=StreamsConfig.OPTIMIZE 설정 및 build(props) 호출하여 활성화

14.5.3 토폴로지 테스트하기

단위 테스트

툴: TopologyTestDriver
단위테스트와 비슷하게 입력 데이터 정의, 목업 입력 토픽에 데이터 적재, 테스트 드라이버로 토폴로지 실행, 목업 출력 토픽에서 결과 읽은 뒤 비교 검증이 가능
단, 캐시 기능을 시뮬레이션 해주지는 않음

통합 테스트

툴: EmbeddedKafkaCluster, Testcontainers
EmbeddedKafkaCluster: 테스트를 수행하는 JVM 상에 카프카 브로커를 하나 띄워줌
(권장) Testcontainers: 도커 컨테이너를 사용해 카프카 브로커와 기타 테스트에 필요한 요소들을 띄워줌

[!NOTE] Testing Kafka Streams - A Deep Dive

14.5.4 토폴로지 규모 확장하기

카프카 스트림즈는 하나의 애플리케이션 인스턴스 안에 다수의 스레드가 실행될 수 있게 함으로써 규모 확장과 서로 다른 애플리케이션 인스턴스 간에 부하 분산이 이루어지도록 한다.

토폴로지 실행을 다수의 태스크로 분할하여 병렬 처리
처리하는 토픽의 파티션 수에 따라 태스크 수 결정
각 태스크는 하나의 파티션을 책임지져 자신이 담당하는 파티션들을 구독하여 이벤트를 읽어오고 처리하여 결과를 싱크
애플리케이션의 스레드 수를 결정하여 처리하는 토픽의 파티션 수만큼 태스크를 생성하도록 설정 가능

태스크간 의존성 발생 케이스

스트림-스트림 조인
- 다수의 파티션에서 입력을 가져와 처리해야하는 경우 조인 작업에 필요한 모든 파티션들을 하나의 태스크에 할당하여 태스크가 필요한 파티션으로부터 독립적으로 조인을 수행할 수 있도록 구성 (카프카 스트림즈가 조인 작업에 사용될 모든 토픽에 대해 동일한 조인 키로 파티션된 동일한 수의 파티션을 가질 것을 요구하는 이유)
리파티셔닝
- 리파티션이 호출되면 새로운 키와 파티션을 가지고 새로운 토픽에 이벤트를 적재하고, 그 다음으로 오는 태스크들이 새 토픽에서 이벤트를 읽어와 처리
- 즉, 전체 토폴로지를 2개의 서브 토폴로지로 분할 (두 번째 서브 토폴로지는 첫 번째에 의존)
- 각 토폴로지 태스크 집합들은 독립적이고 병렬로 실행되기 때문에 태스크 사이 통신이나 공유된 리소스가 필요 없고 동일한 스레드나 서버에서 실행될 필요가 없음

14.5.5 장애 처리하기

장애가 발생하기 전 마지막으로 커밋된 오프셋을 카프카에서 가져와 처리하던 스트림의 마지막으로 처리된 지점부터 처리를 재개할 수 있다. 단, 로컬 상태 저장소가 유실됐을 경우 카프카로부터 체인지로그를 읽어와 로컬 상태 저장소를 복구한다.

태스크 고가용성을 지원하기 위해 컨슈머 코디네이션 기능 사용
- 태스크에 장애가 발생했지만 다른 스레드 혹은 인스턴스가 정상 작동 중일 경우 해당 태스크는 사용 가능한 다른 스레드에서 재시작
- 이는 컨슈머 그룹에 속한 컨슈머 중 하나에 장애가 발생할 경우 거기에 할당 됐던 파티션을 남은 컨슈머 중 하나에 할당해주어 장애를 대응하는 것과 유사
저장된 상태 복구
- 태스크를 다른 스레드가 넘겨받아 재시작할 때 현재 집계중인 윈도우와 같은 상태를 복구해야 함

복구 시간을 줄이는 것은 상태를 복구시키는 시간을 줄이는 문제와 동일하다. 따라서 아래 방법들을 사용한다.

카프카 스트림즈 토픽에 강력한 압착 설정
- min.compaction.lag.ms를 낮춤
- 세그먼트 크기를 기본값 1GB 대신 100MB 정도로 낮춤 ※ 각 파티션의 마지막 즉, 활성화된 세그먼트는 압착되지 않음
스탠바이 레플리카 설정
- 스트림 처리 애플리케이션에서 현재 작동 중인 태스크를 따라가는 태스크
- 다른 서버에서 현재 상태를 유지하는 역할
- 이 태스크가 최신의 현재 상태를 보유하고 있기 때문에 장애 시 처리를 재개할 수 있음

[!NOTE] Kafka Streams - State restoration during workload rebalance

14.6 스트림 처리 활용 사례

각자 읽어보자.

14.7 스트림 처리 프레임워크 선택하기

상황 별 솔루션

데이터 수집
- 스트림 처리 시스템이 필요한건지, 더 단순한 카프카 커넥트가 필요한건지 고민 필요
- 스트림 처리 시스템이 필요하다면 데이터 전송을 위한 커넥터가 충분한지, 기능이 충분한지 확인 필요
밀리초 단위 작업
- 이 경우는 요청-응답 패턴이 더 나음
- 그럼에도 필요하다면 마이크로배치 방식보단 이벤트 단위 low-latency 방식을 지원하는 쪽을 권장
비동기 마이크로 서비스
- 사용하고 있는 메세지 버스(ex: 카프카)와 잘 통합되고, 업스트림의 변경 사항을 마이크로 서비스의 로컬 상태에 쉽게 반영할 수 있으며, 로컬 상태를 캐시 혹은 구체화된 휴? 형태로 활용 가능한 스트림 처리 시스템이 필요
준 실시간 데이터 분석
- 로컬 저장소를 잘 지원해주는 스트림 처리 시스템이 필요
- 로컬 캐시 없이는 구현하기 까다로운 복잡한 집계 연산, 윈도우, 조인 등을 잘 지원하는 것이 중요
- 가령 커스텀 집계, 윈도우, 다양한 조인 타입을 지원하는 API 등이 필요

보편적 고려사항

시스템 운용성
- 프로덕션 배포 환경 용이, 모니터링과 트러블슈팅 용이, 규모 확장이나 축소 가능여부, 현재의 인프라스트럭처와의 통합 유연성, 재처리에 대한 대응 등을 고려
사용 및 디버깅 용이성
어려운 일을 쉽게 해줌
- 규모 확장과 장애 복구 등에 있어 세부 작동을 알아서 처리해주는지, 혹은 애매한 추상으로 상황 제어가 필요한지 등을 고려
커뮤니티
- 정보 찾는 데에 용이한지 고려

iamzin commented 1 week ago

진짜 재밌게 읽었다. 여러 장점들과 기능을 나열해두어 활용할 사례들이 떠올랐는데, 마지막에 고려해야할 사항이 있어서 다시 한 번 검토해볼 수 있도록 해준 점이 인상 깊었다. 근데 결국 카프카 스트림즈 짱! 이라고 하는 것 같았다.

CHOICORE commented 1 week ago

스트림 처리와 사용사례에 대해 알게 되서 좋은 챕터였다. 꼭 한번에 해내려고 하지 않아도 된다는 것, 큰 문제를 작게 나눠 작은 문제로 만들듯 하나의 페이즈 별로 처리하는 방법들을 고민해보려고 한다. 스트림이라는 개념에서 사실 경계해야 하는 부분은 손실이다. 킹치만 카프카는 다 해준다고 ㄹㅇㅋㅋ 다들 고생했습니다.

zinokim commented 1 week ago

소감

지금까지 공부했던 카프카 내용들도 매우 좋았는데, 알고보니 지금까지 공부했던 것들은 기본편이고 카프카 스트림이 고급 심화편이라고 느껴졌다. 카프카를 활용해서 시스템을 확장시키고 이렇게까지 할 수 있다고 보여주는 것 같았다. 카프카 스트림이 아니더라도 엔지니어로서 가져야 할 시각과 방향성도 보여주고 있어 유익했지만, 아직 나에게 고급 심화편은 어렵긴 하다.

hubtwork commented 1 week ago

이번 랭킹 시스템의 컨슈머 파이프라이닝 설계할 때, 카프카 스트림즈로 제어하면 좀 더 AGG 가 쉽지 않을까? 하고 고민했었는데 ( 물리적 시간이 없어 포기 ) 주식 예제랑 클릭 예제가 딱 적절한 처리를 할 수 있음을 보여주는 예시여서 좋았다. 하지만, 아직까지는 경험치 자체가 적어서 섣불리 시도는 해보기 힘들 것 같고 ( 기존 시스템 설계로 throughput 이 떨어지지도 않는다 ) 추후에 쓸 일이 있으면 poc 로 한번 진행해볼 생각이다.

카프카. 고생했습니다.

kimsunhak commented 1 week ago

소감

이번장은 카프카를 사용할때 무조건 참고해서 볼 만한 내용들이 많았다. 특히 그중 스트림처리에 대해서는 실제 우리가 내부 어플리케이션에서 데이터를 가공하는거와 많이 유사했고 비순차 이벤트, 재처리, 장애 관련해서도 카프카를 사용할때 유심히 봐야한다는걸 느꼈다!

결론 : 결국 카프카가 또 짱이라는 소리를 하는 느낌을 받음ㅋㅋㅋㅋ

KangHun-Lee commented 1 week ago

소감

카프카 스트림이 데이터를 단순히 소비하는 것을 넘어 데이터의 상태를 관리하고, 여러 스트림을 조합하거나 윈도우 연산을 통해 데이터를 집계할 수 있다는 점이 흥미로웠다. 언제 써볼 수 있을까 싶기는 한데, 언젠가는 경험할 수 있지 않을까 하면서 지금은 그냥 이런게 있구나 정도로 읽고 넘겨야겠다.

다들 고생하셨습니다.

hlab-books / kafka-the-definitive-guide

Chapter 14. 스트림 처리 #15

Chapter 14. 스트림 처리

14.1 스트림 처리란 무엇인가?

데이터 스트림

이벤트 스트림

프로그래밍 패러다임

14.2 스트림 처리 개념

14.2.1 토폴로지(topology)

14.2.2 시간

카프카 스트림즈가 결과물을 카프카 토픽에 쓸 때 '이벤트에 타임스탬프를 부여하는 규칙'

14.2.3 상태

14.2.4 스트림-테이블 이원성

14.2.5 시간 윈도우

14.2.6 처리 보장

14.3 스트림 처리 디자인 패턴

14.3.1 단일 이벤트 처리

14.3.2 로컬 상태와 스트림 처리 (그룹별 집계)

14.3.3 다단계 처리/리파티셔닝 (전체 집계)

14.3.4 외부 검색을 상요하는 처리: 스트림-테이블 조인

14.3.5 테이블-테이블 조인

14.3.6 스트리밍 조인

14.3.7 비순차 이벤트

14.3.8 재처리하기

14.4 예제로 보는 카프카 스트림즈

14.4.1 단어 개수 세기

단일 이벤트 처리 패턴(맵/필터 패턴)

14.4.2 주식 시장 통계

집계

14.4.3 클릭 스트림 확장

스트리밍 조인

14.5 카프카 스트림즈: 아키텍처 개요

14.5.1 토폴로지 생성하기

14.5.2 토폴로지 최적화하기

14.5.3 토폴로지 테스트하기

14.5.4 토폴로지 규모 확장하기

14.5.5 장애 처리하기

14.6 스트림 처리 활용 사례

14.7 스트림 처리 프레임워크 선택하기

상황 별 솔루션

보편적 고려사항

소감

소감

소감