Kafka Streams

토폴로지 만들기

모든 스트림 애플리케이션은 최소한 하나의 토폴로지를 구현하고 실행한다. 토폴로지(다른 스트림 처리 프레임워크에서는 DAG, 즉 방향성 비순환 그래프라고도 부른다)는 모든 이벤트가 입력에서 출력까지 통과하는 연산과 전이의 집합이다.

단순한 애플리케이션조차 사소하지 않은 토폴로지를 갖는다. 토폴로지는 프로세서들로 이루어지며, 프로세서는 토폴로지 그래프의 노드다. 대부분의 프로세서는 데이터에 대한 연산 — filter, map, aggregate 등 — 을 구현한다. 그 외에 토픽에서 데이터를 소비해 넘겨 주는 소스 프로세서와, 앞선 프로세서로부터 데이터를 받아 토픽으로 생산하는 싱크 프로세서가 있다. 토폴로지는 언제나 하나 이상의 소스 프로세서로 시작해 하나 이상의 싱크 프로세서로 끝난다.

토폴로지 스케일링 — 태스크는 파티션 수로 정해진다

Kafka Streams는 하나의 애플리케이션 인스턴스 안에서 여러 실행 스레드를 허용하고, 분산된 인스턴스들 사이의 로드 밸런싱을 지원함으로써 확장한다. 여러 스레드로 한 대의 머신에서 실행하든 여러 머신에서 실행하든, 애플리케이션의 모든 활성 스레드가 데이터 처리 작업을 나눠 갖는다.

Streams 엔진은 토폴로지를 태스크로 쪼개어 실행을 병렬화한다. **태스크의 수는 Streams 엔진이 결정하며, 애플리케이션이 처리하는 토픽의 파티션 수에 달려 있다.** 각 태스크는 파티션의 부분집합을 책임진다. 그 파티션들을 구독하고 이벤트를 소비하며, 소비한 모든 이벤트에 대해 이 파티션에 적용되는 모든 처리 단계를 순서대로 실행한 다음 결과를 싱크에 쓴다. 이 태스크들이 Kafka Streams에서 병렬성의 기본 단위다. 각 태스크가 다른 태스크와 독립적으로 실행될 수 있기 때문이다.

애플리케이션 인스턴스마다 실행할 스레드 수는 개발자가 고를 수 있다. 여러 스레드가 있으면 각 스레드가 애플리케이션이 만든 태스크의 부분집합을 실행한다. 여러 서버에서 여러 인스턴스가 돌고 있다면 서버마다 스레드마다 서로 다른 태스크가 실행된다.

스트리밍 애플리케이션이 확장되는 방식은 이렇다. 처리 중인 토픽의 파티션 수만큼 태스크가 생긴다. 더 빨리 처리하고 싶으면 스레드를 더 추가한다. 서버의 자원이 부족해지면 다른 서버에서 애플리케이션 인스턴스를 하나 더 시작한다. Kafka가 작업을 자동으로 조정한다 — 각 태스크에 자기 파티션 부분집합을 할당하고, 각 태스크는 그 파티션의 이벤트를 독립적으로 처리하며 토폴로지가 요구한다면 관련 집계를 담은 자기 로컬 상태를 유지한다.

태스크 사이의 의존성 — 조인과 리파티셔닝

어떤 처리 단계는 여러 파티션의 결과를 필요로 하며, 이것이 태스크 사이의 의존성을 만든다.

**조인**이 첫 번째 경우다. 클릭스트림 예제처럼 두 스트림을 조인하려면 결과를 내기 전에 각 스트림의 파티션에서 데이터가 필요하다. Kafka Streams는 하나의 조인에 필요한 모든 파티션을 같은 태스크에 할당함으로써 이 상황을 해결한다. 그러면 그 태스크가 관련된 모든 파티션에서 소비해 독립적으로 조인을 수행할 수 있다. **이것이 Kafka Streams가 조인 연산에 참여하는 모든 토픽이 같은 수의 파티션을 가지고 조인 키로 파티셔닝되어 있을 것을 현재 요구하는 이유다.**

**리파티셔닝**이 두 번째 경우다. 클릭스트림 예제에서 모든 이벤트는 사용자 ID로 키가 매겨져 있다. 그런데 페이지별 통계를 내고 싶다면? 우편번호별이라면? 데이터를 우편번호로 리파티셔닝하고 새 파티션으로 집계를 돌려야 한다. 태스크 1이 파티션 1의 데이터를 처리하다가 데이터를 리파티셔닝하는 프로세서(groupBy 연산)에 도달하면, 셔플이 필요하다. 즉 이벤트를 다른 태스크로 보내 처리하게 해야 한다.

다른 스트림 프로세서 프레임워크와 달리, **Kafka Streams는 새 키와 새 파티션을 가진 새 토픽에 이벤트를 씀으로써 리파티셔닝한다.** 그러면 다른 태스크 집합이 그 새 토픽에서 이벤트를 읽어 처리를 이어간다. 리파티셔닝 단계는 토폴로지를 두 개의 서브토폴로지로 쪼개며, 각각이 자기 태스크를 갖는다. 두 번째 태스크 집합은 첫 번째 서브토폴로지의 결과를 처리하므로 첫 번째에 의존한다. 그러나 첫 번째와 두 번째 태스크 집합은 여전히 독립적으로, 병렬로 실행될 수 있다. 첫 번째 집합은 자기 속도로 토픽에 데이터를 쓰고 두 번째 집합은 그 토픽에서 소비해 자기 속도로 이벤트를 처리하기 때문이다. 태스크 사이에는 통신도 공유 자원도 없고, 같은 스레드나 서버에서 돌 필요도 없다. 파이프라인의 서로 다른 부분 사이의 의존성을 줄이는 것이 Kafka가 하는 더 유용한 일들 중 하나다.

장애 극복

확장을 가능하게 하는 바로 그 모델이 장애를 우아하게 다루는 것도 가능하게 한다.

먼저 Kafka 자체가 고가용이므로 Kafka에 영속화한 데이터도 고가용이다. 그래서 애플리케이션이 실패해 재시작해야 하면, Kafka에서 스트림 안의 마지막 위치를 조회해 실패 전에 커밋한 마지막 오프셋부터 처리를 이어갈 수 있다. 로컬 상태 저장소가 사라진 경우(예를 들어 그것이 저장된 서버를 교체해야 했던 경우)에도, 스트림 애플리케이션은 Kafka에 저장해 둔 변경 로그(change log)로부터 언제든 그것을 다시 만들 수 있다.

Kafka Streams는 또한 Kafka의 컨슈머 조정을 활용해 태스크에 고가용성을 제공한다. 태스크가 실패했는데 활성 상태인 다른 스레드나 다른 스트림 애플리케이션 인스턴스가 있다면, 그 태스크는 가용한 스레드 중 하나에서 재시작한다. 이는 컨슈머 그룹이 그룹 내 컨슈머 하나의 장애를 남은 컨슈머에게 파티션을 할당함으로써 처리하는 방식과 같다.

스트림 처리 사용 사례

스트림 처리(연속 처리)는 이벤트를 다음 배치까지 몇 시간 기다리지 않고 빠르게 처리하고 싶지만, 밀리초 안의 응답까지 기대하지는 않는 경우에 유용하다. 이는 맞는 말이지만 매우 추상적이므로, 실제 시나리오를 보자.

**고객 서비스.** 대형 호텔 체인에 방을 예약하고 확인 이메일과 영수증을 기다린다고 하자. 몇 분이 지나도 확인이 오지 않아 고객 서비스에 전화했더니 '시스템에 주문이 보이지 않는데, 예약 시스템에서 호텔과 고객 서비스 데스크로 데이터를 넣는 배치 작업이 하루에 한 번만 돌아서 내일 다시 전화 주세요. 이메일은 영업일 기준 2~3일 안에 보실 수 있습니다'라는 답을 듣는다. 좋은 서비스로 들리지 않는다. 우리가 정말 원하는 것은 호텔 체인의 모든 시스템 — 고객 서비스 센터, 호텔, 확인 메일을 보내는 시스템, 웹사이트 등 — 이 예약이 이루어진 지 몇 초 또는 몇 분 안에 갱신을 받는 것이다. 고객 서비스 센터가 체인 내 어느 호텔의 과거 방문 기록이든 즉시 조회할 수 있고, 프런트 데스크가 단골 고객임을 알아 업그레이드를 해 줄 수 있기를 바란다. 이 시스템들을 스트림 처리 애플리케이션으로 만들면 거의 실시간으로 갱신을 받고 처리할 수 있어 고객 경험이 좋아진다.

**사물 인터넷.** 세탁 세제 리필을 주문하는 가정용 기기부터 제약 제조의 실시간 품질 관리까지 폭이 넓다. 센서와 기기에 스트림 처리를 적용하는 매우 흔한 사용 사례는 예방 정비가 언제 필요한지 예측하는 것이다. 애플리케이션 모니터링과 비슷하되 하드웨어에 적용한 것으로, 제조업, 통신(고장 난 기지국 식별), 케이블 TV(사용자가 불평하기 전에 고장 난 셋톱박스 식별) 등 여러 산업에서 흔하다. 사례마다 패턴은 다르지만 목표는 같다. 대규모로 도착하는 기기 이벤트를 처리해 정비가 필요함을 알리는 패턴을 식별하는 것이다. 스위치의 패킷 유실, 제조 공정에서 나사를 조이는 데 필요한 힘의 증가, 케이블 TV 사용자의 잦은 재시작 등이 그런 패턴이다.

**사기 탐지(이상 탐지).** 신용카드 사기, 주식 거래 사기, 비디오 게임 치터, 사이버 보안 위험 탐지 등이 예다. 이 모든 분야에서 사기를 가능한 한 일찍 잡는 것에 큰 이익이 있으므로, 이벤트에 빠르게 반응할 수 있는 — 어쩌면 승인되기 전에 나쁜 거래를 멈출 수 있는 — 준실시간 시스템이, 사흘 뒤에 사기를 탐지해 수습이 훨씬 복잡해지는 배치 작업보다 훨씬 낫다.

사이버 보안에는 비커닝(beaconing)이라는 기법이 있다. 해커가 조직 내부에 악성코드를 심으면, 그것이 이따금 바깥으로 연결해 명령을 받아 온다. 아무 때나 아무 빈도로 일어날 수 있어 탐지가 어렵다. 일반적으로 네트워크는 외부 공격에는 잘 방어되어 있지만, 조직 내부에서 바깥으로 나가는 것에는 더 취약하다. 대규모 네트워크 연결 이벤트 스트림을 처리해 비정상적인 통신 패턴을 인식하면(예를 들어 이 호스트는 평소 그 특정 IP들에 접근하지 않는다는 것을 탐지), 보안 조직이 더 큰 피해가 생기기 전에 조기에 경보를 받을 수 있다.

스트림 처리 프레임워크를 고르는 기준

프레임워크를 고를 때는 어떤 종류의 애플리케이션을 쓸 계획인지가 중요하다. 애플리케이션 유형마다 다른 해법이 필요하다.

**Ingest** — 데이터를 한 시스템에서 다른 시스템으로 옮기되 대상 시스템에 맞게 약간의 수정을 하는 것이 목표인 경우. 이 문제를 풀려는 것이라면 스트림 처리 시스템이 정말 필요한지, Kafka Connect 같은 더 단순한 인제스트 전용 시스템이 낫지 않은지 다시 생각해 봐야 한다. 그래도 스트림 처리 시스템이 필요하다고 확신한다면, 대상 시스템에 대한 커넥터의 종류가 충분하고 품질이 높은지 확인해야 한다.

**Low milliseconds actions** — 거의 즉각적인 응답이 필요한 애플리케이션. 일부 사기 탐지 사례가 여기 해당한다. 이 경우에도 스트림이 정말 맞는 선택인지 다시 생각해야 한다. request-response 패턴이 더 잘 맞는 경우가 많다. 그래도 스트림 처리 시스템을 쓴다면 마이크로배치에 초점을 둔 것이 아니라 이벤트 단위 저지연 모델을 지원하는 것을 골라야 한다.

**비동기 마이크로서비스** — 상점 재고 갱신처럼 더 큰 비즈니스 프로세스를 대신해 단순한 동작을 수행하는 마이크로서비스. 성능 향상을 위해 이벤트를 캐싱하는 로컬 상태를 유지해야 할 수도 있다. 이 경우에는 선택한 메시지 버스(가급적 Kafka)와 잘 통합되고, 상류의 변경을 마이크로서비스의 로컬 캐시로 쉽게 전달하는 변경 캡처 능력이 있으며, 캐시나 구체화된 뷰 역할을 할 로컬 저장소를 잘 지원하는 시스템이 필요하다.

**준실시간 데이터 분석** — 복잡한 집계와 조인으로 데이터를 잘라 보며 비즈니스 인사이트를 만드는 애플리케이션. 이 경우에도 로컬 저장소 지원이 훌륭해야 하는데, 이번에는 로컬 캐시나 구체화된 뷰 유지가 아니라 다른 방법으로는 구현하기 어려운 고급 집계·윈도우·조인을 지원하기 위해서다. API가 커스텀 집계, 윈도우 연산, 여러 종류의 조인을 지원해야 한다.

사용 사례별 고려사항 외에 전반적으로 볼 것이 네 가지 더 있다. **시스템의 운영성** — 프로덕션 배포가 쉬운가, 모니터링과 문제 해결이 쉬운가, 필요할 때 확장·축소가 쉬운가, 기존 인프라와 잘 통합되는가, 실수가 있어 데이터를 재처리해야 하면 어떻게 되는가. **API의 사용성과 디버깅 용이성** — 같은 프레임워크의 서로 다른 버전 사이에서도 고품질 애플리케이션을 쓰는 데 걸리는 시간이 자릿수 단위로 차이 나는 경우가 있다. 개발 시간과 출시 시점이 중요하므로 자신을 효율적으로 만들어 주는 시스템을 골라야 한다. **어려운 일을 쉽게 만들어 주는가** — 거의 모든 시스템이 고급 윈도우 집계와 로컬 캐시 유지를 할 수 있다고 주장하지만, 질문은 그것을 쉽게 해 주느냐다. 규모와 복구에 관한 지저분한 세부를 알아서 처리해 주는가, 아니면 새는 추상(leaky abstraction)을 제공하고 대부분의 뒤처리를 개발자에게 떠넘기는가. **커뮤니티** — 고려 대상 대부분이 오픈소스이며, 활발한 커뮤니티를 대신할 것은 없다. 좋은 커뮤니티는 새 기능이 꾸준히 나오고, 품질이 상대적으로 좋고, 버그가 빨리 고쳐지고, 사용자 질문에 제때 답이 달린다는 뜻이다. 이상한 오류를 검색했을 때 정보를 찾을 수 있다는 뜻이기도 하다.