14. 스트림 처리

Last updated - 2025년 08월 07일 Edit Source

• 📚 카프카 핵심가이드

카프카는 이벤트 스트림을 전달하는 것만 가능한 메세지 버스 정도로 인식되어 왔지만, 신뢰성 있는 데이터 스트림 저장소 역할을 넘어서서, 외부 처리 프레임워크에 의존할 필요 없이 애플리케이션에 이벤트를 읽고, 처리하고, 쓰게 할 수 있다.

무한히 늘어나는 데이터세트를 추상화한 것으로, 시간이 흐름에 따라 새로운 레코드가 계속해서 추가되므로 데이터 세트가 무한해짐

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

• e.g. 신용카드 결제, 주식 거래, 택배 배송, 네트워크 이벤트 등등
• 거의 모든 것을 이벤트의 연속으로 볼 수 있음

이벤트 스트림에는 순서가 있다

이벤트는 그 자체로 다른 이벤트 전에 혹은 후에 발생했다는 것을 의미

데이터 레코드는 불변하다

이벤트는 한 번 발생한 뒤 절대 고칠 수 없는데, 이벤트 스트림은 모든 트랜잭션을 포함하므로 작업 내역들을 기록

​

이벤트 스트림은 재생이 가능하다

• 대부분의 애플리케이션에서 재생이 불가능한 스트림을 생각하는 것은 어려운 일이 아니나 몇달 전 혹은 몇년 전에 발생한 raw stream을 그대로 재생할 수 있다는 것이 중요한데, 에러를 수정하거나 새로운 분석 방법을 시도하거나 혹은 감사를 수행하기 위해 필요

• 카프카는 이벤트 스트림을 캡처하고 또 재생할 수 있음

• 이벤트 스트림의 정의나 속성들은 이벤트 안에 저장된 데이터나 초당 생성되는 이벤트의 수와 같은 것과는 아무런 상관이 없는데 데이터는 시스템마다 다름

• 완전히 내부 구조가 없는 키-쌍이거나 약간의 구조를 가지고 있는 JSON이거나 구조화된 Avro나 Protobuf 메시지 일 수도 있는데 빅데이터라 불릴 정도로 초당 수백만 개의 이벤트가 쏟아질 경우도 있지만 처리 방식을 초당 심지어 분당 몇 개의 이벤트만 가지는 훨씬 작은 이벤트 스트림에도 동일하게 적용할 수 있음

• 하나 이상의 이벤트 스트림을 계속해서 처리하는 것을 의미

• 요청-응답이나 배치 처리와 마찬가지로 프로그래밍 패러다임 중 하나

​

요청-응답

• 응답 시간이 1밀리초 미만~몇 밀리초 수준인 패러다임으로, 가장 지연이 적은 패러다임으로 처리 방식이 보통 블로킹 방식이라 애플리케이션이 요청을 보낸 뒤 처리 시스템이 응답을 보내줄 때까지 대기하는 것이 보통으로 데이터베이스 세계에서 이 패러다임은 OLTP로 알려져 있음

• POS 시스템, 신용카드 결제 시스템, 시간 추적 시스템을 예로 들 수 있음

배치 처리

• 지연이 크지만 처리량도 큰데, 사전 설정된 시각에 시작됨

• 필요한 모든 입력 데이터를 읽고, 모든 출력 데이터를 쓰고 다음 번 실행시간까지 대기하는 식으로 되어 있음

• 처리 시간이 몇 분에 불과한 것에서부터 몇 시간에 이르는 것까지 다양하며 사용자들은 결과물을 볼 때 다소 시간이 지난 데이터라는 것을 감안하고 봄

• 데이터 웨어하우스나 비즈니스 인텔리전스 시스템이 속함

• 하루에 한 번 대량의 배치 단위로 적재되고, 리포트가 생성되고 사용자들은 다음 번 데이터 적재가 일어날 때까지 똑같은 리포트를 보게 됨

• 많은 경우 효율성이 높고 규모의 경제를 달성할 수 있다는 장점이 있으나 최근의 비즈니스 보다 시기적절하고 효율적인 의사 결정을 위해 더 짧은 시간 간격 안에 사용이 가능한 데이터를 필요로 하기 때문에 적은 지연보다 규모의 경제에 초점을 맞춰서 새발된 시스템 입장에서는 엄청난 부담이 됨

스트림 처리

• 연속적이고 논블로킹하게 작동하는 방식을 말함

• 이벤트 처리에 2밀리초 정도 기다리는 요청-응답과 하루 한 번 작어빙 실행되고 완료되는데 8시간이 걸리는 배치 처리 사이의 간격을 메워줌

• 대부분의 비즈니스 프로세스는 연속적으로 발생하고 비즈니스 리포트가 지속적으로 업데이트되고 최일선의 비즈니스 애플리케이션들이 역시 계속해서 응답할 수만 있다면 굳이 수 밀리초 내의 응답 같은 걸 기다릴 필요 없이 처리를 진행할 수 있음

• 의심스러운 신용카드 결제나 네트워크 사용내역을 알리는 기능이나 수요와 공급에 맞춰 실시간으로 가격을 조정하거나 물품 배송을 추적하는 것 등이 연속적이지만 논블로킹한 처리에 맞음

→ 스트림 처리에 대한 정의가 특정 프레임워크나 API, 기능을 요구하지 않으며 무한한 크기의 데이터세트에서 연속적으로 데이터를 읽어와서 뭔가를 하고 결과를 내보내는 한 스트림 처리를 수행하는 것으로, 지속적으로 계속되어야 함

​

스트림 처리는 다른 형태의 데이터 처리와 매우 비슷한데, 데이터를 쓰고 무엇인가 처리를 한 뒤 어딘가에 결과물을 쓰는 식의 코드를 작성하는 것을 말함

​

• 스트림 처리 애플리케이션은 하나 이상의 처리 토폴로지를 포함하는데, 하나의 처리 토폴로지는 하나 이상의 소스 스트림, 스트림 프로세서의 그래프, 하나 이상의 싱크 스트림이 서로 연결된 것으로서 하나 이상의 소스 스트림에서 시작된 이벤트 스트림은 연결된 스트림 프로세서들을 거쳐가면서 처리되다가 마지막에는 하나 이상의 싱크 스트림에 결과를 쓰는 것으로 끝남

• 각 스트림 프로세서는 이벤트를 변환하기 위해 이벤트 스트림에 가해지는 연산 단계라고 할 수 있음

스트림 처리에서 가장 중요한 개념으로, 대부분의 스트림 애플리케이션이 시간 윈도우에 대해 작업을 수행하는 만큼 시간에 대해 공통적인 개념을 가지는 것은 매우 중요함

1)스트림 처리 시스템의 시간 개념

1-1) 이벤트 시간

• 다루고자 하는 이벤트가 발생하여 레코드가 생성된 시점을 말함

• 0.10.0 이후부터 카프카는 프로듀서 레코드를 생성할 때 기본적으로 현재 시간을 추가하도록 되어 있는데, 애플리케이션의 이벤트 시간 개념과 일치하지 않는다면 레코드에 이벤트 시간을 가리키는 필드를 하나 추가하여 나중에 처리할 때 두 시간을 모두 호라용할 수 있는 방법을 권함

• 대부분의 경우 스트림 데이터를 처리할 때 이벤트 시간이 가장 중요함

1-2) 로그 추가 시간

• 이벤트가 카프카 브로커에 전달되어 저장된 시점을 말하는데, 접수 시간이라고도 불림

• 0.10.0 이후부터 카프카가 로그 추가 시간을 저장하도록 설정되어 있거나 타임스탬프가 포함되어 있지 않은 구버전 프로듀서에서 보낸 레코드일 경우 레코드에 로그 추가 시간을 자동으로 추가함

• 이벤트 시간이 기록되지 않은 경우, 로그 추가 시간은 일관성 있는 시간 기준으로서 사용할 수 있어 레코드가 생성된 다음부터는 변하지 않는데다가 이벤트 시간에 대한 합리적인 근사값으로 볼 수 있음

1-3) 처리 시간

• 스트림 처리 애플리케이션이 뭔가 연산을 수행하기 위해 이벤트를 받는 시간을 의미함

• 이벤트가 발생한 뒤 몇 밀리초, 몇 시간, 며칠 뒤일 수도 있어 동일한 이벤트라고 하더라도 정확히 언제 스트림 처리 애플리케이션이 이벤트를 읽었느냐에 따라 전혀 다른 타임스탬프가 주어질 수 있으며 애플리케이션 안에서도 스레드별로 다 다를 수 있어 매우 신뢰성이 떨어지며 가능한 피하는 것이 좋음

2) TimestampExtractor

카프카 스트림즈는 TimestampExtractor 인터페이스를 사용해서 각 이벤트에 시간을 부여하는데, 카프카 스트림즈를 사용하는 개발자는 서로 다른 구현체를 사용함으로써 이벤트 시간, 로그 추가 시간, 처리 시간 중 하나를 사용하거나 아니면 이벤트 내용에서 타임스탬프를 결정하는 등 완전히 다른 시간 개념을 사용할 수 있음

​

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

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

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

• 결과 레코드가 두 스트림을 조인한 결과물일 경우, 조인된 두 레코드 타임스탬프 중 큰 쪽의 타임스탬프를 결과 레코드의 타임스탬프로 사용하며 스트림과 테이블을 조인한 경우, 스트림 레코드 쪽의 타임스템프가 상요됨

• punctutate()와 같이 입력과 상관없이 특정한 스케줄에 따라 데이털르 생성하는 카프카 스트림즈 함수에 의해 생성된 결과 레코드의 경우, 타임스탬프 값은 스트림 처리 애플리케이션의 현재 내부 시각에 따라 결정됨

​

4) 시간대 주의

전체 데이터 파이프라인이 표준화된 시간 대 하나만 쓰거나 하지않으면 스트림 작업이 혼란스러운 결과를 내놓거나 의미가 없을 수도 있는데, 서로 다른 시간대의 데이터 스트림을 다루어야 한다면 윈도우에 작업을 수행하기 전에 이벤트 시각을 하나의 시간대로 변환해줄 필요가 있으며 아예 레코드에 시간대 정보를 저장해 넣는 경우도 많음

​

• 스트림 처리는 다수의 이벤트가 포함된 작업을 할 때, 각각의 이벤트 자체만을 살펴보는 것만으로는 충분하지 않는데, 한 시간 동안 발생한 타입별 이벤트 수나 조인, 합계, 평균을 계산해야 모든 이벤트 등 더 많은 정보를 추적 관리해야 하는 것으로 이런 정보를 상태라고 부름

• 스트림 이벤트의 개수를 저장하는 간단한 해시 테이블처럼 스트림 처리 애플리케이션의 로컬 변수에 상태를 저장하면 된다고 생각할 수 있으나 스트림 처리 애플리케이션이 정지하거나 크래시가 발생할 경우 상태가 유실되고 결과가 달라지므로 스트림 처리에서 상태를 관리하는 방법으로는 신뢰성이 떨어지므로 최신 상태를 보존하면서 애플리케이션을 재시작할 때 상태가 복구되도록 신경을 쓸 필요가 있음

1) 스트림 처리 유형의 상태

1-2) 로컬 혹은 내부 상태

• 스트림 처리 애플리케이션이 특정 인스턴스에만 사용할 수 있는 상태를 말함

• 대개 애플리케이션에 포함되어 구동되는 내장형 인메모리 데이터베이스를 사용해서 유지 관리됨

• 로컬 상태의 장점은 매우 빠르다는 점이고 단점은 사용 가능한 메모리 크기의 제한을 받음

• 스트림 처리의 많은 디자인 패턴들은 데이터를 분할해서 한정된 크기의 로컬 상태를 사용하여 처리 가능한 서브스트림으로 만드는데 초점을 둠

​

1-3) 외부 상태

• 외부 데이터 저장소에서 유지되는 상태는 카산드라와 같은 NoSQL 시스템을 사용해서 저장됨

• 외부 상태의 장점은 크기에 제한이 없고 여러 애플리케이션 인스턴스, 심지어 다른 애플리케이션에서도 접근이 가능하다는 점이고 단점은 다른 시스템을 추가하는데 따른 지연 증가, 복잡도 증가, 가용성 문제라고 할 수 있음

• 많은 스트림 처리 애플리케이션은 외부 저장소를 사용하는 걸 피하거나 내용물을 로컬 상태에 캐싱함으로써 외부 저장소와 가능한 한 통신하지 않게 함으로써 지연 부담을 최소화하는데, 내부 상태와 외부 상태를 일관적으로 유지하는 것이 과제로 남게 됨

• 스트림은 테이블과 달리 변경 내역을 저장함

• 스트림은 변경을 유발하는 이벤트의 연속이며 테이블은 여러 상태 변경의 결과물인 현재 상태를 저장함

1) 테이블을 스트림을 변환

테이블을 수정한 변경 내역을 잡아야 하는데, 모든 추가, 변경, 삭제 이벤트를 가져와서 스트림에 저장하면 되며 많은 데이터베이스에서는 이러한 변경점들을 잡아내기 위한 CDC 솔루션을 제공하며 이러한 변경점을 스트림 처리에서 활용할 수 있도록 카프카로 전달해줄 수 있는 카프카 커넥터가 많이 있음

​

2) 스트림을 테이블로 변환

스트림에 포함된 모든 변경 사항을 테이블에 적용해야 하는데, 이런 작업을 스트림을 구체화한다고 하며 메모리든 내부 저장소든 외부 데이터베이스든 테이블을 생성한 뒤 스트림에 포함된 이벤트를 처음부터 끝까지 모두 읽어서 상태를 변경하며 이 작업이 끝나면 특정 시점의 상태를 나타내는 테이블을 얻을 수 있음

​

• 대부분의 스트림 작업은 시간을 윈도우라 불리는 구간 단위로 잘라서 처리하는데, 이동 평균을 계산하거나 이번주 가장 많이 팔린 상품을 계산하거나 시스템의 99분위 부하를 찾아내는 식이 있음

• 두 스트림을 조인하는 작업도 윈도우 작업으로, 동일한 시간 간격 안에 발생한 이벤트들끼리 조인함

1) 계산 시 알아두어야 하는 사항

1-1) 윈도우 크기

• 카프카 스트림은 윈도우의 크기가 비활동 기간의 길이에 따라 결정되는 세션 윈도우도 지원함

• 개발자가 세션 간격을 정의하면 세션 간격보다 작은 시간 간격을 두고 연속적으로 도착한 이벤트들은 하나의 세션에 속하게 됨

• 세션 갭 이상으로 이벤트가 도착하지 않으면 새로운 세션이 생성되어 이후 도착하는 이벤트들을 담게 됨

1-2) 시간 윈도우의 진행 간격

• 호핑 윈도우 (hopping window) : 윈도우의 크기 = 윈도우 사이의 고정된 시간 간격인 윈도우
• 텀블링 윈도우 (tumbling window) : 윈도우의 크기 = 진행 간격인 윈도우

​

1-3) 윈도우를 업데이트할 수 있는 시간

이벤트가 이벤트에 해당하는 윈도우에 추가될 수 있는 기한을 정의할 수 잇는 것이 이상적으로 만약 이벤트가 최대 4시간까지 지연될 수 있다면 결과를 다시 계산하고 업데이트 해줘야 할것이며 만약 이벤트가 그 이상으로 지연된다면 무시하면 됨

​

2) 윈도우 계산 방식

벽시계 시간에 맞춰 정렬될 수 있는데, 그게 아니라면 정렬을 하지 않고 단순히 애플리케이션이 실행된 시점을 기준으로 삼을 수 있음

​

• 스트림 처리 애플리케이션에 있어 핵심적인 요구 조건 중 하나는 장애가 발생했을 경우에도 각 레코드를 한번만 처리할 수 있는 능력으로, 정확히 한 번 보장이 없는 스트림 처리는 정확한 결과가 요구되는 상황에서 사용될 수 없음

• 아파치 카프카는 트랜잭션적이고 멱등적 프로듀서 기능을 통해 정확히 한 번 의미 구조를 지원하는데, 카프카 스트림즈는 카프카의 트랜잭션 기능을 사용해서 스트림 처리 애플리케이션에 정확히 한 번 보장을 지원함

• 카프카 스트림즈 라이브러리를 사용하는 모든 애플리케이션은 processing.guarantee 설정을 exactly_once로 잡아줌으로써 정확히 한 번 보장 기능을 활성화시킬 수 있음

• 2.6 버전 이후의 카프카 스트림즈는 2.5버전 이후 브로커를 필요로하는 좀 더 효율적인 정확히 한번 구현체를 호함하는데 이는 processing.guarantee 설정값을 exactly_once_beta로 잡아줌으로써 활성화시킬 수 있음

• 근데, 3.0부터 지원 중단, 4.0에서 제거 예정이니 exactly_once_v2 사용 권장

1) 맵/필터 패턴

• 가장 단순한 스트림 처리 패턴은 각 이벤트를 개별적으로 처리하는 것을 말함

• 불필요한 이벤트를 스트림에서 걸러 내거나 각 이벤트를 변환하기 위해 사용되는 경우가 많음

• 스트림 처리 애플리케이션은 스트림의 이벤트를 익어와서 각 이벤트를 수정한 뒤 수정된 이벤트를 다른 스트림에 씀

• 각 이벤트가 독립적으로 처리될 수 있어 애플리케이션은 애플리케이션 안에 상태를 유지할 필요가 없으므로 상태로 복구할 필요도 없어 장애 복구나 부하 분산이 매우 쉬우며 다른 애플리케이션 인스턴스가 이벤트를 넘겨받아 처리하게 하면 됨

• 간단한 프로듀서와 컨슈머를 사용하여 쉽게 처리가 가능함

• 대부분의 스트림 처리 애플리케이션은 윈도우 집계와 같이 정보의 집계에 초점을 맞추는데, 집계 시 스트림의 상태를 유지할 필요가 있음

• 그룹별 집계의 경우, 공유 상태가 아닌 로컬 상태를 사용해서 수행할 필요가 있음

• 카프카 파티셔너를 사용하여 동일한 항목에 대한 모든 이벤트가 동일한 파티션에 쓰여지도록 할 수 있고 그 후 각 애플리케이션 인스턴스는 자신에게 할당된 파티션에 저장된 모든 이벤트 즉, 카프카 컨슈머 단위에서 보장되는 것으로 애플리케이션의 각 인스턴스는 자신에게 할당된 파티션에 쓰여진 전체 항목의 부분 집합에 대한 상태를 유지할 수 있음

• 스트림 처리 애플리케이션은 로컬 상태를 보유하게 되는 순간 복잡해짐

1) 스트림 처리 애플리케이션이 고려해야 하는 사항

1-1) 메모리 사용

로컬 상태는 애플리케이션 인스턴스가 사용 가능한 메모리 안에 들어갈 수 있는게 이상적인데 어떤 로컬 저장소는 디스크에 내용물을 저장하는 기능을 지원하나 성능에 상당한 영향을 미침

1-2) 영속성

• 애플리케이션 인스턴스가 종료되었을 때 상태가 유실되지 않아야 하고 인스턴스가 재실행되거나 다른 인스턴스에 의해 복구될 수 있음을 확신해야 함

• 카프카 스트림즈는 내장된 RocksDB를 사용하여 로컬 상태를 인메모리 방식으로 저장함과 동시에 재시작 시 빠르게 복구가 가능하도록 디스크에 데이터를 영속적으로 저장하지만 로컬 상태에 대한 모든 변경 사항은 카프카 토픽에도 보내짐

• 스트림 처리를 담당하고 있는 노드에 장애가 발생하더라도 로컬 상태는 유실 되지 않는데, 이는 카프카 토픽으로부터 이벤트를 읽어옴으로써 쉽게 복구될 수 있음

• 카프카는 토픽들이 끝없이 자라나는 것을 방지하고 상태 복구를 언제고 실행 가능하게 하기 위해 로그 압착을 사용함

1-3) 리밸런싱

파티션은 이따금 서로 다른 컨슈머에게 다시 할당될 수 있는데, 재할당이 발생하면 파티션을 상실한 애플리케이션 인스턴스는 마지막 상태를 저장함으로써 해당 파티션을 할당받은 인스턴스가 재할당 이전 상태를 복구시킬 수 있도록 해야 함

​

2) 지원되는 로컬 사앹 관리 기능의 수준은 스트림 처리 프레임워크별로 다름

애플리케이션이 로컬 상태를 유지해야 한다면 사용 중인 프레임워크가 이를 보장하는 지의 여부를 확인할 필요가 있음

​

• 그룹별 집계가 필요할 때 로컬 상태를 사용하면 좋은데 모든 정보를 사용하여 내야 하는 결과가 있는 경우에 여러 단계로 접근해야 함

• 각 애플리케이션 인스턴스에 로컬 상태만을 가지고 할 수 있고, 그 후 하나의 파티션만 가진 새로운 토픽에 결괄르 쓴 뒤 이 파티션을 하나의 애플리케이션 인스턴스에 읽어서 찾거나 특정 조건만을 포함하는 토픽의 경우 전체 정보를 포함하는 토픽에 비해 크기도 트래픽도 훨씬 작으므로 단일 인스턴스만 가지는 애플리케이션만으로도 충분히 처리할 수 있음

• 맵 리듀스는 리듀스 단계를 여러 번 거치는 경우가 흔한데 리듀스 단계별로 애플리케이션이 하나씩 필요함

• 대부분의 스트림 처리 프레임워크는 어느 애플리케이션 인스턴스 혹은 워커가 어느 단계를 수행할지 등의 세부적인 사항들을 프레임워크가 알아서 해주도록 함으로써 모든 단계를 하나의 애플리케이션에 담을 수 있도록 함

• 외부 검색을 사용하는 처리에서 사용

• 데이터 확장을 위해 외부 검색을 수행할 때에는 각 레코드를 처리하는 데 있어서 5~15밀리초 사이의 상당한 지연을 발생시키는데, 외부 데이터 저장소에 걸리는 추가 부하도 용인하기 어려움

• 스트림 처리 시스템은 보통 초당 10만~50만개의 이벤트를 처리할 수 있는데 반해 데이터베이스는 초당 1만개 가량의 이벤트를 처리할 수 있는게 보통이므로 가용성을 보장하기 위해 수반되는 복잡성도 있으므로 애플리케이션 외부 DB가 사용 가능하지 않은 상황을 처리할 수 있어야 함

• 성능과 가용성을 모두 유지하기 위해서는 스트림 처리 애프리케이션 안에 데이터베이스에 저장된 데이터를 캐시할 필요가 있지만 캐시를 관리하는 것도 어려울 수 있는데 데이터베이스 테이블에 가해지는 모든 변경점을 이벤트 스트림에 담으면 스트림 처리 작업이 이 스트림을 받아와서 캐시를 업데이트하는데 사용될 수 있음 ​
  

1) CDC

• 데이터베이스의 변경 내역을 이벤트 스트림으로 받아오는 것을 말함

• 카프카 커넥트는 CDC를 수행하여 데이터베이스 테이블을 변경 이벤트 스트림으로 변환할 수 있는 커넥터가 여러개 있어 이를 사용하면 테이블의 복사본을 따로 유지할 수 있는 동시에 데이터베이스 변경 이벤트가 발생할 때마다 알림을 받아서 테이블 복사본을 적절히 업데이트할 수 있음

• 확장이 용이하면 데이터베이스와 이를 사용하는 다른 애플리케이션에 영향을 주지 않음

​

2) 스트림 테이블 조인

스트림 중 하나가 로컬에 캐시된 테이블에 대한 변경사항을 나타내는 것을 말함

• 두 개의 테이블을 조인하는 것은 윈도우 처리되지 않은 연산으로, 작업이 실행되는 시점에서의 양 테이블의 현재 상태를 조인함

• 카프카 스트림에서는 동일한 방식으로 파티션된 동일한 키를 가지는 두 개의 테이블에 대해 동등 조인을 수행할 수 있고 이렇게 함으로써 조인 연산이 많은 수의 애플리케이션 인스턴스와 장비에 효율적으로 분산될 수 있게 함

• 카프카 스트림즈도 두 개의 테이블에 대해 외래키 조인을 지원하는데, 스트림 혹은 테이블의 키와 다른 스트림 혹은 테이블의 임의의 필드를 조인할 수 있음

• 스트림은 무한이라는 특징이 있는데 테이블에서는 현재 상태만 관심사이므로 스트림을 사용하여 테이블을 나타낼 때 스트림에 포함된 과거 이벤트는 무시될 수 있지만 두 개의 스트림을 조인할 경우 한쪽 스트림에 포함된 이벤트를 같은 키 값과 함께 같은 시간 윈도우에 발생한 다른쪽 스트림 이벤트와 맞춰야 하므로 과거와 현재의 이벤트를 전체 조인하게 됨

• 카프카 스트림즈는 조인할 두 스트림이 똑같이 조인 키에 대해 파티셔닝되어 이을 경우 동등 조인을 지원하는데, 이 경우 카프카 스트림즈는 두 토픽의 동일 파티션에 대한 작업을 같은 태스크에 할당함으로써 모든 연관된 이벤트를 볼 수 있음

• 카프카 스트림즈는 두 토픽에 대한 조인 우니도우를 내장된 RocksDB 상태 저장소에 유지함으로써 조인을 수행함

비순차 이벤트는 상당히 자주 발생하는데, 만약 몇 시간 동안 끊긴 상태에서 재접속하는 경우 몇 시간 치의 이벤트를 한꺼번에 전송하므로 이 현상은 네트워크 장비를 모니터링하는 상황이나 제조업 현장에서도 발생할 수 있음

1) 비순차 이벤트 상황 처리

• 이벤트가 순서를 벗어났음을 알아야 하는데, 이를 위해 애플리케이션이 이벤트 시간을 확인해서 현재 시각보다 더 이전인지를 확인할 수 있어야 함

• 비순차 이벤트의 순서를 복구할 수 있는 시간 영역을 정의해야 함(N시간 정도 복구, N주 이상 오래된 것 복구 포기)

• 순서를 복구하기 위해 이벤트를 묶을 수 있어야 하는데 스트리밍 애플리케이션과 배치 작업의 주요한 차이점으로 만약 매일 돌아가는 배치 작업이 끝난 후 몇 개의 이벤트가 추가로 도착했다면 보통 어제 작업을 다시 돌려서 이벤트를 변경해주지만 스트림 처리에는 이런 개념이 없으므로 계속 돌아가는 동일한 프로세스가 주어진 시점 기준으로 오래된 이벤트와 새로운 이벤트를 모두 처리해야 함

• 결과를 변경할 수 있어야 하는데, 스트림 처리의 결과가 데이터베이스에 쓰여질 경우 결과를 변경하는데 put 혹은 update 정도면 충분하지만 스트림 애플리케이션이 결과를 이메일로 전송할 경우 변경이 곤란할 수 있음

2) 스트림 처리 프레임워크의 기능 지원

• 구글의 데이터플로나 카프카 스트림과 같은 스트림 처리 프레임워크는 처리 시간과 독립적인 이벤트 시간의 개념을 자체적으로 지원하고 현재 처리 시간 이전 혹은 이후의 이벤트 시간을 가지는 이벤트를 다룰 수 있는 기능도 가지고 있음

• 보통 로컬 상태에 다수의 집계 윈도우를 변경 가능한 상태로 유지해주고, 개발자가 이 윈도우를 얼마나 오랫동안 유지할 수 있게 해주는 식으로 구현하면 되며 집계 윈도우를 더 오랫동안 변경 가능한 형태로 유지할수록 로컬 상태를 유지하기 위해 메모리 역시 더 많이 필요함

3) 카프카 스트림즈 API

언제나 집계 결과를 결과 토픽에 쓰는데 이 토픽들은 대체로 로그 압착이 설정되어 있는 토픽으로 각 키값에 대해 마지막 밸류만 유지되므로 집계 윈도우의 결과가 늦게 도착한 이벤트로 인해 변경되어야 한다면 카프카 스트림즈는 단순히 해당 집계 윈도우의 새로운 결과값을 씀으로써 기존 결과값을 대체함

이벤트 재처리 패턴

1) 새로 개선된 버전의 스트림 처리 애플리케이션

• 구버전에서 사용하던 이벤트 스트림을 신버전 애플리케이션에서 읽어와서 산출된 새로운 결과 스트림을 씀

• 기존 구버전의 결과를 교체하는 것이 아닌 한동안 두 버전의 결과를 비교한 뒤 어느 시점에 구버전 대신 신버전의 결과를 사용하도록 함

1-1) 하나의 스트림 처리 애플리케이션의 두 버전이 동시에 두 개의 결과 스트림을 쓰기 위해 지켜야 하는 사항

• 신버전 애플리케이션을 새 컨슈머 그룹으로 실행시킴

• 신버전 애플리케이션이 입력 토픽의 첫번째 오프셋부터 처리를 시작하도록 설정해서 입력 스트림의 모든 이벤트에 대한 복사본을 가질 수 있도록 함

• 신버전 애프리케이션이 처리를 계속하도록 하고, 신버전 처리 작업이 따라잡았을 때 클라이언트 애플리케이션을 새로운 결과 스트림으로 전환함

2) 이벤트 스트림 재처리 결과 재산출

• 기존의 스트림 처리 애플리케이션에 버그가 많아 버그 수정 뒤 이벤트 스트림을 재처리해서 결과를 재산출함

• 이미 존재하는 애플리케이션을 초기화해서 두 애플리케잇녀 버전에서 나온 결과물이 뒤섞이지 않도록 입력 스트림의 맨 처음부터 다시 처리하도록 되돌린 후, 로컬 상태를 초기화하여 기존 출력 스트림 내용물도 지워야 함

→ 같은 애플리케이션을 두 개 돌려서 결과 스트림도 두개가 나올 정도로 용량이 충분하다면 첫번째 방식을 택하는게 안전한데 2개 이상의 버전을 왔다갔다 할 수도 있고 버전 간의 결과물을 비교할 수 있고 정리 과정에서 중요 데이터가 유실되거나 에러가 발생할 위험도 없음

​

• 스트림 처리 애플리케이션은 상태를 보유하고 이 상태는 애플리케이션의 여러 인스턴스 아이에 분산될 수 있음

• 스트림 처리 애플리케이션의 사용자는 결과 토픽을 읽어들임으로써 처리 결과를 받아볼 수 있지만 상태 저장소 그 자체에서 바로 결과를 읽어올 필요가 있는 경우에는 처리 결과가 테이블 형태인 경우 흔한데 이 경우 결과 스트림은 곧 테이블에 대한 업데이트 스트림이므로 스트림 처리 애플리케이션의 상태에서 테이블을 바로 읽어오는 것이 훨씬 더 빠르고 쉬움

• 카프카 스트림즈는 스트림 처리 애플리케이션의 상태를 쿼리하기 위한 유연한 API를 포함함

아파치 카프카는 저수준의 Processor API, 고수준의 스트림 DSL의 2개의 스트림즈를 제공

1) 저수준 API

• 변환을 직접 생성할 수 있게 해줌

​

2) 고수준 스트림 DSL

• 스트림에 포함된 이벤트에 일련의 연속적인 변환을 정의함으로써 스트림 처리 애플리케이션을 정의할 수 있음

• 변환은 필터와 같이 단순한 것일수도 있고 스트림-스트림 조인처럼 복잡한 것일 수도 있음

• DSL API를 사용하는 애플리케이션은 항상 StreamBuilder를 사용하여 처리 토폴로지를 생성함으로써 시작함

• 처리 토폴로지를 생성한 뒤 여기에서부터 KafkaStreams 실행 객체를 생성하는데 객체를 시작시키면 스트림 안의 이벤트에 처리 토폴로지를 적용하는 다수의 스레드가 시작되고 KafkaStreams 객체를 닫으면 처리가 끝남

처리 토폴로지

스트림 안의 이벤트에 적용되는 변환을 정점으로 하는 유향 비순환 그래프를 말함(DAG)

• 방향성이 존재하는 비순환 그래프를 의미

​

• 스트림 처리 애플리케이션을 개발하기 위해 가장 먼저 해야할 것은 카프카 스트림즈를 설정하는 것으로 설정 가능한 항목이 매우 많음

• Properties 객체에 임의의 프로듀서나 컨슈머 옵션을 추가함으로써 카프카 스트림즈에 내장될 프로듀서와 컨슈머를 설정하는 것도 가능함

1) 카프카 스트림즈 설정

1
2
3
4
5

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount"); // (1)
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // (2)
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); // (3)
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

1-1) 애플리케이션 ID

• 모든 카프카 스트림즈 애플리케이션은 애플리케이션 ID를 가짐

• 서로 다른 애플리케이션 인스턴스들이 서로 협락하게 하는 데에도 사용되지만 내부에서 사용하는 로컬 저장소와 여기 연관된 토픽에 이름을 정할 때도 사용됨

• 애플리케이션 ID는 같은 카프카 클러스터를 사용하는 각각의 카프카 스트림즈 애플리케이션별로 서로 달라야 함

1-2) 카프카 찾을 방법 지정

• 카프카 스트림즈 애플리케이션은 항상 카프카 토픽에서 데이터를 읽어서 출력된 결과물을 카프카 토픽에 쓰는데 이는 카프카 스트림즈 애플리케이션은 인스턴스끼리 서로 협력하도록 하는데도 카프카를 사용하므로 애플리케이션이 카프카를 찾을 방법을 지정해주어야 함

1-3) 직렬화/역직렬화

• 데이터를 읽고 쓸 때 애플리케이션은 직렬화/역직렬화를 해야 하므로, 기본값으로 쓰일 Serde 클래스를 지정해주어야 함

• 필요 시 스트림즈 토폴로지를 생성할 때 기본값을 재정의할 수 있음

2) 스트림 토폴로지 생성

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

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");

• StreamBiulder 객체를 생성하고 입력으로 사용할 토픽을 지정하여 스트림 정의를 시작함

• 입력 토픽에서 읽어오는 각 이벤트는 단어로 읽어오는 문자열 한줄이므로, 정규식을 사용하여 문자열을 다수의 단어들로 분할하고 각 단어를 가져다가 이벤트 레코드 키로 넣어줌으로써 그룹화에 사용할 수 있음

3) 카프카 스트림즈 실행

1
2
3

streams.start(); 
Thread.sleep(5000L); 
streams.close();

• start() 메소드로 실행시킴

• close() 메소드는 카프카 스트림즈를 멈춤

4) 프로덕션 클러스터에 배포

• YARN이나 Mesos를 설치하고 처리 프레임워크를 모든 장비에 설치하고 애플리케이션을 클러스터에 제출해야 함

• 카프카 스트림즈 API를 사용하면 단순히 애플리케이션 인스턴스를 여러 개 띄우는 것만으로도 처리 클러스터를 하나 구성할 수 있음

1) Gson 라이브러리를 활용한 시리얼라이저/디시리얼라이저 처리

• 키와 밸류 둘다 문자열이면 시리얼라이저/디시리얼라이저 둘 다 Serdes.String() 클래스를 사용하면 됨

• 키가 문자열, 밸류는 문자열이 아닌 다른 객체인 경우에는 직렬화와 역직렬화를 위해 Gson 라이브러리를 사용하여 시리얼라이저/디시리얼라이저 처리를 하면 됨

• 카프카에 저장하고자 하는 모든 객체에 대해 Serde 객체를 지정해야 하는데 이는 Gson이나 Avro, Protobuf와 같은 라이브러리를 사용해서 Serde를 생성할 것을 권장함

1
2
3
4
5
6

static public final class TradeSerde extends WrapperSerde<Trade> { 
	public TraeSerde() { 
		super(new JsonSerializer<Trade>(), 
				new JsonDeserializer<Trade>(Trade.class)); 
	} 
}

​

2) aggregate

• Meterialized는 마지막 파라미터로, 이는 저장소를 설정하는데 사용하는 객체로서 고유한 이름도 저장소 이름으로 사용할 수 있음

• withValueSerde는 상태 저장소 설정의 일부로 집계 결과를 직렬화/역직렬화하기 위한 Serde 객체를 지정해주어야 함

​

3) WindowedSerde

• 결과물을 윈도우 타임스탬프와 함께 윈도우가 적용된 데이터 형식으로 저장함

• 윈도우 크기가 직렬화 과정에서 사용되는 것은 아니나 Serde의 일부로 전달함

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

KStream<Windowed<String>, TradeStats> stats = source
	.groupByKey()
	.windowedBy(TimeWindows.of(Duration.ofMillis(windowSize))
							.advanceBy(Duration.ofSeconds(1)))
	.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)));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

KStream<Integer, PageView> views = 
	builder.stream(Constants.PAGE_VIEW_TOPIC, 
		Consumed.with(Serdes.Integer(), new PageViewSerde())); 

KStream<Integer, Search> searches = 
	builder.stream(Constants.SEARCH_TOPIC, 
		Consumed.with( Serdes.Integer(), new SearchSerde())); 

KTable<Integer, UserProfile> profiles =  
	builder.table(Constants.USER_PROFILE_TOPIC, 
		Consumed.with(Serdes.Integer(), new ProfileSerde())); 
		
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)).before(Duration.ofSeconds(0)), 
		StreamJoined.with(
			Serdes.Integer(), new UserActivitySerde(), new SearchSerde()));

1) 다수의 스트림을 조인하는 토폴로지

1-1) 스트림 객체 생성

조인하고자 하는 두 개의 스트림 객체를 생성하는데, 이때 입력 토픽 뿐만 아니라 토픽 데이터를 읽어서 객체로 역직렬화할 때 사용될 키, 밸류에 대한 Serde도 지정해주어야 함

1-2) KTable 정의

변경 스트림에 의해 갱신되는 구체화된 저장소를 말함

1-3) 스트림-테이블 조인

• 스트림-테이블에서 스트림의 각 이벤트는 테입르의 캐시된 사본에서 정보를 받고 설정한 조인을 수행하는데 없는 정보도 보존됨

• 스트림과 레코드에서 하나씩 값을 받아 또 다른 값을 리턴하는데 데이터베이스와 달리 두 개의 값을 결합해서 어떻게 하나의 결과로 만들지를 결정해야 함

• 이후 같은 key에 의해 수행도니 정보들을 조인하는데 두 개의 스트림을 조인하는 것이지 스트림을 테이블에 조인하는 것이 아님

1-4) 스트림-스트림 조인

• 시간 윈도우를 사용한 조인으로 각 key별로 모든 정보를 조인하는 것은 적절하지 않은데 짧은 시간 안에 발생한 정보를 조인함으로써 연관된 정보들만 조인해야 하므로 조인 윈도우를 정의해야 함

• of를 호출하여 전과 후의 짧은 시간의 윈도우를 생성한 뒤 특정 초 간격으로 before를 호출하여 짧은 시간 동안의 정보만을 조인하도록 하여 모든 정보에 대한 분석을 수행할 수 있음

1-5) 조인 결과

조인 결과에 대한 Serde에 정의하며, 조인 양쪽에 공통인 키 값에 대한 Serde와 조인 결과에 포함될 양쪽의 밸류값에 대한 Serde를 포함함

• 모든 스트림즈 애플리케이션은 하나의 토폴로지를 구현하고 실행하는데, 토폴로지는 다른 스트림 처리 프레임워크에서는 DAG 혹은 유향 비순호나 그래프라고 불리는데 모든 이벤트가 입력에서 출력으로 이동하는 동안 수행되는 작업과 변환 처리의 집합이라고 할 수 있음

• 토포롤지는 프로세서들로 구성되는데 대부분의 프로세서는 필터, 맵, 집계 여산과 같은 데이터에 대한 처리 작업을 구현함

• 토픽으로부터 데이터를 읽어와서 넘겨주는 프로세서도 있고, 앞 프로세서로부터 데이터를 넘겨받아서 토픽에 쓰는 싱크 프로세서도 있음

• 토폴로지는 항상 하나 이상의 소스 프로세서로 시작해서 한 개 이상의 싱크 프로세서로 끝남

카프카 스트림즈는 DSL API를 사용해서 개발된 애플리케이션의 각 DSL 메소드를 독립적으로 저수준 API로 변환하여 실행하는데 각 DSL 메소드를 독립적으로 변환하므로 결과 토폴로지는 전체적으로 그리 최적화되지 않은 상태일 수 있음

1) 카프카 스트림즈 애플리케이션의 실행 단계

• KStream, KTable 객체를 생성하고 필터, 조인과 같은 DSL 작업을 수행함으로써 논리적 토폴로지는 정의함

• StreamBuilder.build() 메소드가 논리적 토폴로지로부터 물리적 토폴로지를 생성해냄

• KafkaStreams.start()가 토폴로지를 실생시키고 데이터를 읽고, 처리하고, 쓰는 것이 해당됨

2) 최적화 적용 방식

• StreamsConfig.TOPOLOGY_OPTIMIZATION 설정값을 StreamsConfig.OPTIMIZE로 잡아준 뒤 build(props)를 호출함으로써 활성화시킬 수 있음

• 설정값 없이 build()를 호출할 경우 최적화는 적용되지 않음

• 애플리케이션 테스트 시 최적화를 적용한 것과 적용되지 않은 것을 비교하여 실행 시간과 카프카에 쓰여지는 데이터 양을 비교하는 것이 좋음

1) TopologyTestDriver

• 카프카 스트림즈 애프리케이션의 주된 테스트 툴로, 1.1.0에서 처음 도입됨

• 일반적인 단위 테스트와 비슷하게 작동하는데 입력 데이터를 정의하고, 목업 입력 토픽에 데이터를 쓰고 테스트 드라이버를 써서 토폴로지를 실행시키고 목업 출력 토픽에서 결과를 읽은 뒤 예상하는 결과와 비교 검증함

• TopologyTestDriver 클래스는 카프카 스트림즈의 캐시 기능을 시뮬레이션해주지 않으므로 찾을 수 없는 에러도 많음

• 단위 테스트는 통합 테스트로 보강되는 것이 보통인데 카프카 스트림즈의 경우 EmbeddedKafkaCluster와 Testcontainsers의 두 통합 테스트 프레임워크가 자주 쓰이는데 EmbeddedKakfaCluster는 테스트를 수행하는 JVM 상에 카프카 브로커를 하나 띄워주는 방식이고 Testcontainers는 도커 컨테이너를 사용하여 카프카 브로커와 기타 테스트에 필요한 다른 요소들을 띄워주는 방식을 말함

• 도커를 사용해서 카프카와 그 의존성, 사용되는 리소스를 테스트 애플리케이션으로부터 완전히 격리시키기 때문에 Testcontianers가 더 권장됨

1) 카프카 스트림즈의 부하 분산

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

• 하나의 장비에서 다수의 스레드를 사용하여 카프카 스트림즈 애플리케이션을 실행시킬 수 있고 여러 대의 장비에서 실행시킬 수도 있는데 어느 경우건 애플리케이션의 모든 활성화된 스레드들은 데이터 처리에 수반되는 작업을 균등하게 수행함

2) 카프카 스트림즈 엔진의 병렬 처리

• 카프카 스트림즈 엔진은 토폴로지의 실행을 다수의 태스크로 분할함으로써 병렬 처리함

• 스트림즈 엔진은 애플리케이션이 처리하는 토픽의 파티션 수에 따라 태스크 수를 결정하고 각 태스크는 전체 파티션 중 일부의 처리를 책임짐

• 각 태스크는 자신이 담당하는 파티션들을 구독해서 이벤트를 일겅오는데 이벤트를 읽어올 때마다 태스크는 이 파티션에 적용될 모든 처리 단계를 실행시킨 후 결과를 싱크에 쓰고 이런 태스크들은 서로 완전히 독립적으로 실행될 수 있으므로 카프카 스트림즈에서 병렬 처리의 기본 단위가 됨

​

3) 스레드 수 결정

• 애플리케이션 개발자는 애플리케이션 인스턴스가 실행시킬 스레드의 수를 결정할 수 있음

• 다수의 스레드를 활용할 수 있다면 각 스레드는 해당 애플리케이션이 생성하는 전체 태스크의 일부를 실행하게 됨

• 다수의 애플리케이션의 인스턴스가 다수의 서버에서 실행될 경우, 각 서버의 스레드별로 서로 다른 태스크가 실행되며 처리하는 토피의 파티션 수만큼의 태스크를 생성하는 것이 스트리밍 애플리케이션의 규모를 확장하는 방식이 됨

• 더 빨리 처리를 하고 싶다면 스레드 수를 늘리면 되고, 서버의 자원이 고갈되었다면 다른 서버에 추가 인스턴스를 띄우면 되므로 카프카가 자동으로 작업을 코디네이션하는데 즉, 카프카가 각각의 태스크에 파티션을 나눠서 할당해주면 각 태스크는 자신이 할당받은 파티션에서 독립적으로 이벤트를 받아와서 처리하고 토폴로지에 정의된 집계 연산에 관련된 로컬 상태를 유지하게 됨

​

4) 다수의 파티션에서 입력을 가져와서 처리

• 태스크 사이에 의존 관계가 생길 수도 있음

• 카프카 스트림즈는 각 조인 작업에 필요한 모든 파티션들을 하나의 태스크에 할당함으로써 해당 태스크가 필요한 파티션 전부로부터 데이터를 읽어온 뒤 조인을 수행할 수 있도록 하는데 이는 현재 카프카 스트림즈가 조인 작업에 사용될 모든 토픽에 대해 동일한 조인 키로 파티션된 동일한 수의 파티션을 가질 것을 요구함

5) 리파티션닝

• 특정 태스크가 특정 파티션의 데이터를 처리한 후 데이터 리파티셔닝을 수행하는 프로세스가 뒤따르는 경우, 셔플을 하거나 아니면 다른 태스크로 이벤트를 보내야 하는데 이는 다른 스트림 처리 프레임워크와는 달리 카프카 스트림즈는 리파티션이 호출되면 새로운 키와 파티션을 가지고 새로운 토픽에 이벤트를 쓰고 그 다음에 오는 태스크들이 새 토픽에서 이벤트를 읽어와서 처리를 계속함

• 리파티셔닝은 전체 토폴로지를 2개의 서브 토폴로지로 분할하는데, 두번쨰 서브 토폴로지는 첫번째의 결과물을 받아서 처리하는 만큼 첫번째에 의존하게 되지만 첫번쨰 태스크 집합은 자기 속도대로 데이터를 토픽에 쓰고 두번째 태스크 집합 역시 자기 속도다로 토픽에서 데이터를 읽어와서 처리하기 떄문에 두 태스크 집합은 서로 독립적이고 ㅂ여렬로 실행됨

• 태스크 사이에 통신이나 공유된 리소스가 없으므로 동일한 스레드나 서버에서 실행될 필요도 없음(카프카의 장점)

1) 마지막 커밋 오프셋으로 처리

카프카는 매우 가용성이 높은 시스템이고 카프카에 저장하는 데이터 역시 마찬가지이므로 애플리케이션에 장애가 발생하여 재시작이 필요할 경우, 장애가 발생하기 전 마지막으로 커밋된 오프셋을 카프카에서 가져옴으로써 처리하던 스트림의 마지막으로 처리된 지점부터 처리를 재개할 수 있음

​

2) 체인지 로그

로컬 상태 저장소가 유실되었을 경우 스트림즈 애플리케이션은 항상 카프카로부터 체인지 로그를 카프카에서 읽어옴으로써 로컬 상태 저장소를 복구함

​

3) 카프카 컨슈머의 코디네이션 기능

• 카프카 스트림즈는 태스크 고가용성을 지원하기 위해 카프카 컨슈머의 코디네이션 기능을 사용함

• 태스크에 장애가 발생했으나 다른 스레드 혹은 인스턴스가 작동 중일 경우, 해당 태스크는 사용 가능한 다른 스레드에서 재시작하게 됨

• 컨슈머 그룹에 속한 컨슈머 중 하나에 장애가 발생할 경우, 장애가 발생한 컨슈머에 할당되어 있던 파티션을 남은 컨슈머 중 하나에 할당해줌으로써 장애에 대응하는 것과 유사함

• 카프카의 정확히 한 번 의미 구조 뿐만 아니라 정적 그룹 멤버십이나 협력적 리밸런스와 같은 카프카 컨슈머 그룹 코디네이션 프로토콜이 개선되면서 카프카 스트림즈도 혜택을 보게 됨

​

4) 복구 속도

• 장애가 발생한 스레드에서 실행되고 있던 태스크를 다른 스레드가 넘겨받아 처리를 시작해야 할 때 가장 먼저 해야할 일은 저장된 상태를 복구시키는 것으로, 카프카에 저장된 내부 토픽을 다시 읽어와서 카프카 스트림즈의 상태 저장소를 업데이트 하는 식으로 복구할 수 있으나 작업을 수행하는 동안은 일부 데이터에 대해서 스트림즈 처리 작업이 진행되지 않을 것이며 그만큼 가용성은 줄어들고 출력 데이터는 뒤떨어지게 됨

• 복구 시간을 줄이는 문제는 곧 상태를 복구시키는데 걸리는 시간을 줄이는 문제가 되는데, 가장 핵심적인 방법은 compaction.lag.ms는 낮추고 세그먼트 크기는 기본값 1GB 대신 100MB 정도로 낮춤으로써 가능함

5) 스탠바이 레플리카 설정

• 스탠바이 레플리카는 스트림 처리 애플리케이션에 현재 작동 중인 태스크를 따라가기만 하는 태스크로, 다른 서버에서 현재의 상태를 유지하는 역할을 함

• 장애가 발생하면 이 태스크는 이미 거의 최신 현재 상태를 보유하고 있으므로, 중단 시간이 거의 없이 바로 처리를 재개할 수 있음

1) 고객 서비스

스트림 처리 애플리케이션 사용 시 이를 시스템에 구축하면 거의 실시간으로 변경 사항을 전달받아서 처리할 수 있어 더 좋은 고객 경험을 제공할 수 있음

2) 사물 인터넷

• 센서와 장비에 스트림 처리를 적용하는 것이 흔한 이용 사례 중 하나로, 언제 유지 관리가 필요한지를 예측하는 것은 애플리케이션 모니터링과 비슷하지만 하드웨어에 적용되는 것이며 제조업, 이동통신, 케이블 TV 등 흔한 산업 현장에서 볼 수 있음

• 장비로부터 도착하는 이벤트를 대규모로 처리하고 유지 관리가 필요한 장비의 신호 패턴을 찾을 수 있음

3) 사기 탐지(이상 탐지)

시스템의 악성 이용자를 찾아내게 하는데, 신용카드 부정 사용 적발이나 주식 거래 부정 적발과 같은 보안 위협 감지 등에 부정 사용을 사용하면 일찍 탐지할수록 큰 이익이 되므로 이벤트에 빠르게 반응할 수 있는 거의 실시간에 가까운 시스템이 부정을 탐지하는데 오래 걸리고 처리를 되돌리는게 훨씬 복잡한 배치 방식보다 더 선호되므로 대규모의 이벤트 스트림에서 패턴을 식별하는 문제라고 할 수 있음

1)개발하고자 하는 애플리케이션 종류에 따른 스트림 처리 솔루션

개발하고자 하는 애플리케이션의 종류에 따라 사용할 스트림 처리 솔루션의 종류가 달라짐

1-1) 데이터 수집

하나의 시스템에서 데이터를 가져다 다른 시스템으로 전달하는 것이 목적으로, 대상 시스템에 맞춰 데이터에 약간의 변형을 가해야 함

1-2) 밀리초 단위 작업

거의 즉각적인 응답을 필요로 하는 애플리케이션들로, 사기 탐지 활용 사례의 상당수가 해당됨

1-3) 비동기 마이크로서비스

더 큰 비즈니스 프로세스의 일부로서 단일 기능을 수행하는데 이러한 애플리케이션은 성능 향상을 위해 로컬 상태에 이벤트 캐시를 유지해야 할 수 있음

​

1-4) 준 실시간 데이터 분석

데이터를 작게 분할하여 비즈니스에 유용한 인사이트를 얻어내기 위해 복잡한 집계 연산과 조인을 수행함

​

2) 스트림 처리 시스템 선택을 위한 현재 다루고 있는 문제에 대한 고려사항

2-1) 데이터 수집 문제 해결중 상태

스트림 처리 시스템이 필요한 것인지 아니면 좀 더 단순한 수집에 최적화된 카프카 커넥트 같은 시스템이 필요한 것인지 숙고해 볼 필요가 있는데, 스트림 처리 시스템이 필요하다면 데이터를 전송할 시스템에 대한 커넥터가 충분히 있는지 기능이 충분한지 확인해야 함

​

2-2) 수밀리초에 완료되어야 하는 작업 해결중 상태

카프카 스트림즈를 사용하는 것 자체를 다시 고려해봐야 하는데, 대체로 요청-응답 패턴이 더 나음

스트림 처리 시스템을 필요로 하는 것이 확실하다면 마이크로배치 방식을 택하는 것보다 이벤트 단위 저지연 방식을 지원하는 쪽을 선택해야 함

​

2-3) 비동기 마이크로서비스를 개발하고 있는 상태

사용하는 메시지 버스와 잘 통합되고 업스트림의 변경 사항을 마이크로서비스의 로컬 상태에 쉽게 반영할 수 있어야 하고 로컬 상태를 캐시 혹은 구체화된 휴 형태로 활용 가능한 스트림 처리 시스템이 필요함

​

2-4) 복잡한 분석 엔진을 개발하고 있는 상태

• 로컬 저장소를 잘 지원하는 스트림 처리 시스템이 필요한데 롴러 캐시나 구체화된 뷰를 유지하는 것보다 로컬 캐시 없이 구현하기 까다로운 복잡한 집계 연산, 윈도우, 조인 등을 잘 지원하는 것이 더 중요함

• 커스텀 집계, 우니도우, 다양한 조인 타입을 지원하는 API도 필요함

​

3) 보편적인 고려사항

3-1) 시스템 운영성

• 프로덕션 환경 배포가 쉬운가?

• 모니터링과 트러블슈팅이 쉬운가?

• 필요할 때 구모를 확장하거나 축소할 수 있는가?

• 현재 인프라스트럭처와 잘 통합되는가?

• 뭔가 실수를 해서 재처리를 해야할 때 어떻게 대응해야 하는가?

3-2) 사용 및 디버깅 용이성

같은 프레임워크를 사용해도 사용 버전에 따라 고품질의 애플리케이션을 작성하는데 걸리는 시간이 차리아 날 수 있는데, 개발 시간과 배포에 걸리는 시간이 매우 중요하므로 효율적인 시스템을 골라야 함

3-3) 어려운 일을 쉽게 해줌

대부분의 시스템이 고급 윈도우 집계 연산과 로컬 저장소 유지를 할 수 있다는 점을 내세우지만 문제가 있는데, 개발하는 당사자가 그걸 쉽게 쓸 수 있는지, 규모 확장과 장애 복구에 있어서의 세부 작동을 알아서 처리해주는지, 추상에 뭔가 구멍이 있거나 엉망이 된 상황을 제어해야 하는지 등을 시스템이 더 깔끔한 API와 추상화를 제공하고 세부사하을 알아서 처리할수록 개발자는 더 생산적일 수 있음

​

3-4) 커뮤니티

정기적으로 새롭고 좋은 기능이 추가되고 상대적으로 품질이 더 나을 수 있고, 버그가 빨리 수정되고 질문에 대한 답을 시기 적절하게 받을 수 있음