빅쿼리의 셔플링

In-memory query execution in Google BigQuery

 

빅쿼리의 In-memory query 실행

 

최유석

원글 주소 : https://cloud.google.com/blog/big-data/2016/08/in-memory-query-execution-in-google-bigquery

원글 작성자 : Hossein Ahmadi, BigQuery Technical Lead

 

개요

빅쿼리는 대규모의 데이터에 대해서 실시간에 가까운 쿼리실행 속도를 제공한다. 빅쿼리는 높은 성능을 제공하기 위해서 모든 연산이 메모리에서 이루어진다. 이러한 쿼리 실행 속도의 배경에는 빅쿼리의 심플한 아키텍쳐 구조와, 빠른 쿼리 실행이 가능하게 하는 메모리 기반의 연산, 그리고 페타바이트 급의 분석이 가능하게 하는 확장성 있는 데이터의 재분할(repartitioning), 셔플(shuffle) 기능이 있다. 이 글에서는 빅쿼리의 셔플(shuffle)에 대해서 자세히 알아보고, 구글의 페타비트급(petabit-scale) 네트워킹 기술(Jupiter)을 활용해 어떻게 높은 성능으로 메모리상에서 쿼리 실행을 가능하게 하는지 알아보고자 한다.

 

구글의 빅데이터 분석

시작하기에 앞서 구글의 경우, 구글의 수많은 서비스들(YouTube, Gmail. Search,등) 각각의 이용자들만 해도 수억 명에 이른다. 따라서 자연스럽게 구글 내부적으로 빅데이터를 분석하기 위한 기술들이 개발되고 사용되어 왔다. 구글은 오픈소스 플랫폼을 지향하는 모습으로 빅데이터를 포함한 여러 기술들의 논문을 공개해왔다. 몇가지 사례를 보면, Apache Hadoop의 근간이 되는 GFS와 MapReduce, Apache HBase의 근간이 되는 Big Table등이 있으며, BigQuery는 2010년 문서로 공개된 Dremel이라는 기술을 근간으로 한다.

빅쿼리의 성능과 인프라

빅쿼리의 쿼리 실행의 대한 내용을 알아보기에 앞서, 먼저 예제를 통해 빅쿼리의 성능과 인프라에 관련한 부분을 알아보도록 하자.

예제로 보는 빅쿼리의 성능

빅쿼리의 쿼리 실행속도, 즉 성능을 보기 위한 사례로 다음의 예제가 많이 사용된다.

(출처: https://cloud.google.com/blog/big-data/2016/01/anatomy-of-a-bigquery-query )

 

빅쿼리에서 제공하는 공개 데이터 셋에 저장 되어있는 위키피디아 페이지의 제목, 뷰수, 등의 정보가 저장된 테이블에서 1000억(100 billion record)개의 레코드를 스캔하고 제목(title)컬럼에서 정규표현식(regular expression)과 3개의 wildcard를 사용해서 해당하는 문자열("G.*o.*o.*g")을 검색한 결과를 가지고 해당 제목(title)을 가진 페이지의 뷰(views) 수를 카운트하고 각각의 언어별로 그룹으로 묶고 내림차순으로 정렬해서 결과를 보여주는 예제이다.

 

위의 예제에서 쿼리가 실행된 시간은 24.7초가 소요되었으나, 항상 동일한 속도로 실행되지는 않는다. 하지만 직접 실행해서 확인하더라도 동일한 쿼리에 대해서 약 30초 이내에 실행이 되는 것을 확인 할 수 있다.

 

예제로 보는 빅쿼리의 인프라

이제 앞의 예제가 실행되는 동안, 빅쿼리 내부적으로는 사용된 인프라에 대해서 알아보도록 하자. 먼저 위의 쿼리가 실행되는 약 30초 동안 약 4TB의 데이터를 처리하게 된다. 이때 사용되는 빅쿼리 인프라에 대한 대략적인 수치는

• 8800개의 CPU
• 3600 개의 Disk
• 330 Gigabit의 네트워크 트래픽

이 사용되고 위의 예제의 쿼리를 실행하는데 약 $20 정도의 비용이 발생한다. 위와 같은 쿼리실행 및 인프라 사용을 가능하게 하는 것은 빅쿼리가 방대한 규모의 인프라를 공유하는 멀티 테넌트(multitenancy)서비스이기 때문이다.

빅쿼리의 심플한 사용

일반적인 관계형 데이터베이스(RDBMS) 또는 여러 빅데이터 솔루션을 사용하더라도 위의 예제 분량의 쿼리실행을 하기 위해서는 해당 솔루션에 대한 이해와 설치, 분석에 필요한 코드실행이 필요하고, 또한 제대로 실행하기 위해서 끊임없는 튜닝과 모니터링을 포함한 유지보수가 필요하다. 하지만 빅쿼리의 경우 표준 ANSI SQL과 유사한 SQL(표준 ANSI SQL은 현재 베타로 지원)을 사용하여 쿼리를 입력하고 쿼리 실행버튼인 RUN QUERY만 클릭하면 자동적으로 빅쿼리 내부적으로 위와 비슷하거나, 더 많은 규모의 인프라를 사용하여 실행된 쿼리의 결과를 볼 수 있다.

 

빅쿼리 실행엔진(기본 아키텍쳐)

구글 클라우드 플랫폼(Google Cloud Platform)에서 서비스되고 있는 빅쿼리는 앞서 언급한 Dramel뿐만아니라 구글의 자체적인 기술들인 Borg, Colossus, Jupiter가 융합되어 놀라운 성능을 만들어낸다. 그렇다면 이와 같은 기술들이 어떤 구조로 동작하는지 알아보도록 하자.

 

빅쿼리의 기본 구조

(출처 : https://cloud.google.com/blog/big-data/2016/01/bigquery-under-the-hood )

• Dremel – 실행엔진
• Colossus – 분산 스토리지
• Borg – 대규모 클러스터 관리 시스템
• Jupiter – 네트워크 (양방향 대역폭 1 Petabit/sec)

Dremel

빅쿼리에서 요청이 들어오면 분산 스토리지인 Colossus에서 데이터를 읽고

그 해당 데이터를 페타비트(Petabit)급 네트워크인 Jupiter를 통해 컴퓨팅 처리 계층인 Dremel로 전달한다. 이 때, 데이터를 전달 받은 Dremel에서는 디스크 없이 컴퓨팅 처리가 이루어 진다. Dremel에서는 Leaf Nodes(slots), Mixers, Root server계층으로 나눠지고, Leaf Nodes (slots)에서 필요한 모든 계산을 수행하고, 앞의 예제 기준으로는 Colossus에서 읽은 1000억개의 각각의 레코드에 대한 정규표현식 체크를 하는 것이 이 계층(slots)이다. 그 다음 Leaf Nodes (slots)에서 처리된 데이터에 대해서 Mixer에서 aggregation(집계) 작업을 수행한다. 마지막으로 Mixer에서 집계된 데이터에 대한 정렬 등의 작업을 처리 하고 결과를 리턴 한다.

이렇게 데이터를 처리하는 단계에서 셔플(shuffle)작업이 이루어지며, Jupiter를 통해 빠른 속도로 데이터의 이동이 가능하게 한다. 또한 Dremel, Colossus, Jupiter의 리소스는 Borg를 통해 클러스터로 관리되고 동작한다.

 

 

Colossus

GFS(Google File System)의 후속 버전인 분산 파일 시스템으로 구글 데이터 센터에서 Colossus 클러스터를 가지고 있으며, 한번에 모든 빅쿼리 사용자의 요청에 대해 쿼리 실행 시 수천개의 디스크를 제공할 수 있도록 되어있다. 또한 콜로서스에 저장되는 모든 빅쿼리의 데이터는 자동으로 여러 데이터 센터에 분산 및 복제(replication)되어 저장되기 때문에 높은 안정성을 제공한다. Colossus를 통해 빅쿼리는 In-memory 데이터베이스와 유사한 성능을 보이지만, 저렴한 가격, 높은 병렬성, 확장성, 내구성을 가진 인프라이기 때문에 활용가치가 높다.

빅쿼리는 콜로서스에 구조화된 데이터에 대한 컬럼(Column) 기반 저장형식과 압축 알고리즘을 활용해서 데이터를 저장한다. 이를 통해 빠른 쿼리 실행을 가능하게 하며, 높은 비용을 들이지 않고도 데이터에 대한 저장을 할 수 있다.

Borg

구글의 대규모의 클러스터 관리시스템으로 구글의 서비스들에 사용되는 자원(CPU, RAM, disk, network)을 관리하고 할당하는 역할을 하며, 빅쿼리에서도 쿼리실행,등의 작업요청이 들어오면 미리 예약한 자원을 기반으로 필요한 자원을 할당해서 해당 작업의 실행을 가능하게 한다. 위의 예제처럼 쿼리가 실행 될 때, Dremel클러스터에 수천개의 CPU를 제공하는 것도 Borg의 역할이다. 또한 서버를 포함해서 전원, 네트워크에 이르기까지 수많은 오류에 대한 보호를 하고 관리를 하기 때문에 사용자 입장에서는 발생하는 문제를 개의치 않고 이용 할 수 있다.

 

Jupiter

대규모의 데이터를 원활하게 처리하려면 그에 걸맞는 네트워크도 필수로 필요하다. 구글이 클라우드 플랫폼을 통해 구축하여 제공하는 네트워크(Jupiter)는 양방향 1 Petabit/sec의 대역폭을 제공한다. 10만VM이 각각 10Gb로 통신할 수 있는 속도이다. 이렇듯 상상이상의 대역폭을 자랑하는 Jupiter를 통해 모든 쿼리에 대해서 데이터가 저장된 스토리지(Colossus)에 직접 접근해서 수초만에 테라바이트(terabytes)의 데이터를 읽을 수 있다.

빅쿼리 셔플(Shuffle)의 변화

하둡(Hadoop), 스파크(Spark), 구글 데이터 플로우(Google Cloud Dataflow)에 이르기까지 모든 분산 데이터 처리 시스템에서 셔플은 핵심요소로 작용한다. 데이터 처리 중간의 셔플단계는 크고 복잡한 조인, 집계 및 분석 작업의 실행을 위해 필요하다. 빅쿼리의 셔플은 쿼리 특성과 요구사항의 증가로 인한 셔플 처리 단계의 개선 요구로 인해, 2014년 메모리 기반(디스크 스풀링 지원) 및 구글의 데이터 센터에서 네트워크 기술(Jupiter)로 특별하게 설계되었고 새롭게 개발된 인프라로 이전되었다. 게다가, 특정한 분산 작업을 넘어선 활용사례(예를 들면, 해시조인)와 유연한 데이터 전송 시스템으로 설계되었다. 이 프로젝트는 데이터 전송 기술에 다년간의 연구 및 개발 노력의 산물이다.

빅쿼리 셔플의 차이점

빅쿼리의 셔플은 전용된 호스팅 원격 메모리의 노드 집합에 쿼리 처리의 다양한 단계에서 생산되는 중간 데이터를 셔플링해서 저장한다. 스파크(Spark), 피콜로(Piccolo) 등의 많은 시스템에서 일반적으로 데이터를 처리 할 때 중간 데이터 결과를 지속하여 저장한다. 그러나 빅쿼리는 셔플 작업에서 긴밀한 통합으로 메모리에서 처리된 중간 결과에 대해 다른 방향을 보인다.

 

빅쿼리 셔플의 구성요소

빅쿼리의 셔플 구성은 3가지의 컴포넌트로 구성된다.

빅쿼리 셔플의 구성요소

• producer - 파티션이 될 데이타를 생성한다.
• consumer - 파티션된 데이타를 수신 한다.
• controller – 셔플 프로세스를 관리한다.

 

셔플에서 Producer, Consumer, Controller는 다음과 같이 구현되어 있다.

Producer (producer_id) { void SendRow(row, consumer_id) : Called to send a row to a given consumer on behalf of this producer. } Consumer (consumer_id) { string ReceiveRow() : Called to receive one row for this consumer. } Controller { StartShuffle() : Called before any producers or consumers start sending or receiving rows. EndShuffle() : Called after all producers and consumers have successfully sent and received all rows. }

위의 API들은 공유 메모리의 개념을 제공하도록 설계되었기 때문에, 데이타 프로세싱 파이프라인상에서, 데이타를 파티셔닝하는데 범용적으로 사용될 수 있다.

 

빅쿼리 셔플(shuffle)의 개념

빅쿼리 셔플의 기본 동작은 다음의 그림과 같이 설명될 수 있다.

(출처 : https://cloud.google.com/blog/big-data/2016/08/in-memory-query-execution-in-google-bigquery )

 

빅쿼리의 셔플은 많은 Producer들이 효율적으로 원격지의 머신의 메모리에 데이타를 저장할 수 있도록 하며, Consumer역시 높은 처리량으로 동시 읽기가 가능하다. 특히 Producer는 인접한 메모리 블록에 생성된 행(rows)에 로그(Log)하고 색인(index)을 남긴다.

이 색인(index)은 Consumer가 해당 행을 효율적으로 검색하고 읽을 수 있도록 해준다

빅쿼리 셔플의 복합적인 동작

(출처 : https://cloud.google.com/blog/big-data/2016/08/in-memory-query-execution-in-google-bigquery )

 

재분할한 데이타를 메모리에 저장하는 특징이외에도, 빅쿼리 셔플은 또 다른 측면에서 MapReduce스타일의 셔플과 다르다. MapReduce 스타일의 셔플은 모든 행이 재정렬 된 다음에 데이타를 접근할 수 있는데 반하여 빅쿼리에서는 producers에 의해 셔플된 각각 행(rows)은 바로 workers에 의해 접근이 가능하다. 그래서 파이프 라인에서 분산된 작업을 수행 하게 할 수 있게 한다.

빅쿼리의 데이터 분할(partitioning)

데이터의 파티셔닝(partitioning)은 BigQuery의 쿼리의 성능에 상당한 영향을 미친다.

제대로 된 결과를 얻기 위해서는 Consumer와, Provider를 적절한 숫자로 맞추는 것이 중요하다. (빅쿼리가 자동으로 수행)

최적화 되지 않은 데이터의 분할로 쿼리가 매우 느리게 실행되거나, 심지어는 자원의 제약으로 실패할 수 도 있다.

빅쿼리의 파티셔닝은 데이터 크기, 백그라운드의 부하, 기타 요인에 기초하여 쿼리에 사용 된 연산자의 종류에 따라 파티셔닝(분할)을 지능적으로 선택하는 동적 분할 메카니즘을 사용한다. 이로인하여 빅쿼리는 데이타의 특정 분포나, 키에 따른 정렬에 따른 오버헤드 없이 임의의 데이터 셋에 대한 효율적인 쿼리 실행을 할 수 있게 한다.

 

빅쿼리에서 콜로서스를 활용한 이점

빅쿼리는 페타바이트의 데이터에 쿼리를 사용할 수 있다. 대부분의 케이스에서 빅쿼리는 인메모리 셔플링을 통하여 데이타를 재분할 하는데, 메모리만 사용할 경우, 연산 비용이 매우 크기 때문에, 이를 해결하기 위해서 콜로서스 파일 시스템에 경우에 따라 데이타를 메모리에서 부터 이동하여 저장한다.

디스크의 경우 메모리에 비해서 많이 느리기 때문에, 빅쿼리에서는 디스크 억세스를 최소화하는 방법으로 성능 문제를 최소화한다.

 

빅쿼리는 multi-tenant service 이다.

빅쿼리는 사용자가 인프라를 공유해서 사용하는 멀티 테넌트 (multi-tenant) 서비스로 모든 고객이 쿼리를 실행하기 위해 VM의 클러스터의 크기를 조정하고 배포하거나, 리소스(자원)에 대한 프로비저닝이 필요 없다. 그렇게 하기 위해 빅쿼리의 셔플은 메모리에서 대부분의 쿼리를 실행 할 수 있는 지능적인 메모리 자원 관리 시스템을 사용한다. 빅쿼리는 고객으로부터 발생하는 부하의 변동에 따라 즉각적으로 적응한다.

 

결론

모든 빅쿼리의 쿼리는 하나 또는 다수의 셔플 동작을 포함하고, 단일 행 데이터 또는 페타바이트의 쿼리 데이터를 전송하는데 동일한 인프라를 사용한다. 구글 네트워크 기술과 함께 긴밀한 통합(tight intergration)으로 만들어내는 빅쿼리 셔플의 극한의 유연성은 빅쿼리의 사용자가 모든 규모에서 빠른 데이터 분석을 할 수 있게 한다.

 

 

참고자료

https://cloud.google.com/blog/big-data/2016/08/in-memory-query-execution-in-google-bigquery

 

https://cloud.google.com/blog/big-data/2016/01/anatomy-of-a-bigquery-query

 

https://cloud.google.com/blog/big-data/2016/01/bigquery-under-the-hood