Kafka Streams

시간 — event time, log append time, processing time

시간은 스트림 처리에서 아마도 가장 중요하면서 가장 혼란스러운 개념이다. 분산 시스템에서 시간이 얼마나 복잡해질 수 있는지는 Justin Sheehy의 논문 'There is No Now'가 잘 보여 준다.

스트림 처리 맥락에서 시간에 대한 공통된 관념이 결정적으로 중요한 이유는, 대부분의 스트림 애플리케이션이 시간 윈도우 위에서 연산을 수행하기 때문이다. 예를 들어 주가의 5분 이동 평균을 계산한다고 하자. 프로듀서 하나가 네트워크 문제로 두 시간 동안 오프라인이었다가 두 시간치 데이터를 들고 돌아오면 어떻게 해야 하는가? 그 데이터 대부분은 이미 오래전에 지나가 버렸고 결과도 계산되어 저장된 5분 윈도우들에 속한다.

스트림 처리 시스템이 흔히 언급하는 시간 개념은 세 가지다.

**Event time**은 우리가 추적하는 이벤트가 실제로 일어나 레코드가 만들어진 시각이다. 측정이 이루어진 시각, 상점에서 물건이 팔린 시각, 사용자가 웹사이트의 페이지를 본 시각 등이다. 0.10.0 이후 버전에서 Kafka는 프로듀서 레코드가 생성될 때 현재 시각을 자동으로 붙인다. 만약 이 값이 애플리케이션이 생각하는 event time과 맞지 않는다면 — 예를 들어 이벤트가 일어난 한참 뒤에 데이터베이스 레코드를 근거로 Kafka 레코드를 만드는 경우 — event time을 레코드 안의 필드로 직접 넣어야 한다. 스트림 데이터를 처리할 때 보통 가장 중요한 시간이 이 event time이다.

**Log append time**은 이벤트가 Kafka 브로커에 도착해 저장된 시각이다. 0.10.0 이후 브로커는 그렇게 설정되어 있거나, 타임스탬프가 없는 구형 프로듀서로부터 레코드가 오면 이 시각을 자동으로 붙인다. 이 시간 개념은 스트림 처리에서 상대적으로 덜 중요하다. 우리가 보통 궁금한 것은 이벤트가 일어난 시각이기 때문이다. 예를 들어 하루에 생산된 장치 수를 센다면, 네트워크 문제로 이벤트가 다음 날에야 Kafka에 도착했더라도 실제로 그날 생산된 장치를 세고 싶을 것이다. 다만 실제 event time이 기록되지 않은 경우라면, log append time은 레코드 생성 이후 값이 바뀌지 않으므로 일관성 있게 사용할 수 있다.

**Processing time**은 스트림 처리 애플리케이션이 계산을 하기 위해 이벤트를 받은 시각이다. 이벤트 발생으로부터 밀리초일 수도, 몇 시간일 수도, 며칠일 수도 있다. 이 시간 개념은 각 스트림 처리 애플리케이션이 정확히 언제 이벤트를 읽었는지에 따라 같은 이벤트에 서로 다른 타임스탬프를 부여한다. 심지어 같은 애플리케이션 안의 두 스레드 사이에서도 달라질 수 있다. 그래서 이 시간 개념은 매우 신뢰할 수 없고 피하는 것이 최선이다.

시간을 다룰 때는 타임존에도 주의해야 한다. 데이터 파이프라인 전체가 하나의 타임존으로 표준화되어야 하며, 그렇지 않으면 스트림 연산 결과가 혼란스럽고 대개 무의미해진다. 서로 다른 타임존의 데이터 스트림을 다뤄야 한다면, 시간 윈도우 연산을 수행하기 전에 이벤트를 단일 타임존으로 변환할 수 있어야 한다. 이는 종종 레코드 자체에 타임존을 저장한다는 뜻이다.

상태 — 로컬 상태와 외부 상태

이벤트를 하나씩 독립적으로 처리하기만 하면 되는 동안에는 스트림 처리는 매우 단순한 작업이다. 예를 들어 Kafka에서 온라인 쇼핑 거래 스트림을 읽어 1만 달러가 넘는 거래를 찾아 담당 영업사원에게 이메일을 보내는 일이라면, Kafka 컨슈머와 SMTP 라이브러리로 몇 줄이면 된다.

스트림 처리가 정말 흥미로워지는 것은 여러 이벤트에 걸친 연산이 필요할 때다. 타입별 이벤트 개수 세기, 이동 평균, 두 스트림을 조인해 정보가 풍부해진 스트림 만들기 같은 것들이다. 이 경우 각 이벤트만 봐서는 부족하고 더 많은 정보를 추적해야 한다 — 이번 시간에 각 타입의 이벤트를 몇 개 봤는지, 조인이 필요한 모든 이벤트, 합계, 평균 등이다. 이렇게 이벤트 사이에 저장되는 정보를 상태(state)라고 부른다.

이동 카운트를 담을 단순한 해시 테이블처럼, 상태를 애플리케이션의 지역 변수에 저장하고 싶은 유혹이 자주 생긴다. 실제로 책의 여러 예제도 그렇게 했다. 그러나 이는 스트림 처리에서 상태를 관리하는 신뢰할 만한 방법이 아니다. 애플리케이션이 멈추면 상태가 사라지고 결과가 달라지기 때문이다. 이는 보통 원하는 결과가 아니므로, 가장 최근 상태를 영속화하고 애플리케이션을 시작할 때 복구하도록 신경 써야 한다.

스트림 처리에서 말하는 상태에는 두 종류가 있다.

**로컬(내부) 상태**는 스트림 처리 애플리케이션의 특정 인스턴스만 접근할 수 있는 상태다. 보통 애플리케이션 안에서 도는 임베디드 인메모리 데이터베이스로 유지·관리된다. 장점은 극도로 빠르다는 것이고, 단점은 사용 가능한 메모리 양에 제한된다는 것이다. 그 결과 스트림 처리의 많은 디자인 패턴이, 제한된 로컬 상태로 처리할 수 있도록 데이터를 서브스트림으로 파티셔닝하는 방법에 초점을 맞춘다.

**외부 상태**는 외부 데이터스토어, 흔히 Cassandra 같은 NoSQL 시스템에 유지되는 상태다. 장점은 사실상 크기 제한이 없고 애플리케이션의 여러 인스턴스, 심지어 다른 애플리케이션에서도 접근할 수 있다는 것이다. 단점은 시스템이 하나 더 늘어나면서 생기는 추가 지연과 복잡도다. 대부분의 스트림 처리 애플리케이션은 외부 저장소를 다루는 것 자체를 피하거나, 최소한 정보를 로컬 상태에 캐싱하고 외부 저장소와는 되도록 드물게 통신해 지연 오버헤드를 줄이려 한다. 이 방식은 대개 내부 상태와 외부 상태 사이의 일관성 유지라는 새로운 과제를 낳는다.

스트림-테이블 이중성

테이블은 레코드의 모음이며, 각 레코드는 기본 키로 식별되고 스키마가 정의한 속성들을 담는다. 테이블 레코드는 가변이다(갱신과 삭제가 가능하다). 테이블을 조회한다는 것은 특정 시점의 데이터 상태를 확인하는 일이다. 예컨대 CUSTOMERS_CONTACTS 테이블을 조회하면 모든 고객의 '현재' 연락처를 얻는다. 이력을 담도록 특별히 설계하지 않은 한, 과거 연락처는 테이블에 없다.

테이블과 달리 스트림은 변경의 이력을 담는다. 스트림은 각각이 어떤 변경을 일으킨 이벤트들의 나열이다. 테이블은 수많은 변경의 결과인 현재의 세계 상태를 담는다. 여기서 스트림과 테이블이 같은 동전의 양면임이 분명해진다. 세계는 늘 변하고, 어떤 때 우리는 그 변화를 일으킨 이벤트에 관심이 있고 어떤 때는 현재의 세계 상태에 관심이 있다. 두 관점을 자유롭게 오갈 수 있는 시스템이 한쪽만 지원하는 시스템보다 강력하다.

**테이블을 스트림으로 바꾸려면** 테이블을 수정하는 변경들을 붙잡아야 한다. insert, update, delete 이벤트를 모두 스트림에 저장하는 것이다. 대부분의 데이터베이스는 이 변경을 붙잡는 CDC(change data capture) 솔루션을 제공하며, 이 변경들을 Kafka로 흘려보내 스트림 처리에서 쓸 수 있게 해 주는 Kafka 커넥터도 많다.

**스트림을 테이블로 바꾸려면** 스트림이 담고 있는 모든 변경을 적용해야 한다. 이를 스트림을 '실체화한다(materializing)'고 부른다. 메모리든 내부 상태 저장소든 외부 데이터베이스든 테이블을 하나 만들고, 스트림의 모든 이벤트를 처음부터 끝까지 훑으며 상태를 바꿔 나간다. 끝나면 특정 시점의 상태를 나타내는 테이블을 얻는다.

신발 가게를 예로 들자. 소매 활동의 스트림 표현은 이런 이벤트들이다.

"빨강, 파랑, 초록 신발이 든 배송이 도착했다"
"파란 신발 판매"
"빨간 신발 판매"
"파란 신발 반품"
"초록 신발 판매"

지금 재고가 무엇인지, 지금까지 얼마를 벌었는지 알고 싶다면 뷰를 실체화해야 한다. 그러면 현재 파란 신발과 노란 신발이 있고 은행에 170달러가 있다는 결과를 얻는다. 반면 가게가 얼마나 바쁜지 알고 싶다면 스트림 전체를 보고 다섯 건의 거래가 있었음을 알 수 있다. 파란 신발이 왜 반품되었는지 조사할 수도 있다. 같은 데이터를 두 관점으로 보는 것이 각각 다른 질문에 답해 준다는 점이 핵심이다.

시간 윈도우 — 크기, 이동 간격, 갱신 가능 기간

스트림 위의 대부분의 연산은 윈도우 연산, 즉 시간 조각 위에서 이루어지는 연산이다. 이동 평균, 이번 주 최다 판매 상품, 시스템 부하의 99번째 백분위수 같은 것들이다. 두 스트림에 대한 조인 연산도 윈도우 연산이다 — 같은 시간 조각에서 일어난 이벤트끼리 조인하기 때문이다. 그런데 자기 연산에 어떤 종류의 윈도우가 필요한지 멈춰서 생각하는 사람은 매우 적다. 이동 평균을 계산할 때 정해야 할 것은 세 가지다.

**윈도우의 크기.** 5분 윈도우마다 평균을 낼 것인가, 15분마다인가, 하루 전체인가? 큰 윈도우는 결과가 더 매끄럽지만 지연이 크다. 가격이 오르면 작은 윈도우보다 알아채는 데 시간이 더 걸린다.

**윈도우가 얼마나 자주 이동하는가(advance interval).** 5분 평균이 1분마다 갱신될 수도, 1초마다일 수도, 새 이벤트가 들어올 때마다일 수도 있다. advance interval이 윈도우 크기와 같을 때 이를 텀블링 윈도우(tumbling window)라 부르기도 한다. 윈도우가 레코드마다 움직일 때는 슬라이딩 윈도우(sliding window)라 부르기도 한다.

**윈도우가 얼마나 오래 갱신 가능한 상태로 남는가.** 5분 이동 평균이 00:00-00:05 윈도우의 평균을 계산했다고 하자. 한 시간 뒤에 event time이 00:02인 결과가 몇 개 더 들어온다. 00:00-00:05 구간의 결과를 갱신할 것인가, 아니면 지난 일은 지난 일로 둘 것인가? 이상적으로는 이벤트가 각자의 시간 조각에 추가될 수 있는 일정 기간을 정의할 수 있어야 한다. 예를 들어 4시간까지 늦은 이벤트는 결과를 재계산해 갱신하고, 그보다 늦게 도착하면 무시하는 식이다.

윈도우는 시계 시간에 정렬(aligned)될 수 있다. 1분마다 움직이는 5분 윈도우라면 첫 조각이 00:00-00:05, 둘째가 00:01-00:06이 된다. 또는 정렬되지 않은 채 애플리케이션이 시작한 시점부터 시작할 수도 있고, 그러면 첫 조각이 03:17-03:22가 될 수 있다. 슬라이딩 윈도우는 새 레코드가 들어올 때마다 움직이므로 결코 정렬되지 않는다.