Kafka Streams

단일 이벤트 처리 (map/filter 패턴)

스트림 처리의 가장 기본적인 패턴은 각 이벤트를 고립시켜 처리하는 것이다. 스트림에서 불필요한 이벤트를 걸러 내거나 각 이벤트를 변환하는 데 흔히 쓰이므로 map/filter 패턴이라고도 부른다('map'이라는 용어는 map 단계가 이벤트를 변환하고 reduce 단계가 그것을 집계하는 map/reduce 패턴에서 온 것이다).

이 패턴에서 애플리케이션은 스트림에서 이벤트를 소비하고 각 이벤트를 수정한 다음 다른 스트림으로 생산한다. 로그 메시지를 읽어 ERROR 이벤트는 고우선순위 스트림으로, 나머지는 저우선순위 스트림으로 쓰는 애플리케이션이 한 예다. 이벤트를 JSON에서 Avro로 바꾸는 애플리케이션도 마찬가지다.

이런 애플리케이션은 각 이벤트를 독립적으로 다룰 수 있으므로 애플리케이션 안에 상태를 유지할 필요가 없다. 덕분에 장애 복구나 로드 밸런싱이 대단히 쉽다. 복구할 상태가 없으므로 이벤트를 다른 인스턴스에 넘겨 처리하게 하면 그만이다. 이 패턴은 단순한 프로듀서와 컨슈머만으로도 손쉽게 구현된다.

로컬 상태를 쓰는 처리와 그것이 만드는 세 가지 문제

대부분의 스트림 처리 애플리케이션은 정보 집계, 특히 시간 윈도우 집계에 관심이 있다. 거래일마다 주식의 최저·최고가를 찾고 이동 평균을 계산하는 것이 예다. 이런 집계는 스트림에 대한 상태 유지를 요구한다. 하루의 최저가와 평균가를 계산하려면 지금까지 본 최소·최대값을 저장해 두고 새 값과 비교해야 한다.

이 모든 것을 공유 상태가 아닌 로컬 상태로 할 수 있다. 예시의 각 연산이 group by 집계이기 때문이다. 즉 주식 시장 전체가 아니라 종목별로 집계한다. Kafka 파티셔너를 써서 같은 종목 기호를 가진 모든 이벤트가 같은 파티션에 쓰이도록 하고, 그러면 애플리케이션의 각 인스턴스는 자기에게 할당된 파티션의 모든 이벤트를 받는다(이는 Kafka 컨슈머의 보장이다). 결과적으로 각 인스턴스는 자기 파티션에 쓰인 종목 부분집합에 대한 상태만 유지하면 된다.

다만 애플리케이션에 로컬 상태가 생기면 훨씬 복잡해지고, 세 가지 문제를 반드시 해결해야 한다.

**메모리 사용.** 로컬 상태는 애플리케이션 인스턴스가 쓸 수 있는 메모리 안에 들어가야 한다.

**영속성.** 인스턴스가 종료될 때 상태가 사라지지 않아야 하고, 인스턴스가 다시 시작되거나 다른 인스턴스로 교체될 때 상태를 복구할 수 있어야 한다. 이것이 Kafka Streams가 아주 잘 처리하는 부분이다. 로컬 상태는 임베디드 RocksDB를 써서 인메모리로 저장되며, RocksDB는 재시작 후 빠른 복구를 위해 디스크에도 데이터를 영속화한다. 그런데 로컬 상태에 대한 모든 변경은 Kafka 토픽으로도 전송된다. 스트림 노드가 다운되어도 로컬 상태는 사라지지 않는다 — Kafka 토픽의 이벤트를 다시 읽어 쉽게 재생성할 수 있다. 예를 들어 로컬 상태에 'IBM의 현재 최저가 = 167.19'가 있다면 이를 Kafka에 저장해 두었다가 나중에 이 데이터로 로컬 캐시를 다시 채운다. Kafka는 이 토픽들에 로그 컴팩션을 적용해, 토픽이 끝없이 자라지 않고 상태 재생성이 언제나 가능하도록 한다.

**리밸런싱.** 파티션은 때때로 다른 컨슈머에게 재할당된다. 이때 파티션을 잃는 인스턴스는 마지막 정상 상태를 저장해야 하고, 파티션을 받는 인스턴스는 올바른 상태를 복구할 줄 알아야 한다.

스트림 처리 프레임워크들은 개발자의 로컬 상태 관리를 얼마나 도와주는지가 서로 다르다. 애플리케이션에 로컬 상태가 필요하다면 프레임워크와 그 보장을 반드시 확인해야 한다.

다단계 처리와 리파티셔닝

로컬 상태는 group by 형태의 집계에는 훌륭하지만, 가용한 정보 전체를 써야 나오는 결과라면 어떨까? 예를 들어 매일 상위 10개 종목 — 그날 개장부터 폐장까지 가장 많이 오른 10개 종목 — 을 발표하고 싶다고 하자. 각 인스턴스에서 로컬로 하는 어떤 작업으로도 충분하지 않다. 상위 10개 종목 전부가 다른 인스턴스에 할당된 파티션에 있을 수도 있기 때문이다.

필요한 것은 2단계 접근이다. 먼저 종목 기호별로 그날의 등락을 계산한다. 이것은 각 인스턴스에서 로컬 상태로 할 수 있다. 그런 다음 결과를 파티션이 하나뿐인 새 토픽에 쓴다. 이 파티션은 애플리케이션 인스턴스 하나가 읽어서 그날의 상위 10개 종목을 찾아낸다. 종목별 일일 요약만 담은 두 번째 토픽은 거래 자체를 담은 토픽보다 훨씬 작고 트래픽도 훨씬 적으므로, 애플리케이션 인스턴스 하나로 처리할 수 있다. 때로는 결과를 내기까지 더 많은 단계가 필요하다.

이런 다단계 처리는 map-reduce 코드를 짜 본 사람에게 익숙하다. reduce 단계가 여러 번 필요한 경우와 같기 때문이다. MapReduce에서는 reduce 단계마다 별도의 애플리케이션이 필요했지만, 대부분의 스트림 처리 프레임워크는 모든 단계를 하나의 애플리케이션에 담을 수 있게 해 주고, 어느 인스턴스(또는 워커)가 어느 단계를 실행할지는 프레임워크가 알아서 처리한다.

외부 조회 대신 스트림-테이블 조인

스트림 처리가 스트림 바깥의 데이터와 통합되어야 할 때가 있다. 데이터베이스에 저장된 규칙으로 거래를 검증하거나, 클릭스트림 정보를 클릭한 사용자 데이터로 보강하는 경우다.

가장 먼저 떠오르는 방법은 이렇다. 스트림의 클릭 이벤트마다 프로필 데이터베이스에서 사용자를 조회해, 원래 클릭에 사용자 나이와 성별을 더한 이벤트를 다른 토픽에 쓴다.

이 뻔한 방법의 문제는 외부 조회가 레코드마다 상당한 지연 — 보통 5~15밀리초 — 을 더한다는 것이다. 많은 경우 이는 실현 가능하지 않다. 외부 데이터스토어에 걸리는 추가 부하도 대개 받아들일 수 없다. 스트림 처리 시스템은 흔히 초당 10만~50만 이벤트를 처리할 수 있지만, 데이터베이스는 합리적인 성능으로는 초당 1만 건 정도밖에 처리하지 못한다. 더 잘 확장되는 해법이 필요하다.

좋은 성능과 확장성을 얻으려면 데이터베이스의 정보를 스트림 처리 애플리케이션 안에 캐싱해야 한다. 그런데 이 캐시 관리가 만만치 않다. 캐시 정보가 낡지 않게 하려면 어떻게 해야 하는가? 너무 자주 갱신하면 여전히 데이터베이스를 두들기는 셈이라 캐시의 의미가 없고, 너무 오래 기다리면 낡은 정보로 스트림 처리를 하게 된다.

그러나 데이터베이스 테이블에 일어나는 모든 변경을 이벤트 스트림으로 붙잡을 수 있다면, 스트림 처리 작업이 이 스트림을 듣고 있다가 데이터베이스 변경 이벤트를 근거로 캐시를 갱신할 수 있다. 데이터베이스 변경을 스트림의 이벤트로 붙잡는 것이 바로 CDC이며, Kafka Connect를 쓰면 CDC를 수행해 데이터베이스 테이블을 변경 이벤트 스트림으로 바꿔 주는 커넥터를 여럿 찾을 수 있다. 이렇게 하면 테이블의 개인용 사본을 갖게 되고, 데이터베이스 변경 이벤트가 있을 때마다 통보받아 자기 사본을 갱신할 수 있다.

그런 다음 클릭 이벤트가 오면 로컬 캐시에서 user_id를 조회해 이벤트를 보강한다. 로컬 캐시를 쓰기 때문에 훨씬 잘 확장되고 데이터베이스나 그것을 쓰는 다른 애플리케이션에 영향을 주지 않는다. 스트림 중 하나가 로컬에 캐싱된 테이블에 대한 변경을 나타내므로 이를 스트림-테이블 조인이라 부른다.

스트리밍 조인 — 언제나 윈도우 조인이다

테이블이 아니라 두 개의 '진짜' 이벤트 스트림을 조인하고 싶을 때가 있다. 무엇이 스트림을 '진짜'로 만드는가? 스트림은 무한하다. 스트림으로 테이블을 표현할 때는 스트림 이력의 대부분을 무시할 수 있다. 우리가 관심 있는 것은 테이블의 현재 상태뿐이기 때문이다. 그러나 두 스트림을 조인할 때는 이력 전체를 조인하는 것이고, 한 스트림의 이벤트를 같은 키를 갖고 같은 시간 윈도우에서 일어난 다른 스트림의 이벤트와 맞추려는 것이다. 그래서 스트리밍 조인을 윈도우 조인(windowed-join)이라고도 부른다.

예를 들어 사람들이 웹사이트에 입력한 검색 질의 스트림과, 검색 결과에 대한 클릭을 포함하는 클릭 스트림이 있다고 하자. 어떤 결과가 어떤 질의에 대해 가장 인기 있는지 알기 위해 검색 질의와 그들이 클릭한 결과를 맞추고 싶다. 검색어를 기준으로 맞추되 일정 시간 윈도우 안에서만 맞춰야 한다. 결과 클릭은 질의를 입력한 지 몇 초 뒤에 일어난다고 가정하므로, 각 스트림에 몇 초짜리 작은 윈도우를 유지하고 각 윈도우의 결과를 맞춘다.

Kafka Streams에서 이것이 동작하는 방식은 이렇다. 두 스트림(질의와 클릭)이 같은 키로 파티셔닝되고, 그 키가 곧 조인 키다. 그래서 user_id:42의 모든 클릭 이벤트는 clicks 토픽의 파티션 5에, user_id:42의 모든 검색 이벤트는 search 토픽의 파티션 5에 들어간다. Kafka Streams는 두 토픽의 파티션 5가 같은 태스크에 할당되도록 보장한다. 따라서 그 태스크는 user_id:42에 관련된 모든 이벤트를 보게 된다. 그리고 두 토픽에 대한 조인 윈도우를 임베디드 RocksDB 캐시에 유지하면서 조인을 수행한다.

순서를 벗어난 이벤트와 재처리

**순서를 벗어난 이벤트(out-of-sequence events)**를 다루는 일은 스트림 처리뿐 아니라 전통적인 ETL 시스템에서도 과제다. 이런 이벤트는 IoT 시나리오에서 자주, 그리고 예상대로 발생한다. 모바일 기기가 몇 시간 동안 WiFi 신호를 잃었다가 재접속하면서 몇 시간치 이벤트를 한꺼번에 보내는 경우다. 네트워크 장비 모니터링(고장 난 스위치가 수리될 때까지 진단 신호를 보내지 않는다)이나 제조업(특히 개발도상국 공장의 네트워크 연결은 악명 높게 불안정하다)에서도 마찬가지다.

스트림 애플리케이션은 다음 네 가지를 할 수 있어야 한다. 첫째, 이벤트가 순서를 벗어났음을 인식해야 한다 — event time을 살펴 현재 시각보다 오래되었음을 발견해야 한다. 둘째, 순서를 벗어난 이벤트를 조정할 기간을 정의해야 한다. 3시간 지연은 조정하고 3주가 지난 이벤트는 버리는 식이다. 셋째, 이 이벤트를 in-band로 조정할 능력이 있어야 한다. 이것이 스트리밍 애플리케이션과 배치 작업의 핵심 차이다. 일일 배치 작업이라면 작업이 끝난 뒤 몇 건이 도착했을 때 어제 작업을 다시 돌려 갱신하면 된다. 스트림 처리에는 '어제 작업 다시 돌리기'가 없다 — 동일한 연속 프로세스가 언제든 옛 이벤트와 새 이벤트를 함께 처리해야 한다. 넷째, 결과를 갱신할 수 있어야 한다. 결과가 데이터베이스에 쓰인다면 put이나 update로 충분하지만, 결과를 이메일로 보내는 애플리케이션이라면 갱신이 까다로울 수 있다.

Google의 Dataflow와 Kafka Streams를 포함한 몇몇 프레임워크는 processing time과 독립적인 event time 개념과, 현재 처리 시각보다 오래되거나 새로운 event time을 가진 이벤트를 다루는 능력을 내장하고 있다. 이는 보통 여러 개의 집계 윈도우를 로컬 상태에 갱신 가능한 상태로 유지하고, 그 윈도우 집계를 얼마나 오래 갱신 가능하게 둘지를 개발자가 설정하게 함으로써 이루어진다. 당연히 집계 윈도우를 오래 열어 둘수록 로컬 상태를 유지하는 데 더 많은 메모리가 필요하다.

Kafka의 Streams API는 집계 결과를 항상 결과 토픽에 쓴다. 그 토픽은 보통 컴팩트 토픽이라 키마다 최신 값만 보존된다. 늦게 도착한 이벤트 때문에 어떤 집계 윈도우의 결과를 갱신해야 하면, Kafka Streams는 그 집계 윈도우에 대한 새 결과를 쓰기만 하면 되고 그것이 이전 결과를 덮어쓴다.

**재처리(reprocessing)**에는 두 가지 변형이 있다. 하나는 개선된 새 버전의 애플리케이션을 옛 버전과 같은 이벤트 스트림 위에서 돌려, 첫 버전을 대체하지 않는 새 결과 스트림을 만들고 두 버전의 결과를 비교한 다음 어느 시점에 클라이언트를 새 결과로 옮기는 것이다. 다른 하나는 기존 애플리케이션에 버그가 있어 이를 고친 뒤 이벤트 스트림을 재처리해 결과를 다시 계산하는 것이다.

첫 번째 경우는 Kafka가 이벤트 스트림 전체를 확장 가능한 저장소에 오랜 기간 보관한다는 점 덕분에 간단하다. 필요한 것은 세 가지다. 새 버전 애플리케이션을 새 컨슈머 그룹으로 띄우고, 입력 토픽의 첫 오프셋부터 처리하도록 설정하고(그래야 입력 스트림의 모든 이벤트에 대한 자기 사본을 얻는다), 새 버전이 따라잡으면 클라이언트 애플리케이션을 새 결과 스트림으로 전환하는 것이다.

두 번째 경우는 더 까다롭다. 기존 애플리케이션을 입력 스트림의 처음으로 '리셋'하고, 두 버전의 결과가 섞이지 않도록 로컬 상태도 리셋하고, 경우에 따라 이전 출력 스트림도 정리해야 한다. Kafka Streams에 스트림 처리 애플리케이션의 상태를 리셋하는 도구가 있긴 하지만, 책의 권장은 애플리케이션 사본 두 개를 돌리고 결과 스트림 두 개를 만들 용량이 있다면 언제나 첫 번째 방법을 쓰라는 것이다. 첫 번째 방법이 훨씬 안전하다. 여러 버전을 오갈 수 있고 버전 간 결과를 비교할 수 있으며, 정리 과정에서 중요한 데이터를 잃거나 오류를 들여올 위험이 없다.