Part3. 11장 스트림 처리

실제로는 많은 데이터가 시간이 지나면서 점진적으로 도착하기 때문에 한정되지 않는다. 사용자는 데이터를 어제는 물론 오늘도 생산한다. 그리고 내일도 데이터 생산을 계속할 것이다. 사업을 중단하지 않는 한 그 과정은 절대 끝나지 않는다. 절대 “완료”되지 않는다.

인위적으로 일정 기간씩 데이터 청크를 나눠야 한다. 이런 지체를 줄이려면 좀 더 자주 처리를 실행해야 한다. 고정된 시간 조각이라는 개념을 완전히 버리고 단순히 이벤트가 발생할 때마다 처리해야 하는 방법이 스트림 처리의 기본이다.

 

이벤트 스트림 전송

입력이 파일일 때 대개 첫 번째 단계로 파일을 분석해 레코드의 연속으로 바꾸는 처리를 한다. 스트림 처리 문맥에서 레코드는 보통 이벤트라고 하지만 특정 시점에 일어난 사건에 대한 세부 사항을 포함하는, 작고 독립된 불변 객체라는 점에서 동일하다. 일반적으로 일 기준 시계를 따르는 발생 타임스탬프를 포함한다.

이론상으로는 파일이나 데이터베이스가 있으면 생산자와 소비자를 연결하기에 충분하다. 생산자는 데이터스토어에 기록하고 소비자는 주기적으로 폴링해 마지막으로 처리한 이벤트 이후에 새로 발생한 이벤트가 있는지 확인하는 방법이다. 그러나 지연 시간이 낮으면서 지속해서 처리하는 방식을 지향할 때 데이터스토어를 이런 용도에 맞게 설계하지 않았다면 폴링 방식은 비용이 크다. 보통 관계형 데이터베이스에는 트리거 기능이 있다. 그러나 기능이 제한적이고 데이터베이스를 설계한 이후에 도입한 개념이다.

 

메시징 시스템

일반적인 방법은 메시징 시스템을 사용하는 것이다. 생산자는 이벤트를 포함한 메시지를 전소한다. 그리고 메시지는 소비자에게 전달된다.

발행/구독(publish/subscribe) 모델에서는 여러 시스템들이 다양한 접근법을 사용한다. 

1. 생산자가 소비자가 메시지를 처리하는 속도보다 빠르게 메시지를 전송한다면 어떻게 될까? 시스템은 메시지를 버리거나 큐에 버퍼링 하거나 흐름을 제어하여 생산자가 메시지를 더 보내지 못하게 막는다.)

2. 노드가 죽거나 일시적으로 오프라인이 된다면 어떻게 될까? 손실되는 메시지가 생길까? 데이터베이스를 사용할 때처럼 지속성을 갖추려면 디스크에 기록하거나 복제 본 생성을 해야 한다. 그렇기 때문에 비용이 든다.

메시지의 유실을 허용하지 말지는 애플리케이션에 따라 상당히 다르다. 이벤트 수를 세는 경우 메시지가 유실됐다는 것은 카운터가 잘못됐다는 의미이기 때문에 메시지를 신뢰성 있게 전송하는 일은 매우 중요하다. 자동으로 재시도하고 실패한 태스크가 남긴 부분적인 출력을 자동으로 폐기한다.

 

생산자에서 소비자로 메시지를 직접 전달하기

1. UDP 멀티캐스트는 낮은 지연이 필수인 주식 시장과 같은 금융 산업에서 널리 사용된다.

2. 브로커가 필요 없는 메시징 라이브러리.

3. 네트워크 상의 모든 장비로부터 지표를 수집하고 모니터링하는 데 UDP를 사용한다.

4. 소비자가 네트워크에 서비스를 노출하면 생산자는 직접 HTTP나 RPC요청을 직접 보낼 수 있다.

직접 메시징 시스템은 설계 상황에서는 잘 동작하지만 일반적으로 메시지가 유실될 수 있는 가능성을 고려해서 애플리케이션 코드를 작성해야 한다. 허용 가능한 결함은 상당히 제한적이다. 소비자가 오프라인이라면 메시지를 전달하지 못하는 상태에 있는 동안 전송된 메시지는 잃어버릴 수 있다.

 

메시지 브로커

직접 메시징 시스템의 대안으로 사용되는 방법은 메시지 브로커(메시지 큐)를 통해 메시지를 보내는 것이다. 근본적으로 메시지 스트림을 처리하는 데 최적화된 데이터베이스의 일종이다. 서버로 구동되고 생산자와 소비자는 서버의 클라이언트로 접속한다.

브로커에 데이터가 모이기 때문에 이 시스템은 클라이언트의 상태 변경에 쉽게 대처할 수 있다. 지속성 문제가 브로커로 옮겨갔기 때문이다. 메시지를 메모리에만 보관하거나, 디스크에 기록한다. 소비 속도가 느린 소비자가 있으면 일반적으로 브로커는 큐가 제한 없이 계속 늘어나게 한다.

또한 큐 대기를 하면 소비자는 일반적으로 비동기로 동작한다. 생산자가 메시지를 보낼 때 생산자는 메시지를 버퍼에 넣었는지 만 확인하고 처리하기까지 기다리지 않는다.

메시지 브로커와 데이터베이스의 비교

1. 데이터베이스는 명시적으로 삭제될 때까지 데이터를 보관하지만 메시지 브로커는 소비자에게 데이터 배달이 성공할 경우 삭제한다.

2. 메시지를 빨리 지우기 때문에 작업 집합이 상당히 작다고 가정한다. 소비자가 느려 메시지 브로커가 많은 메시지를 버퍼링 해야 한다면 전체 처리량이 저하된다.

3. 데이터베이스는 보조 색인을 지원하고 검색을 위한 다양한 방법을 지원하는 반면 메시지 브로커는 특정 패턴과 부합하는 토픽의 부분 집합을 구독하는 방식을 지원한다.

4. 데이터베이스에 질의할 때 그 결과는 그 시점의 데이터 스냅숏을 기준으로 한다. 앞선 결과가 기간이 지나 유효하지 않다는 점을 알 길이 없다. 반대로 메시지 브로커는 임의 질의를 지원하지 않지만 데이터가 변하면 클라이언트에게 알려준다.

 

복수 소비자

복수 소비자가 같은 토픽에서 메시지를 읽을 때 사용하는 주요 패턴

1. 로드 밸런싱 : 각 메시지는 소비자 중 하나로 전달된다. 소비자는 해당 토픽의 메시지를 처리하는 작업을 공유한다.

2. 팬 아웃 : 각 메시지는 모든 소비자에게 전달된다. 여러 독립적인 소비자가 브로드 캐스팅된 동일한 메시지를 서로 간섭 없이 청취할 수 있다. 같은 입력 파일을 읽어 여러 다른 일괄 처리 작업에서 사용하는 것과 동일하다.

확인 응답과 재전송

소비자는 언제라도 장애가 발생할 수 있다. 메시지를 잃어버리지 않기 위해 메시지 브로커는 확인 응답을 사용한다. 메시지 처리가 끝났을 때 브로커가 메시지를 큐에서 제거할 수 있게 명시적으로 알려야 한다. 부하 균형 분산과 결합할 때 이런 재전송 행위는 메시지 순서에 영향을 미친다. 메시지 브로커는 메시지 순서를 유지하려 노력할지라도 부하 균형 분산과 메시지 재전송을 조합하면 필연적으로 메시지 순서가 변경된다.

파티셔닝 된 로그

일괄 처리의 핵심 기능은 입력이 읽기 전용이기 때문에 입력을 손상하지 않고 반복 수행해 각 처리 단계를 실험할 수 있다는 것이다. 메시징 처리는 그렇지 않다. 브로커가 확인 응답을 받으면 메시지를 삭제하기 때문에 이미 받은 메시지는 복구할 수 없다. 소비자를 다시 실행해도 동일한 결과를 받지 못한다.

데이터베이스의 지속성 있는 저장 방법과 메시징 시스템의 지연 시간이 짧은 알림 기능을 조합하는 것이 로그 기반 메시지 브로커의 기본 아이디어다.

 

로그를 사용한 메시지 저장소

로그는 단순히 디스크에 저장된 추가 전용 레코드의 연속이다. 브로커를 구현할 때도 같은 구조를 사용한다. 생산자가 보낸 메시지는 로그 끝에 추가하고 소비자는 로그를 순차적으로 읽어 메시지를 받는다. 로그 끝에 도달하면 새 메시지가 추가됐다는 알림을 기다린다. 디스크 하나를 쓸 때보다 처리량을 높이기 위해 확장하는 방법으로 로그를 파티셔닝하는 방법이 있다. 각 파티션 내에는 브로커는 모든 메시지에 offset의 단조 증가하는 순번을 부여한다. 파티션 내 전체 메시지는 전체 순서가 있기 때문에 순번을 부여하는 것은 타당하다.

아파치 카프카가 이런 방식으로 동작하는 로그 기반 메시지 브로커다. 이런 메시지 브로커는 모든 메시지를 디스크에 저장하지만 여러 장비에 메시지를 파티셔닝해 초당 수백만 개의 메시지를 처리할 수 있고 메시지를 복제함으로써 장애에 대비할 수 있다.

 

로그 방식과 전통적인 메시징 방식의 비교

로그 기반 접근법은 당연히 팬 아웃 메시징 방식을 제공한다. 소비자가 서로 영향 없이 독립적으로 로그를 읽을 수 있고 메시지를 읽어도 로그에서 삭제되지 않기 때문이다.

메시지를 처리하는 비용이 비싸고 메시지 단위로 병렬화 처리하고 싶지만 메시지 순서는 중요하지 않다면 AMQP/JMS 방식의 메시지 브로커가 적합하다. 반면 처리량이 많고 메시지를 처리하는 속도가 빠르지만 메시지 순서가 중요하다면 로그 기반 접근법이 효과적이다.

소비자 오프셋

파티션 하나를 순서대로 처리하면 메시지를 어디까지 처리했는지 알기 쉽다. 따라서 브로커는 모든 개별 메시지마다 보내는 확인 응답을 추적할 필요가 없다. 주기적으로 소비자 오프셋을 기록하면 된다. 추적 오버헤드가 감소하고 일괄 처리와 파이프라이닝을 수행할 수 있는 기회를 제공해 로그 기반 시스템의 처리량을 늘리는 데 도움을 준다.

소비자 노드에 장애가 발생하면 소비자 그룹 내 다른 노드에 장애가 발생한 소비자의 파티션을 할당하고 마지막 기록된 오프셋부터 메시지를 처리하기 시작한다.

디스크 공간 사용

디스크 공간을 재사용하기 위해 실제로는 로그를 여러 조각으로 나누고 가끔 오래된 조각을 삭제하거나 보관 저장소로 이동한다. 소비자 처리 속도가 느려 메시지가 생산되는 속도를 따라잡지 못하면 소비자 오프셋이 이미 삭제한 조각을 가리킬 수도 있다. 즉 메시지 일부를 잃어버릴 가능성이 있다. 원형 버퍼, 링 버퍼의 경우.

소비자가 생산자를 따라갈 수 없을 때

메시지 버리기, 버퍼링, 배압을 적용해야 한다. 로그 기반 접근법은 이 방식으로 분류하자면 대용량이지만 고정 크기의 버퍼를 사용하는 버퍼링 형태다.

소비자가 뒤처져 필요한 메시지가 디스크에 보유한 메시지보다 오래되면 필요한 메시지는 읽을 수 없다. 그래서 브로커는 버퍼 크기를 넘는 오래된 메시지를 자연스럽게 버린다. 버퍼가 커질수록, 사람이 소비자 처리가 느린 문제를 고쳐 메시지를 잃기 전에 따라잡을 시간을 충분히 벌 수 있다. 소비자가 종료되거나 죽으면 자원 소비가 중단되고 소비자 오프셋만 남는다.

오래된 메시지 재생

AMQP같은 인메모리 유형의 메시지 브로커에서는 처리하고 확인 응답하는 작업은 파괴적 연산이다. 로그 기반 메시지 브로커는 메시지를 소비하는 게 오히려 파일을 읽는 작업과 더 유사하다.

전날 분량의 메시지를 재처리하기 위해 다른 위치에 출력을 기록할 수 있다. 로그 기반 메시징과 일괄 처리는 변환 처리를 반복해도 입력 데이터에 영향을 전혀 주지 않고 파생 데이터를 만든다. 로그 기반 메시징 시스템은 많은 실험을 할 수 있고 오류와 버그를 복구하기 쉽기 때문에 데이터 플로를 통합하는 데 좋은 도구다.

 

데이터베이스와 스트림

시스템 동기화 유지하기

사용자 요청에 대응하기 위한 OLTP 데이터베이스, 공통 요청의 응답 속도를 높이기 위한 캐시, 검색 질의를 다루기 위한 전문 색인, 분석용 데이터 웨어하우스가 그 예다. 이 시스템 각각은 데이터의 복제본을 가지고 있고 그 데이터는 목적에 맞게 최적화된 형태로 각각 저장된다.  동일한 데이터가 여러 다른 장소에서 나타나기 때문에 서로 동기화가 필수다.

주기적으로 데이터베이스 전체를 덤프하는 작업이 너무 느리면 대안으로 사용하는 방법으로 이중 기로이 있다. 데데이터가 변할 때마다 애플리케이션 코드에서 명시적으로 각 시스템에 기록한다. 동시 쓰기, 쓰기 실패의 경우로 불일치가 발생하는 현상이 발생한다.

변경 데이터 캡쳐

최근 들어 변경 데이터 캡쳐(CDC)에 관심이 높아지고 있다. 데이터베이스에 기록하는 모든 데이터의 변화를 관찰해 다른 시스템으로 데이터를 복제할 수 있는 형태로 추출하는 과정이다. 변경 내용을 스트림으로 제공할 수 있으면 특히 유용하다.

변경 데이터 캡처의 구현

검색 색인과 데이터 웨어하우스에 저장된 데이터는 레코드 시스템에 저장된 데이터의 또 다른 뷰일 뿐인 파생 데이터 시스템이라 할 수 있다. 변경 데이터 캡처는 파생 데이터 시스템이 레코드 시스템의 정확한 데이터 복제본을 가지게 하기 위해 레코드 시스템에 발생하는 모든 변경 사항을 파생 데이터 시스템에 반영하는 것을 보장하는 메커니즘이다.

로그기반 메시지 브로커는 메시지 순서를 유지하기 때문에 원본 데이터베이스에서 변경 이벤트를 전송하기에 적합하다. 트리거를 설정하는 방법도 있지만 고장 나기 쉽고 성능 오버헤드가 상당하다. 복제 로그를 파싱하는 방식은 스키마 변경 대응 등 해결해야 할 여러 문제가 있다.

변경 데이터 캡처는 메시지 브로커와 동일하게 비동기 방식으로 동작한다. 데이터베이스 시스템은 변경 사항을 커밋하기 전에 변경 사항이 소비자에게 적용될 때까지 기다리지 않는다. 복제 지연의 문제가 발생하는 단점이 있다.

초기 스냅숏

저장 엔진은 주기적으로 같은 키의 로그 레코드를 찾아 중복을 제거하고 각 키에 대해 가장 최근에 갱신된 내용만 유지하는 컴팩션. 컴팩션과 병학 과정은 백그라운드로 실행된다. CDC 시스템에서 모든 변경에 기본키가 포함되게 하고 키의 모든 갱신이 해당 키의 이전 값을 교체한다면 특정 키에 대해 최신 쓰기만 유지하면 충분하다.

로그에 데이터베이스에 있는 모든 키의 최신 값이 존재하는 것이 보장된다. CDC 원본 데이터베이스의 스냅숏을 만들지 않고도 데이터베이스 콘텐츠 전체의 복사본을 얻을 수 있다.

아파치 카프카는 로그 컴팩션 기능을 제공한다. 메시지 브로커는 일시적 메시징뿐만 아니라 지속성 있는 저장소로도 사용 가능하다.

 

이벤트 소싱(sourcing)

이벤트 소싱은 도메인 주도 설계 커뮤니티에서 개발한 기법이다. 스트리밍 시스템에 관련한 유용한 아이디어를 포함한다. 변경 데이터 캡처와 유사하게 애플리케이션 상태 변화를 모두 변경 이벤트 로그로 저장한다. 큰 차이점은 이 아이디어를 적용하는 추상화 레벨이 다르다는 점.

데이터 모델링에 쓸 수 있는 강력한 기법이다. 애플리케이션 관점에서 사용자의 행동을 분변 이벤트로 기록하는 방식은 변경 가능한 데이터베이스 상에서 사용자의 행동에 따른 효과를 기록하는 방식보다 훨씬 유의미하다.

이벤트 소싱은 연대기 데이터 모델과 유사하다. 또한 이벤트 로그와 별 모양 스키마에서 발견한 사실 테이블 사이에도 유사점이 있다. 이벤트 스토어 같은 특화된 데이터베이스는 이벤트 소싱을 사용하는 애플리케이션을 지원하게끔 개발한다.

이벤트 로그에서 현재 상태 파생하기

이벤트 로그 그 자체로는 유용하지 않다. 사용자는 시스템의 현재 상태를 보기를 원하지 수정 히스토리를 원하지 않기 때문이다. 따라서 이벤트 소싱을 사용하는 애플리케이션은 시스템에 기록한 데이터를 표현한 이벤트 로그를 가져와 사용자에게 보여주기에 적당한 애플리케이션 상태로 변환해야 한다. 결정적 과정이어야 한다. 다시 수행하더라도 이벤트 로그로부터 동일한 애플리케이션 상태를 만들 수 있어야 하기 때문이다.

변경 데이터 캡처와 마찬가지로 이벤트 로그를 재현하면 현재 시스템 상태를 재구성할 수 있다. 하지만 로그 컴팩션은 다르게 처리해야 한다. 사용자의 행동의 결과로 발생한 상태 갱신 메커니즘이 아닌 사용자 행동 의도를 표현해야 한다.

 

명령과 이벤트

이벤트와 명령을 구분하는 데 주의한다. 사용자 요청이 처음 도착했을 때 이 요청은 명령이다. 이 시점에서 실패할 수도 있다. 이벤트는 생성 시점에 사실이 된다. 사용자가 나중에 예약을 변경하거나 취소하더라도 그 사실은 여전히 진실이며 변경이나 취소는 나중에 추가된 독립적인 이벤트다. 따라서 명령의 유효성은 이벤트가 되기 전에 동기식으로 검증해야 한다. 예를 들어 하나는 가예약 이벤트, 다른 하나는 유효한 예약에 대한 확정 이벤트다. 분할해서 비동기 처리로 유효성 검사를 할 수 있다.

 

상태와 스트림 그리고 불변성

입력 파일에 손상을 주지 않고 기존 입력 파일에 얼마든지 실험적 처리 작업을 수행할 수 있다. 일반적으로 데이터베이스는 애플리케이션의 현재 상태를 저장한다고 생각한다. 이런 식으로 표현하면 읽기에 최적화되고 질의를 처리하는 데 매우 편리하다. 상태의 본질은 변하는 것이다. 그래서 데이터베이스는 데이터 삽입 외에도 데이터 갱신과 삭제를 지원한다.

상태가 변할 때마다 해당 상태는 이벤트의 마지막 결과다. 모든 변경 로그는 시간이 지남에 따라 바뀌는 상태를 나타낸다. 변경 로그를 지속성 있게 저장한다면 상태를 간단히 재생성할 수 있는 효과가 있다. 이벤트 로그를 레코드 시스템으로 생각하고 모든 변경 가능 상태를 이벤트 로그로부터 파생된 것으로 생각하면 시스템을 거치는 데이터 흐름에 관해 추론하기 쉽다.

불변 이벤트의 장점

불변 이벤트는 현재 상태보다 훨씬 많은 정보를 포함한다. 데이터베이스에 잘못된 데이터를 기록했을 때 코드가 데이터를 덮어썼다면 복구하기가 매우 어렵다. 추가만 하는 불변 이벤트 로그를 썼다면 문제 상황의 진단과 복구가 훨씬 쉽다.

동일한 이벤트 로그로 여러 가지 뷰 만들기

불변 이벤트 로그에서 가변 상태를 분리하면 동일한 이벤트 로그로 다른 여러 읽기 전용 뷰를 만들 수 있다.

일반적으로 데이터에 어떻게 질의하고 접근하는지 신경 쓰지 않는다면 데이터 저장은 상당히 직관적인 작업이다. 스키마 설계, 색인, 저장소 엔진이 가진 복잡성은 특정 질의와 특정 접근 형식을 지원하기 위한 결과로 발생한다. 이런 이유로 데이터를 쓰는 형식과 읽는 형식을 분리해 다양한 읽기 뷰를 허용한다면 상당한 유연성을 얻을 수 있다. 명령과 질의 책임의 분리(CQRS) -> 비정규화에 관한 논쟁은 의미가 거의 없다.

 

동시성 제어

이벤트 로그의 소비가 대개 비동기로 이뤄진다는 점이다. 사용자가 로그에 이벤트를 기록하고 이어서 파생된 뷰를 읽어도 기록한 이벤트가 뷰에 반영되지 않았을 가능성이 있다.

동기식으로 수행하는 방법이다. 이 방법을 쓰려면 트랜잭션에서 여러 쓰기를 원자적 단위로 결합해야 하므로 이벤트 로그와 읽기 뷰를 같은 저장 시스템에 담아야 한다.

반면 이벤트 로그를 현재 상태로 만들면 동시성 제어 측면이 단순해진다. 다중 객체 트랜잭션은 단일 사용자 동작이 여러 다른 장소의 데이터를 변경해야 할 때 필요하다. 그러면 사용자 동작은 한 장소에서 한 번 쓰기만 필요하다. 즉 이벤트를 로그에 추가만 하면 되며 원자적으로 만들기 쉽다.

이벤트 로그와 상태를 같은 방식으로 파티셔닝하면 간단한 단일 스레드 로그 소비자는 쓰기용 동시성 제어는 필요하지 않다.

불변성의 한계

영구적으로 모든 변화의 불변 히스토리를 유지하는 것이 어느 정도까지 가능할까? 대부분은 데이터를 추가하는 작업이고 갱신이나 삭제는 드물게 발생하는 작업부하는 불변으로 만들기 쉽다. 상대적으로 작은 데이터셋에서 매우 빈번히 갱신과 삭제를 하는 작업부하는 불변 히스토리가 감당하기 힘들 정도로 커지거나 파편화 문제가 발생할 수 있다.

성능적인 이유 외에도 관리상의 이유로 데이터를 삭제할 필요가 있는 상황일 수 있다. 실제로 원하는 바는 히스토리를 새로 쓰고 문제가 되는 데이터를 처음부터 기록하지 않았던 것처럼 하는 일이다. 적출, 셔닝.

데이터를 진짜로 삭제하는 작업은 어렵다. 많은 곳에 복제본이 남아 있기 때문이다.

 

스트림 처리

스트림을 처리하는 방법에는 크게 세 가지 방법이 있다.

1. 이벤트에서 데이터를 꺼내 유사한 저장소 시스템에 기록하고 다른 클라이언트가 이 시스템에 해당 데이터를 질의한다.

2, 이벤트를 사용자에 직접 보낸다. 경고, 푸시, 메일 같은 것.

3. 하나 이상의 입력 스트림을 처리해 하나 이상의 출력 스트림을 생산한다. 최종 출력에 (1, 2)에 이르기까지 여러 처리 단계로 구성된 파이프라인을 통과할 수도 있다.

스트림을 처리하는 코드 조각을 연산자, 작업이라 부른다. 유닉스 프로세스, 맵리듀스 작업과 밀접한 관련이 있다. 그 데이터플로의 양식도 비슷한데 스트림 처리자는 읽기 전용 방식으로 입력 스트림을 소비해 추가 전용 방식으로 다른 곳에 출력을 쓴다.

일괄 처리 작업과 가장 크게 다른 점은 스트림은 끝나지 않는 다는 점이다. 끝없는 데이터셋을 정렬하는 것은 말이 안 된다. 내결함성 메커니즘 또한 변경이 필요하다.

 

복잡한 이벤트 처리

CEP(complex event processing는 1990년대에 이벤트 스트림 분석용으로 개발된 방법이다. CEP는 특정 이벤트 패턴을 검색해야 하는 애플리케이션에 특히 적합하다. 정규 표현식으로 문자열에서 특정 문자 패턴을 찾는 방식과 유사하게 스트림에서 특정 이벤트 패턴을 찾는 규칙을 규정할 수 있다.

이 시스템에서 질의와 데이터의 관계는 일반적 데이터베이스와 비교했을 때 반대다. 데이터베이스는 일반적으로 데이터를 영구적으로 저장하고 질의를 일시적으로 다룬다. CEP 엔진은 질의는 오랜 기간 저장되고 입력 스트림으부터 들어오는 이벤트는 지속적으로 질의를 지나 흘러간다.

스트림 분석

CEP와 스트림 분석 사이의 경계는 불분명하지만 일반적으로 분석은 연속한 특정 이벤트 패턴을 찾는 것보다 대량의 이벤트를 집계하고 통계적 지표를 뽑은 것을 더 우선한다.

특정 유형의 이벤트 빈도 측정, 값의 이동 평균, 현재 통계 값의 비교. 등

일반적으로 이런 통계는 고정된 시간 간격 기준으로 계산한다. 집계 시간 간격을 윈도우라 한다.

 

구체화 뷰 유지하기

데이터베이스 변경에 대한 스트림은 캐시, 검색색인, 데이터 웨어하우스 같은 파생 데이터 시스템이 원본 데이터베이스의 최신 내용을 따라잡게 하는데 쓸 수 있다. 구체화 뷰를 유지하는 특별한 사례로 볼 수 있다.

스트림 상에서 검색하기

전통적인 검색 엔진은 먼저 문서를 색인하고 색인을 통해 질의를 실행한다. 반대로 스트림 검색은 처리 순서가 뒤집힌다. 질의를 먼저 저장한다. 그리고 CEP와 같이 문서는 질의를 지나가면서 실행된다.

메시지 전달과 RPC

메시지 전달 시스템을 RPC 대안으로 사용할 수 있다. 즉 액터 모델 등에서 쓰이는 서비스 간 통신 메커니즘으로 사용할 수 있다. 이런 시스템은 메시지와 이벤트에 기반을 두지만 일반적으로 이것들을 스트림 처리자로 생각하지는 않는다.

시간에 관한 추론

스트림 처리자는 종종 시간을 다뤄야 할 때가 있다. 특히 분석 목적으로 사용하는 경우에 그렇다. 일괄 처리에서 태스크는 과거에 쌓인 대량의 이벤트를 빠르게 처리한다.  반면 스트림 처리 프레임워크는 윈도우 시간을 결정할 때 처리하는 장비의 시스템 시계를 이용한다. 간단하다는 장점이 있다. 이벤트 생성과 이벤트 처리 사이의 간격이 무시할 정도로 작다면 꽤 합리적이다. 지연되면 안 된다.

 

이벤트 시간 대 처리 시간

메시지가 지연되면 메시지 순서를 예측하지 못할 수도 있다. 이벤트 시간과 처리 시간을 혼동하면 좋지 않은 데이터가 만들어진다.

이벤트 시간 기준으로 윈도우를 정의할 때 발생하는 문제는 특정 윈도우에서 모든 이벤트가 도착했다 거나 아직도 이벤트가 계속 들어오고 있는지를 확신할 수 없다는 점이다. 윈도우를 이미 종료한 후에 도착한 낙오자 이벤트를 처리할 방법이 필요하다.

1. 낙오자 이벤트는 무시한다. 정상적인 환경에서 낙오자 이벤트는 대체로 적은 비율이다.

2. 수정 값을 발행한다. 낙오자 이벤트가 포함된 윈도우를 기준으로 갱신된 값이다.

어떤 시계를 사용할 것인가?

이벤트가 발생한 시간, 장치 시계를 따른다.

이벤트를 서버로 보낸 시간, 장치 시계를 따른다.

서버에서 이벤트를 받은 시간, 서버 시계를 따른다.

 

윈도우 유형

텀블링 윈도우 : 고정 길이다. 그리고 모든 이벤트는 정확히 한 윈도우에 속한다. 각 이벤트의 타임스탬프를 가져와 가장 가까운 분이 되게끔 초 단위를 버려 윈도우를 결정하는 식으로 1분 텀블링 윈도우를 구현할 수 있다.

홉핑 윈도우 : 고정 길이를 사용한다. 결과를 매끄럽게 만들기 위해 윈도우를 중첩할 수 있다.

슬라이딩 윈도우 : 각 시간 간격 사이에서 발생한 모든 이벤트를 포함한다.

세션 윈도우 : 고정된 기간이 없다. 사용자가 짧은 시간 동안 발생시킨 모든 이벤트를 그룹화해서 세션 윈도우를 정의한다.

 

조인의 시간 의존성

스트림 처리자가 하나의 조인 입력을 기반으로 한 특정 상태를 유지해야 하고 다른 조인 입력에서 온 메시지에 그 상태를 질의한다. 상태를 유지하는 이벤트의 순서는 매우 중요하다. 파티셔닝된 로그에서 단일 파티션 내 이벤트 순서는 보존되지만 다른 스트림이나 다른 파티션 사이에서 순서를 보장하는 일반적인 방법은 없다.

 

내결함성

일괄 처리 프레임워크는 상당히 쉽게 결함에 대처할 수 있다. 작업의 결과와 동일함을 보장한다. 스트림 처리에서는 이 문제를 다루는 방법은 덜 직관적이다. 출력을 노출하기 전에 태스크가 완료될 때까지 기다리는 것은 해결책으로 사용할 수 없다. 스트림은 무한하고 처리를 절대 완료할 수 없기 때문이다.

마이크로 일괄 처리와 체크포인트

한 가지 해결책은 스트림을 작은 블록으로 나누고 각 블록을 소형 일괄 처리와 같이 다루는 방법이다. 마이크로 일괄처리라 한다. 스트림 연산자에 장애가 발생하면 스트림 연산자는 가장 최근 체크 포인트에서 재시작하고 해당 체크포인트와 장애 발생 사이의 출력은 버린다. 스트림 처리 프레임워크 내에서 마이크로 일괄 처리와 체크포인트 접근법은 정확히 한 번 시맨틱을 지원한다. 그러나 출력이 떠나자마자 스트림 처리 프레임워크는 실패한 일괄 처리 출력을 더 이상 지울 수 없다.

원자적 커밋 재검토

장애가 발생했을 때 정확히 한 번 처리되는 것처럼 보이려면 처리가 성공했을 때만 모든 출력과 이벤트 처리의 부수 효과가 발생하게 해야 한다.

멱동성

결국 목표는 처리 효과가 두 번 나타나는 일 없이 안전하게 재처리하기 위해 실패한 태스크의 부분 출력을 버리는 것이다. 멱등성에 의존하는 방법이 있다.

멱등 연산은 여러 번 수행하더라도 오직 한 번 수행한 것과 같은 효과를 내는 연산이다.

연산 자체가 멱등적이지 않아도 약간의 메타데이터로 연산을 멱등적으로 만들 수 있다.

실패 후에 상태 재구축하기

윈도우 집계나 조인용 테이블과 색인처럼 상태가 필요한 스트림 처리는 실패 후에도 해당 상태가 복구됨을 보장해야 한다. 원격 데이터 저장소에 상태를 유지하고 복제하는 방법이다. 또는 스트림 처리자의 로컬에 상태를 유지하고 주기적으로 복제하는 것이다. 그러면 스트림 처리자가 실패한 작업을 복구할 때 새 태스크는 복제된 상태를 읽어 데이터 손실 없이 처리를 재개할 수 있다.

 

정리

스트림 처리는 고정 크기의 입력이 아니라 끝이 없는 스트림 상에서 연속적으로 실행된다. 이런 관점으로 보면 스트림 처리에서 메시지 브로커와 이벤트 로그는 파일 시스템과 같은 역할을 한다.

 

AMQP/JMS 스타일 메시지 브로커

브로커는 개별 메시지를 소비자에게 할당하고 소비자는 받은 메시지를 처리하는 데 성공하면 확인 응답을 보낸다. 확인 응답을 받은 메시지는 삭제된다. RPC와 같은 비동기 양식에 적절하다. 메시지의 처리가 정확한 순서대로 이뤄지는 것이 중요하지 않고 메시지가 처리된 후에 이전으로 돌아가 과거 메시지를 다시 읽을 필요가 없다.

 

로그 기반 메시지 브로커

브로커는 한 파티션 내의 모든 메시지를 동일한 소비자 노드에게 할당하고 항상 같은 순서로 메시지를 전달한다. 병렬화는 파티션을 나누는 방식으로 사용한다. 소비자는 최근에 처리한 메시지의 오프셋을 체크포인트로 남겨 진행 상황을 추적한다. 브로커는 메시지를 디스크에 유지하기 때문에 필요한 경우 뒤로 돌아가 이전 메시지를 다시 읽을 수 있다.

로그 기반 접근법은 데이터베이스에서 사용되는 복제 로그와 로그 구조화 저장엔진과 유사하다. 이 접근법은 스트림 처리 시스템이 입력 스트림을 소비해 파생된 상태나 파생된 출력 스트림을 생성할 때 특히 적합하다.

암묵적으로 변경 데이터 캡처(CDC)를 통하거나 명시적으로 이벤트 소싱을 통해 변경 로그를 캡처할 수 있다. 로그 컴팩션을 사용하면 스트림에서 데이터베이스 내용의 전체 사본을 유지할 수 있다.

 

데이터베이스를 스트림처럼 표현하면 여러 시스템을 손쉽게 통합하는 기회가 열린다. 변경 로그를 소비해 그 로그를 파생 시스템에 적용하면 색인, 캐시, 분석용 시스템과 같은 파생 데이터 시스템을 항상 최신 상태로 유지할 수 있다. 처음부터 시작해 현재까지 모든 변경 로그를 소비하면 기존 데이터의 새로운 뷰를 구성하는 것도 가능하다.

상태를 스트림 형태로 유지하고 메시지를 재생하는 기능은 다양한 스트림 처리 프레임워크에서 스트림을 조인하거나 내결함성을 확보하기 위한 기술의 기초다. 이벤트 패턴 검색, 윈도우 집계 연산, 파생 데이터 최신성 유지를 포함한 스트림 처리의 목적.

 

세 가지 유형의 조인 구별

1. 스트림 스트림 조인(윈도우 조인) : 두 입력 스트림은 활동 이벤트로 구성하고 조인 연산자는 시간 윈도우 내에 발생한 관련 이벤트를 검색한다. 예를 들어 같은 사용자가 취한 행동 중 시간 차가 30분 이내인 두 개의 행동을 매치하는 식이다. 한 스트림 내에서 관련 이벤트를 찾는 다면 두 조인 입력은 사실 같은 스트림이다.

2. 스트림 테이블 조인(스트림 강화) : 한 입력 스트림은 활동 이벤트로 구성하고 다른 스트림은 데이터베이스의 변경 로그로 구성한다. 변경 로그는 데이터베이스의 최신 상태의 사본을 로컬에 유지한다. 각 활동 이벤트마다 데이터베이스에 질의하고 조인한 데이터를 추가한 활동 이벤트를 출력한다.

3. 테이블 테이블 조인(구체화 뷰 유지) : 양쪽 입력 스트림이 모두 데이터베이스의 변경 로그다 이 경우 한 쪽의 모든 변경을 다른 쪽의 최신 상태와 조인한다. 결과는 두 테이블을 조인한 구체화 뷰의 변경 스트림이 된다.

 

일괄 처리가 그랬던 것처럼 실패한 태스크의 부분 출력은 버려야 한다. 대신 마이크로 일괄 처리, 체크 포인팅, 트랜잭션, 멱등적 쓰기 등을 기반으로 한 세밀한 단위의 복구 메커니즘을 사용한다.