스마트 기기와 SNS 덕분에 과거 어느 때보다 많은 데이터가 흘러 다니고 빠르게 쌓입니다. 다음 그림은 2013년에 인터넷에서 60초 동안 얼마나 많은 일이 벌어지는지를 나타낸 그림이다. Facebook에서는 1초마다 글이 4만 천 건 포스팅되고, 좋아요 클릭이 180만 건 발생합니다. 데이터는 350GB씩 쌓입니다. 이런 데이터를 실시간으로 분석하면 사용자의 패턴을 파악하거나 의사를 결정하는 데 참고하는 등 다양하게 사용할 수 있을 것입니다.

그림 1 인터넷에서 60초 동안 일어나는 일(원본 출처)

이 글에서는 주어진 응답 시간 내에 결과를 얻을 수 있는 실시간 데이터 처리 기법들을 소개합니다.

쌓고 보기에는 너무 오래 걸린다

빅데이터를 처리하는 프레임워크로 흔히 Hadoop MapReduce를 사용한다. MapReduce는 페타바이트 이상의 데이터를 여러 노드로 구성된 클라우드 환경에서 병렬 처리하는 기법으로, 함수형 프로그래밍에서 일반적으로 사용되는 Map과 Reduce 방식을 사용해 데이터를 처리한다. MapReduce는 대량 데이터를 분산 처리할 수 있는 좋은 기법이지만, 배치 방식으로 데이터를 처리하기 때문에 실시간으로 데이터를 조회하기 어렵다. 이런 단점을 극복하기 위해 최근 몇 년간 실시간 분산 쿼리나 스트리밍 처리 기법이 많이 연구되었다.

실시간 분산 쿼리는 클러스터를 구성하는 노드가 각자 쿼리를 처리하게 해(push down) 한 번에 처리할 데이터의 크기는 작게 하면서 이를 병렬 처리해 응답 시간을 실시간 수준으로 높이는 방식이다. Dremel의 논문을 기반으로 한 Cloudera의 Impala와 Apache Tez, 그리고 최근 공개된 Facebook의 Presto가 이 방식에 속한다.

스트리밍 처리는 끊임없이 들어오는 데이터를 유입 시점에 분석해 원하는 데이터 뷰로 미리 만드는 방식이다. 이 방식은 CEP(complex event processing)라고도 부르며, Twitter의 Storm과 Apache Spark가 이 방식에 속한다.

분산 쿼리의 기본 동작 방식: 파티셔닝과 셔플

분산 환경에서 데이터를 단일 뷰로 제공하는 것은 쉽지 않다. 이런 환경에서 기본적인 분산 처리 방식을 살펴보면 다음과 같다.

클러스터를 구성하는 여러 노드가 있고, 노드는 각자 데이터를 일부분 가졌다고 하자. 여기서 데이터 구조는 여러 형태가 될 수 있지만 편의상 테이블이라 한다. 파티셔닝은 특정 키를 기준으로 이 테이블을 여러 노드로 분할해 저장하는 방식이다. 키를 범위로 나눠 저장하거나(범위 파티셔닝, range partitioning), 키 값을 해시 키로 사용해 수평적으로 데이터를 나눠 저장할 수 있다(해시 파티셔닝, hash partitioning). 파티셔닝은 노드 간의 키 중첩을 없애기 때문에 각 노드에서 파티션 키를 조인 키로 쓰는 경우 독립적인 조인 처리가 가능하다. 그림 2는 3개 노드로 데이터를 해시 파티셔닝한 후 조인하는 과정을 나타낸다.

그림 2 해시 파티셔닝 후 조인 과정(원본 출처)

하지만 파티션 키가 아닌 다른 키로 조인해야 한다면, 데이터를 조인 키 단위로 모으는 재분할(re-partitioning) 과정이 필요하다. 이 과정을 셔플(shuffle)이라 하는데, 셔플은 노드 간 데이터 전송을 많이 발생시키므로 꽤 부담이 된다. 만약 조인할 테이블들 중 매우 작은 크기의 테이블이 있다면, 굳이 셔플하지 않고도 그 테이블만 다른 모든 노드로 복제한 다음 각 노드에서 조인할 수 있다. 이런 방식을 broadcast join이라 한다.

데이터 집계(aggregation)는 두 단계로 나눠볼 수 있다. 개별 노드에서 각자 결과를 집계하고(local aggregation) 그 결과를 모아 전체 결과로 집계한다(global aggregation). 그루핑(group by) 또는 정렬(order by)도 집계 처리와 같이 두 단계로 수행한다. 각 노드의 수행 결과를 모아 처리하는 방식은 MapReduce와 유사하다.

파이프라이닝

파티셔닝 데이터를 조인하는 기법은 정적인 데이터를 대상으로 하지만 데이터 스트림에 대해서도 적용해 볼 수 있다. 분산 쿼리 과정은 네트워크 토폴로지상의 메시지 처리 방식과 비슷하다. 예를 들어, 조인은 서로 다른 두 노드가 보낸 메시지를 다운스트림 노드가 합하는 과정과 같다.

일반적인 테이블과 데이터 스트림을 조인하려면 어떻게 할까? 그림 3은 테이블 R1과 데이터 스트림 S1, S2, S3을 조인하는 과정이다. R1에 대한 해시 테이블이 이미 생성되어 있다면, 스트림에서 입력되는 데이터 행마다 해시 테이블을 조회해 조인 결과를 만들 수 있다.

그림 3 테이블 R1과 데이터 스트림 S1, S2, S3 조인 과정(원본 출처)

그렇다면, 데이터 스트림 간의 조인은 어떻게 할까? 이는 대칭 해시 조인(symmetric hash join)을 사용해 수행할 수 있다. 일반적인 해시 조인은 조인할 한쪽 테이블에서 해시 테이블을 만든(build) 후, 다른 쪽 테이블의 데이터를 읽으면서 키가 그 해시 테이블에 있는지 확인(probe)하는 방식으로 진행한다. 하지만 스트림 데이터의 경우 해시 테이블을 미리 만들 수 없기 때문에 각 테이블에 데이터가 들어올 때마다 1)자신의 해시 테이블에 키를 넣고 2)상대 테이블의 해시 테이블을 확인해 키가 있으면 조인 결과로 추가한다. 이 과정을 양쪽 스트림에서 모두 수행하기 때문에 조인 결과는 누락되지 않지만 무한히 입력되는 데이터에 대해 해시 테이블을 유지하기는 어렵다. 따라서 해시 테이블을 특정 데이터 윈도만큼만 유지하거나, LFU(least frequently used) 방식으로 유지한다. 이런 경우 조인 키는 시간에 따라 값이 변하게 된다.

경축! 아무것도 안하여 에스천사게임즈가 새로운 모습으로 재오픈 하였습니다.
어린이용이며, 설치가 필요없는 브라우저 게임입니다.
https://s1004games.com

그림 4 대칭 해시 조인(원본 출처)

스트리밍 방식

이제 병렬 스트리밍을 조인 처리할 수 있음을 확인했다. 기존 쿼리 엔진으로도 스트리밍 처리가 어느 정도 가능한 셈이다. 이번에는 반대로 대량의 이벤트를 처리하는 스트리밍 처리 시스템에서 분산 쿼리를 지원할 수 있을까? 스트리밍 데이터는 시간에 따라 지나가 버리지만 쿼리는 시점에 관계없이 동일한 조건으로 요청할 수 있다. 혹은 클러스터 내 노드 중 하나에 장애가 발생하더라도 쿼리를 다시 수행해 쿼리 수행을 마쳐야 할 수도 있다. 끊임없이 들어오는 데이터를 지속적으로 쌓아두려면 많은 리소스가 필요한데, 리소스의 제약이 있다면 또 어떻게 처리해야 할까? 다음 방법들은 이런 물음에 대한 답이다.

lineage tracking

lineage tracking은 Storm에서 사용하는 방식이다. 분산 처리를 위한 노드 토폴로지에서 데이터는 일련의 이벤트처럼 흘러 다닌다. 최초에 입력된 데이터는 여러 노드에 걸쳐 처리된 후 최종 결과를 데이터베이스에 저장한다. 노드는 입력된 데이터를 특정 조건으로 필터링하거나, 이전 노드에서 온 데이터를 합치거나(merging), 그루핑(grouping)하거나, 정렬(ordering)하거나 집계(aggregation)할 수 있다. 이런 처리 과정 중 특정 노드에서 수행에 실패하면 이를 검출하고 다시 수행해야 한다.

lineage tracking은 각 이벤트마다 이진 값 형태의 시그니처(signature)를 부여한다. 어떤 노드로 이벤트를 보내면 다운스트림 노드는 이벤트를 처리한 후 그 이벤트의 시그니처를 XOR 연산하고, 결국 이벤트가 최종 목적지에 도달하면 그 시그니처 값은 0이 된다. 이 시그니처는 이벤트 ID별로 테이블에 유지하며 주기적으로 확인한다. 만약 유입된 지 일정 시간(timeout)이 지난 이벤트의 시그니처가 0이 아니라면, 수행 실패로 간주하고 데이터 유입 지점부터 재수행한다.

그림 5 lineage tracking(원본 출처)

이 장애 검출 방법은 중앙에서 관리할 필요가 없는(decentralized) 방식으로 이벤트 처리가 최소 한 번은 수행됨을 보장한다. 하지만 두 번 이상도 수행될 수 있기 때문에 트랜잭션 처리는 어렵다.

트랜잭션 토폴로지(transaction topology)는 이 문제를 풀기 위해서, 유입 데이터를 일정 단위로 묶어 배치 처리하되 이를 트랜잭션으로 처리한다. 각 배치마다 1)고유의 트랜잭션 식별자(TID)를 부여하고, 2)여러 노드를 거쳐 배치 데이터를 처리한 후 3)데이터베이스에 반영한다(커밋한다). 배치 처리하다 실패하면 TID를 사용해 배치 처리를 처음부터 다시 하고, 데이터베이스에 최종 결과를 반영할 때는 TID 순서대로 반영하도록 한다. 따라서, 정확히 한 번의 배치 처리만 반영되는 것을 보장한다. 여러 배치가 함께 수행될 때는 상태가 없는(stateless) 성격의 배치 처리는 병렬로 수행하고 데이터베이스 반영만 직렬화하기 때문에 효율적이기도 하다. 예를 들어, 트위터는 사용자 URL별로 트윗 수를 집계하기 위해 이 방식을 사용한다. URL별 누적 트윗 수를 유지하기 위해 스트림으로 입력되는 URL을 일정 단위로 묶은 후 URL별로 트윗 수를 세고 카운트 데이터베이스에서 이전에 저장한 URL별 트윗 수 값에 합산한다. 트윗 수를 갱신할 때에는 현재 TID가 마지막으로 데이터베이스에 반영한 TID의 다음 트랜잭션인지 보고, 맞다면 그 숫자를 증가시킨다. 이와 같이 스트림 데이터를 일정 단위로 묶어 처리하면서 제대로 처리되었는지 확인하는 방식은 슬라이딩 윈도우(sliding window) 방식과 비슷하다.

Apache Spark 역시 슬라이딩 윈도우 방식으로 스트림을 처리한다. Spark는 스트림을 작은 단위로 묶고 그 단위별로 MapReduce로 배치 처리를 한다. 하지만 일반적인 MapReduce에 비해 한 작업(job)에서 처리하는 데이터 사이즈가 작기 때문에 중간 결과가 작고, 중간 결과를 메모리에 유지할 수 있기 때문에 수행 속도가 높다.

state checkpointing

checkpoint란 보통 특정 시점의 상태(state)를 영구 저장하는 것을 의미한다. 데이터를 변경할 때마다 영구 저장하지 않는다면, 시점을 복구하기 위해서는 변경 시작부터 복구할 시점까지 수행한 내용을 모두 재수행(redo)해야 한다. 이때, 재수행하는 양을 줄이기 위해 중간중간 상태를 영구 저장해 두고, 저장 시점 이후부터 재수행해도 된다. 배치 단위의 트랜잭션 처리에서도 이런 처리가 필요하다. 배치 처리가 상태를 가진다면(stateful), 즉 이전의 배치 결과를 보고 현재 배치를 처리한다면, 시점 복구를 위해서는 상태 정보가 필요하다. 예를 들면 카운트 계산 방법으로 기존 카운트에 현재 데이터 윈도에 대한 카운트만 증가시키는 경우가 있다. 복구를 하려면 시작부터 재수행을 해야 하는데, 이때 계산할 것이 많아 리소스가 많이 필요할 수 있다. 따라서, 배치 처리를 커밋할 때 계산 상태(computational state)도 같이 저장한다. 즉, 그 시점에 각 노드별로 메모리에 적재된 데이터 상태를 HDFS 같은 곳에 영구 저장한다. 하지만 상태 정보가 많다면 효율적으로 동작하지 않을 수 있다. 이런 경우 상태 변경을 모두 로그(transaction log)로 저장했다가 수행해서 특정 상태로 되돌릴 수 있다.

그림 6 state checkpointing(원본 출처)

Sketches

앞의 카운트 예와 같이 데이터 처리가 가산적(additive)이라면, 이전 결과에 유입된 데이터만 지속적으로 합산하면 최종 결과를 얻을 수 있다. 예를 들어, 일별 페이지 뷰는 시간별 페이지 뷰의 합으로 구할 수 있다. 하지만 가산이 가능한 경우는 데이터 분석에서 극히 일부이다. 예를 들어 평균 계산은 가산적이지 않다. 일 평균 구매량은 시간별 평균 구매량의 평균과는 다르다. 혹은, 이틀간 방문한 순 방문자(unique user) 수는 어제와 오늘 방문한 각 순 방문자 수의 합으로 알아낼 수 없다. 이런 경우, 확률적인 방법으로 근삿값을 계산할 수 있다.

Count-Min Sketch는 일종의 통계 카운트로, 빈도수에 대한 근사치를 구하는 효율적인 방법이다. false positive 방식으로, k × w의 2차원 배열이 있다고 가정하고, 값이 입력되면 그 값에 대해 k개의 해시 함수를 수행해 그 해시 값(w)을 인덱스로 엔트리의 값에 1을 더한다. 특정 값에 대한 빈도는 k개의 해시 값 위치에 있는 값의 최솟값보다 클 수 없다. 이 방식은 해시 값 충돌(collision)로 인해 정확하지 않을 수 있지만, 최대 빈도에 대한 근사치는 된다. 이 방식은 유한한 메모리 사이즈로 대량의 카운트를 유지할 수 있다.

그림 7 sketches(원본 출처)

마치며

빅데이터에 대한 실시간 분석은 여러 서비스 분야에서 요구하고 있다. 이런 분석은 서비스에 대한 사용자 피드백을 추출하고, 서비스 개선 방향을 선택하는 데 사용될 것이다. 혹은 진화된 서비스를 제공하기 위한 머신 러닝용 데이터 추출로도 사용될 수 있다. 빅데이터 처리를 위한 분산 기술들을 모아 플랫폼으로 만든다면 분석을 위한 많은 비용과 시간을 줄일 수 있을 것이라고 생각한다.

참고 문서

• http://highlyscalable.wordpress.com/2013/08/20/in-stream-big-data-processing/
• https://github.com/nathanmarz/storm/wiki/Transactional-topologies
• http://highlyscalable.wordpress.com/2012/05/01/probabilistic-structures-web-analytics-data-mining/