트위터: 수십억 개의 이벤트를 실시간 처리하기

해외 유명 IT 기업은 각자 자신들의 블로그를 운영해 개발 방법과 기업 문화 등을 소개하고 있습니다. 요즘IT는 이러한 IT 기업 블로그의 콘텐츠를 번역해 소개하는 시리즈를 준비했습니다. 이들은 어떻게 사고하고, 어떤 방식으로 일하는 걸까요?

이번 글은 대표적인 소셜 네트워크 서비스 트위터의 ‘Processing billions of events in real time at Twitter’번역했습니다. 트위터는 어떻게 수십억 개의 이벤트를 실시간으로 처리할 수 있을까요? 아래 글을 통해 그들의 데이터 처리 방법에 대해 알아보겠습니다.

트위터는 매일 페타바이트(PB) 규모의 데이터를 생산하며 약 4천억 개의 이벤트를 실시간으로 처리하고 있습니다. 이러한 이벤트는 다양한 데이터 소스로부터 생성되며, 하둡(Hadoop), 버티카(Vertica), 맨하탄(Manhattan) 분산 데이터베이스와 카프카(Kafka), 트위터 이벤트버스(Twitter Eventbus), GCS, 빅쿼리(BigQuery), PubSub 등 다양한 플랫폼과 스토리지 시스템을 사용하고 있습니다.

다양한 데이터 소스와 플랫폼에서 생성되는 수많은 유형의 데이터를 처리하기 위해 트위터의 데이터 플랫폼 팀은 일괄 처리를 위한 스캘딩(Scalding), 스트리밍을 위한 헤론(Heron), 일괄/실시간 처리를 위한 통합 프레임워크인 TSAR(TimeSeries AggregatoR), 데이터 검색과 소비를 위한 데이터 액세스 레이어(Data Access Layer)와 같은 도구를 자체 개발했습니다. 

그러나 엔지니어가 파이프라인을 실행하기 위해 사용하는 데이터 인프라스트럭처는 데이터의 빠른 증가로 인해 확장성을 유지하기가 매우 어렵습니다. 예를 들어, 대규모 데이터를 일괄 및 실시간으로 처리하는 상호작용과 참여(interaction and engagement)라는 파이프라인이 있습니다. 데이터 규모가 빠르게 증가함에 따라 우리는 스트리밍 대기 시간을 줄이고, 동시에 데이터 처리와 실시간 데이터 서비스에서 더 높은 정확도를 보장해야 하는 어려움에 직면하고 있었습니다.

트위터는 상호작용과 참여 파이프라인을 위해 다양한 실시간 스트림과 서버/클라이언트 로그에서 데이터를 수집하고 처리합니다. 이를 통해 서로 다른 집계 수준과 시간 단위 등 다양한 측정항목(Metric)[1]과 측정 기준(Dimension)[2]으로 트윗과 사용자 상호작용 데이터를 추출합니다. 집계된 상호작용 데이터는 특히 중요하며, 트위터의 광고 수익 서비스와 데이터 제품 서비스에서 노출 횟수와 참여도를 측정하는 데 사용합니다. 이뿐만 아니라 상호작용 데이터 조회 시, 서로 다른 데이터 센터의 스토리지에 있는 데이터로부터 낮은 지연 시간과 높은 정확도로 빠른 응답을 보장해야 합니다. 이러한 시스템을 구축하기 위해 전체 워크플로우를 전처리, 이벤트 집계, 데이터 제공 등 몇 가지 단계로 나누었습니다.

기존 아키텍처

트위터가 과거에 사용하던 아키텍처에 대해 먼저 알아보겠습니다. 먼저 일괄 처리와 실시간 파이프라인 처리를 수행하는 있는 람다 아키텍처로 구성되어 있으며, 서밍버드(Summingbird) 플랫폼 위에서 TSAR과 통합되어 있습니다. 람다 아키텍처에 대한 자세한 내용은 람다 아키텍처란 무엇인가?를 참고하세요. 

배치 컴포넌트는 HDFS(Hadoop Distributed File System)에 저장된 클라이언트 이벤트, 타임라인 이벤트, 트윗 이벤트와 같은 하둡 로그를 가져옵니다. 우리는 여러 개의 스콜딩(Scalding) 파이프라인을 구축했습니다. 이를 통해 원시 로그를 전처리하고 서밍버드 플랫폼에 공급했으며, 카프카 토픽(Kafka Topic)[3]에서 실시간 데이터를 가져왔습니다. 

아래의 그림처럼 실시간 데이터는 트위터 나이트호크(Nighthawk) 분산 캐시에 저장되고, 배치 데이터는 맨하탄 분산 스토리지 시스템에 저장됩니다. 그리고 쿼리 서비스는 이 두 개의 저장소로부터 고객 서비스에 필요한 데이터를 가져옵니다.

현재 3개의 서로 다른 데이터 센터에서 실시간 파이프라인과 쿼리 서비스가 동작하고 있습니다. 배치 작업에 소모되는 자원을 줄이기 위해 한 데이터 센터에서 배치 파이프라인을 실행하고, 다른 두 개 데이터 센터에 이를 복제하고 있습니다. 

트위터가 직면한 과제

대규모 데이터를 막힘없이 빠르게 처리해야 하므로 실시간 파이프라인에 대한 데이터 손실과 부정확성을 피해 갈 수 없습니다. 이벤트가 너무 많아 헤론 시스템 토폴로지의 컴포넌트인 볼트(Bolt)가 이를 제시간에 처리하지 못하면, 토폴로지 내부에 배압(back pressure)이 발생합니다. 또한 헤론 볼트는 가비지 컬렉션[4] 수집 비용이 크기 때문에 속도 저하가 발생하게 됩니다.

시스템에 장시간 배압이 가해지면 스파웃 지연이 헤론 볼트에 누적되면서 대기 시간이 길어집니다. 보통 이러한 문제가 한번 발생하면 토폴로지 지연이 감소하기까지 매우 오랜 시간이 걸립니다. 우리가 구축한 헤론 파이프라인에서 동작하는 스트림 매니저(토폴로지 컴포넌트 간의 튜플 라우팅을 관리함)가 자주 중단됐고 지연이 점차 증가했습니다.

이를 해결하기 위해 운영에서 현재 사용하는 방법은, 볼트가 스트림 처리를 다시 시작할 수 있도록 헤론 컨테이너를 재 시작하여 스트림 매니저를 불러오는 것입니다. 하지만 작업 과정에서 이벤트 손실이 발생할 수 있으며, 이로 인해 나이트호크 스토리지의 집계된 통계 정확도가 떨어질 수 있습니다.

배치 컴포넌트의 경우 페타바이트 규모의 데이터를 처리하고, 매시간 실행되며 맨해튼으로 데이터를 가져오는 등 많은 계산 과정을 동반한 파이프라인을 구축했습니다. 중앙 집중식 TSAR 쿼리 서비스는 맨해튼과 나이트호크 데이터를 통합하여, 고객 서비스를 위한 데이터를 제공하고 있습니다. 만약 실시간 데이터가 손실되면 TSAR 서비스가 집계가 완료되지 않은 측정 항목을 고객에게 제공하는 문제가 발생합니다.

이러한 데이터 손실 문제를 극복하고 시스템 대기 시간을 줄이며, 아키텍처를 최적화하기 위해 우리는 카파(Kappa) 아키텍처에 파이프라인을 구축해, 스트리밍 전용 모드로 이벤트를 처리할 것을 제안하고 있습니다. 카파 아키텍처에 대한 자세한 내용은 카파 아키텍처란 무엇인가?를 참고하세요. 이 방식은 배치 컴포넌트를 제거하고 실시간 컴포넌트에 의존해, 데이터의 낮은 대기 시간과 높은 정확도를 제공하므로 아키텍처를 단순화하고 배치 파이프라인에서 계산으로 인한 자원 소모가 없습니다.

새로운 아키텍처에서의 카프카와 데이터 흐름

새로운 아키텍처는 트위터 데이터 센터(Twitter Data Center) 서비스와 구글 클라우드 플랫폼(Google Cloud Platform)을 기반으로 구축되었습니다. 먼저 온프레미스[5]에서 최소 한 번(at-least-once) 카프카 토픽 이벤트를 PubSub[6]토픽 이벤트로 변환하는 전처리 과정을 거칩니다. 구글 클라우드에서는 스트리밍 데이터플로우(Dataflow)[7]를 통해 중복된 데이터를 제거하고, 실시간 집계를 수행하고, 데이터를 빅테이블(BigTable)[8]에 동기화합니다.

이러한 아키텍처를 구현하기 위해 가장 먼저 한 일은, 필드 변환과 매핑을 담당하고 이벤트를 카프카 토픽으로 보내는 이벤트 마이그레이션 전처리 파이프라인을 구축하는 것이었습니다. 우리는 카프카를 기반으로 커스텀 스트리밍 프레임워크를 사용하여, 최소 한 번의 이벤트 전달이 가능한 스트리밍 파이프라인을 개발했습니다. 

그다음 이벤트 스트리밍을 담당하는 이벤트 프로세서를 개발했습니다. 이벤트 프로세서는 이벤트를 PubSub 형식으로 변환하고, 관련된 UUID와 기타 메타 정보로 구성된 이벤트 컨텍스트를 생성합니다. UUID는 다운스트림에서 데이터플로우 작업자가 중복을 제거하기 위해 사용합니다. 또한 트위터 데이터 센터에서 구글 클라우드로 최소 한 번 메시지를 보내도록 내부 PubSub 게시자에 대해 무한에 가까운 재시도 설정을 적용했습니다. 새 PubSub 이벤트가 생성되면, 이벤트 프로세서는 이를 구글 PubSub 토픽으로 보냅니다.

구글 클라우드에서는 실시간 집계를 위해 구글 데이터플로우에서 동작하는 트위터 내부 프레임워크를 사용합니다. 데이터플로우 작업자는 실시간으로 중복 제거와 집계를 수행합니다. 이러한 중복 제거 작업의 정확도는 시간 범위를 어떻게 지정하는지에 따라 달라집니다. 우리는 시스템 튜닝을 통해 지정된 시간 범위에서 가장 효율적인 중복 제거를 수행할 수 있었습니다. 우리는 데이터를 빅쿼리(BigQuery)[9]에 동시에 쓰고, 중복 비율을 지속적으로 모니터링함으로써 높은 중복 제거 정확도를 달성할 수 있다는 사실을 입증했습니다. 자세한 내용은 여기에서 확인하세요. 마지막으로 집계 데이터의 개수를 빅테이블에 기록합니다.

서빙 레이어(Serving Layer)[10]의 경우, 트위터 데이터 센터의 프론트엔드[11]에서부터 빅테이블, 빅쿼리 등 다양한 백엔드[12]를 비롯하여 트위터에서 개발한 LDC(Large Data Collider) 쿼리 서비스를 사용합니다. 전체 시스템은 짧은 지연시간으로(최대 10초) 초당 수백만 개의 이벤트를 스트리밍할 수 있으며, 온프레미스와 클라우드 스트리밍 환경 어디에서나 높은 트래픽을 처리하도록 확장할 수 있습니다. 우리는 온프레미스 스트리밍 시스템 전체에서 데이터 손실이 없도록 보장하기 위해 클라우드 PubSub을 메시지 버퍼로 사용하고 있습니다. 다음으로 정확하게 한 번(exactly-once)에 가까운 데이터 전송을 달성하기 위한 중복 제거가 수행됩니다.

이 새로운 아키텍처를 통해 배치 파이프라인 구축 비용을 절감할 수 있습니다. 실시간 파이프라인의 경우 더 높은 집계 정확도와 안정적인 저지연 데이터 전송이 가능해집니다. 또한 하나 이상의 데이터 센터를 운영하더라도 서로 다른 실시간 이벤트 집계를 따로 수행하지 않아도 됩니다.

평가하기

시스템 성능 평가

아래는 표는 두 아키텍처 간의 성능 차이를 보여줍니다. 새로운 아키텍처는 기존 아키텍처의 헤론 토폴로지와 비교하여, 더 낮은 지연 시간과 더 높은 처리량을 보여줍니다. 또한 새로운 아키텍처에서는 지연된 이벤트 숫자를 확인할 수 있고, 실시간 집계 시 시스템을 재시작해도 기존 이벤트가 사라지지 않습니다. 새로운 아키텍처는 배치 컴포넌트가 없는 한층 진보된 구성으로 설계를 단순화하며, 기존 아키텍처에 부담을 주었던 연산 부하를 줄여줍니다.

표 1. 기존 아키텍처와 새 아키텍처에의 시스템 성능 비교.

* 스트리밍을 위해 헤론 내부적으로 압축한 데이터

집계 결과에 대한 유효성 검사

집계된 숫자를 검증하는 것은 두 단계로 진행됩니다. 첫째, 중복 제거 전후에 데이터플로우 내에서 발생한 중복 이벤트 비율을 확인합니다. 둘째, 모든 키에 대해 원본 TSAR 배치 파이프라인의 수와 데이터플로우에서 중복 제거 후 확인한 숫자를 직접 비교합니다. 

첫 번째 단계에서는 원시 이벤트를 내보내는 별도의 데이터플로우 파이프라인을 만들어, PubSub에서 빅쿼리로 쓰기 전 중복을 제거했습니다. 그런 다음 시간에 맞춰 지속적으로 숫자를 계산하는 쿼리가 실행되도록 만들었고, 이와 동시에 중복 제거된 이벤트 수를 빅쿼리로 내보내는 또 다른 데이터플로우 파이프라인을 생성했습니다. 이를 통해 중복 이벤트 비율과 중복 제거 후, 변경된 비율을 비교할 수 있게 되었습니다.

두 번째 단계에서는 중복 제거된 집계 데이터를 빅쿼리로 내보내고, 트위터 데이터 센터의 원본 TSAR 배치 파이프라인에서 생성된 데이터를 구글 클라우드의 빅쿼리에 적재하는 유효성 검사 워크플로우를 구축했습니다. 이는 모든 키(key)의 개수를 비교하기 위해 예약된 쿼리를 실행합니다.

그 결과 우리는 트윗 상호작용 스트림에 대한 일괄 처리 데이터 대비 95% 이상의 정확도를 확보할 수 있게 되었습니다. 일치하지 않는 5% 미만의 데이터를 조사한 결과, 이는 원본 TSAR 배치 파이프라인이 새로운 스트림 파이프라인에서 캡처한 지연된 이벤트를 버리기 때문이라는 사실을 알아냈습니다. 이를 바탕으로 우리의 새로운 시스템이 더 높은 정확도를 보여준다는 것을 입증해냈습니다.

결론

트위터는 TSAR에 구축된 기존 아키텍처를 트위터 데이터 센터와 구글 클라우드 플랫폼으로 구성된 하이브리드 아키텍처로 이관함으로써, 수십억 개의 이벤트를 실시간으로 처리하고 짧은 지연 시간, 높은 정확도, 안정성, 아키텍처 단순성, 엔지니어 운영 비용 절감을 달성하고 있습니다.

다음 단계

다음 단계로 클라우드 리전에 장애가 발생해도, 빅테이블 데이터 세트를 언제나 사용할 수 있도록 탄력적인 환경을 구성할 것입니다. 또한 고객이 새로운 아키텍처로 이관하여, LDC 쿼리 서버를 사용할 수 있도록 도울 것입니다.

[1] 측정하고자 하는 실제 값으로 설치수, 방문수, 주문량 등이 여기에 해당하며 리소스 모니터링과 같이 소프트웨어나 하드웨어 모니터링에 사용되기도 한다.

[2] 분석의 기준이 되는 대상 또는 객체로 분석 기간, 데모그래픽 정보, 시스템 환경, 유입경로 등이 여기에 해당한다.

[3] 카프카에서 데이터를 구분하고 저장하기 위해 사용하는 단위.

[4] 메모리 관리 기법 중의 하나로, 프로그램이 동적으로 할당했던 메모리 영역 중에서 필요없게 된 영역을 해제하는 것.

[5] 소프트웨어 및 솔루션을 클라우드 같이 원격 환경이 아닌 자체적으로 보유한 전산실 서버에 직접 설치해 운영하는 방식.

[6] Publish/Subscribe의 줄임말로 메시지 기반의 미들웨어 시스템을 말함. Publisher는 어떤 Subscriber가 있는지 모르는 상태에서 메시지를 전송하고 Subscriber는 Publisher에 대한 정보 없이 자신에게 필요한 메시지만을 수신한다.

[7] 구글에서 클라우드 제공하는 스트리밍 분석 서비스로 자동 확장과 일괄 처리를 통해 대기 시간, 처리 시간 및 비용을 최소화한다.

[8] 2015년 구글에서 발표한 분산 스토리지 시스템으로 수천 대의 서버에 걸쳐 페타바이트 단위의 데이터를 대규모로 확장하도록 설계되었다.

[9] 대용량 데이터의 저장 및 분석에 뛰어난 성능을 발휘하는 구글 클라우드 서비스.

[10] 데이터 처리 아키텍처에서 배치에 의해서 처리된 요약 데이터를 제공을 담당하는 부분.

[11] 웹에서 동작하는 UI(User Interface) 부분을 말하며, 사용자가 눈으로 보고 인식할 수 있는 영역.

[12] 사용자 눈에 보이지 않는 영역으로 비즈니스 로직 처리와 데이터 저장 및 관리뿐만 아니라 웹사이트의 클라이언트 측(client-side)에서 모든 것이 매끄럽게 작동할 수 있게 해준다.