빅데이터를 지탱하는 기술 Ch3 - 빅데이터 분산 처리

대규모 분산 처리의 프레임워크

구조화 데이터와 비구조화 데이터

SQL로 데이터를 집계하는 경우, 테이블이 명확하게 정의되어 있는 구조화된 데이터를 사용한다. 기존의 데이터 웨어하우스에서 데이터는 항상 구조화된 데이터로 축적하는 것이 일반적이었다. 하지만 텍스트, 이미지, 동영상 등의 구조화되지 않은 비구조화 데이터도 점차 늘어나 SQL로 제대로 집계할 수 없게 된다.

 

비구조화 데이터를 분산 스토리지 등에 저장하고 그것을 분산 시스템에서 처리하는 것이 데이터 레이크의 개념이다. 이 경우는 데이터를 가공하는 과정에서 스키마를 정의하고, 구조화된 데이터로 변환한다.

스키마리스 데이터: 기본 서식은 있지만, 스키마가 정의 안 됨.

CSV, JSON, XML 등의 데이터는 서식은 정해져 있지만, 칼럼 수나 데이터형은 명확하지 않아, 스키마리스 데이터라고 한다. 최근에는 JSON 형식을 이용하는 경우가 많고, 일일이 스키마를 정하는 것은 시간과 비용이 소모되기 때문에 JSON은 JSON 그대로 저장하고 거기서 데이터 분석에 필요한 필드만을 추출하는 편이 간단하다.

데이터 구조화의 파이프라인

기본적인 데이터 파이프라인은 다음과 같다. 각 데이터 소스에서 수집된 비구조화 데이터, 또는 스키마리스 데이터는 처음에는 분산 스토리지에 보존된다. 이는 SQL로 집계할 수 없다. 따라서 스키마를 명확하게 하는 구조화 데이터로 변환해야 한다.

 

일반적으로 구조화 데이터는 압축율을 높이기 위해 열 지향 스토리지로 저장한다. 이 중 시간에 따라 증가하는 데이터를 팩트 테이블, 그에 따른 부속 데이터를 디멘전 테이블로 취급한다.

열 지향 스토리지의 작성

MPP 데이터베이스의 경우, 제품에 따라 스토리지의 형식이 고정되어 있어 사용자가 그 상세를 몰라도 괜찮지만, Hadoop에서는 사용자가 직접 열 지향 스토리지의 형식을 선택하고, 자신이 좋아하는 쿼리 엔진에서 그것을 집계할 수 있다.

 

Hadoop에서 사용할 수 있는 열 지향 스토리지에는 몇 가지 종류가 있다. Apache ORC는 구조화 데이터를 위한 열 지향 스토리지로 처음에 스키마를 정한 후 데이터를 저장한다. 반면 Apache Parquet은 스키마리스에 가까운 구조로 되어 있어 JSON 같은 뒤얽힌 데이터도 그대로 저장할 수 있다.

 

비구조화 데이터를 읽어들여 열 지향 스토리지로 변환하는 과정에서는 컴퓨터 리소스가 많이 소비된다.

그래서 사용되는 것이 Hadoop과 Spark와 같은 분산 처리 프레임워크이다.

Hadoop: 분산 데이터 처리의 공통 플랫폼

Hadoop은 단일 소프트웨어가 아니라 분산 시스템을 구성하는 다수의 소프트웨어로 이루어진 집합체이다.

분산 시스템의 기본 구성 요소: HDFS, YARN, MapReduce

Hadoop의 기본 구성 요소는 분산 파일 시스템인 "HDFS", 리소스 관리자 "YARN", 그리고 분산 데이터 처리의 기반인 "MapReduce" 이 3가지이다. 이는 Hadoop의 기반이라고 얘기할 수도 있다.

위처럼 Hadoop에 의존하는 것이 아니라, 일부만 사용하거나 혹은 전혀 이용하지 않는 구성도 있다. 예를 들어, 분산 파일 시스템으로는 HDFS, 리소스 관리자는 Mesos, 분산 데이터 처리에는 Spark를 사용하는 구성도 가능하다.

분산 파일 시스템과 리소스 관리자: HDFS, YARN

Hadoop에서 처리되는 데이터 대부분은 분산 파일 시스템인 HDFS에 저장된다. 이것은 네트워크에 연결된 파일 서버와 같은 존재이지만, 다수의 컴퓨터에 파일을 복사하여 중복성을 높인다는 특징이 있다.(레플리카)

 

YARN(Yet Another Resource Negotiator)은 CPU나 메모리 등의 계산 리소스 매니저 역할을 한다. YARN은 애플리케이션이 사용하는 CPU 코어와 메모리를 컨테이너라고 불리는 단위로 관리한다. 리소스 매니저는 어느 애플리케이션에 얼마만큼의 리소스를 할당할 지 관리함으로써 모든 애플리케이션이 차질없이 실행되도록 제어한다.

분산 데이터 처리 및 쿼리 엔진: MapReduce, Hive

MapReduce는 YARN 상에서 동작하는 분산 애플리케이션 중 하나이며, 분산 시스템에서 데이터 처리를 실행하는 데 사용된다. 이는 임의의 자바 프로그램을 실행시킬 수 있기 때문에 비구조화 데이터를 가공하는 데 적합하다.

 

한편, SQL 등의 쿼리 언어에 의한 데이터 집계가 목적이라면 Hive를 이용한다. 이는 쿼리를 자동으로 MapReduce 프로그램으로 변환하는 소프트웨어로 개발되었다.

 

MapReduce는 대량의 데이터를 배치 처리하기 위한 시스템이므로, 작은 프로그램을 실행하려면 오버헤드(어떤 처리를 하기 위해 들어가는 간접적인 처리 시간, 메모리 등을 말한다)가 너무 크기 때문에 몇 초 밖에 안 걸리는 쿼리 실행에는 적합하지 않다. Hive도 MapReduce의 성질을 계승하고 있기 때문에 마찬가지이다. 애드 혹 쿼리에는 부적합하다.

Hive on Tez

Hive를 가속화하기 위해 개발된 것이 Tez이다. Tez는 기존의 MapReduce를 대체할 목적으로 개발되어 고속화를 실현하고 있다. 현재의 Hive는 Tez를 사용해도 동작하게 구성되어 있어 Hive on Tez라고 불린다.

대화형 쿼리 엔진: Impala, Presto

Hive를 고속화하는 것이 아니라 처음부터 대화형의 쿼리 실행만 전문으로 하는 쿼리 엔진도 개발되고 있다. Impala와 Presto가 대표적이다. 대화형 쿼리 엔진은 순간 최대 속도를 높이기 위해 모든 오버헤드가 제거되어 사용할 수 있는 리소스를 최대한 활용하여 쿼리를 실행한다.

 

Hadoop에서는 이와 같이 성질이 다른 쿼리 엔진을 목적에 따라 구분한다. 대량의 비구조화 데이터를 가공하는 무거운 배치 처리에는 높은 처리량으로 리소스를 활용할 수 있는 Hive를, 그렇게 해서 완성한 구조화 데이터를 대화식으로 집계하고자 할 때는 Impala와 Presto를 사용한다.

Spark

Spark는 MapReduce보다 더 효율적인 데이터 처리를 실현하기 위한 도구이다. 이는 Tez와 달리 Hadoop과는 다른 독립된 프로젝트이다.

 

Spark의 특징은 대량의 메모리를 활용하여 고속화를 실현한다는 것이다. MapReduce는 데이터 처리의 대부분을 디스크의 읽고 쓰기에 사용하였다. 그러나 메모리의 양이 증가함에 따라, 디스크에서 읽고 쓰는 것이 아니라 가능한 많은 데이터를 메모리상에 올린 상태로 두어 디스크에는 아무것도 기록하지 않는다는 선택이 현실화되었다. 이 경우 컴퓨터가 비정산 종료하면 중간까지 처리한 중간 데이터는 사라지지만, 그때는 처리를 다시 시도해서 중간 데이터를 다시 만들면 된다는 것이 Spark의 기본 개념이다.

MapReduce를 대체하는 Spark의 입지

Spark는 Hadoop의 대체제가 아니라 MapReduce의 대체제이다. HDFS나 YARN 등은 Spark에서 그대로 사용할 수 있다. 반대로 Hadoop을 이용하지 않고 분산 스토리지로 Amazon S3, 분산 데이터베이스로 카산드라를 이용할 수도 있다.

 

또한 Spark는 자바, 스칼라, 파이썬, R 등 다양한 스크립트 언어를 사용하여 데이터 처리를 할 수 있고, 문서도 충실하기 때문에 도입이 쉽다.

 

그리고 SQL로 쿼리를 실행하기 위한 Spark SQL과 스트림 처리를 수행하기 위한 Spark Streaming이라는 기능이 포함되어 있다. 따라서 대규모 배치 처리뿐만 아니라, SQL에 의한 대화형 쿼리 실행과 실시간 스트림 처리에 이르기까지 널리 이용되고 있다.

쿼리 엔진

데이터 마트 구축의 파이프라인

Hadoop에 의한 구조화 데이터의 작성과 이를 이용한 쿼리의 실행이 어떤 것인지를 알기 위해,

쿼리 엔진을 사용하여 데이터 마트를 만들기까지의 흐름을 살펴보겠다.

 

처음에 분산 스토리지에 저장된 데이터를 구조화하고 열 지향 스토리지 형식으로 저장한다. 이는 Hive를 이용한다.

그리고 완성한 구조화 데이터를 결합, 집계하고 비정규화 테이블로 데이터 마트에 써서 내보낸다. 열 지향 스토리지를 이용한 쿼리의 실행에는 Presto를 사용하여 실행 시간을 단축할 수 있다.

Hive에 의한 구조화 데이터 작성

Hive는 외부 테이블을 생성함으로써 외부에 있는 특정 파일을 참고해 마치 거기에 테이블이 존재하는 것처럼 읽어 들일 수 있다.

CREATE EXTERNAL TABLE access_log_csv(
  time string, request string, status int , bytes int
)

Hive를 비롯한 대부분의 SQL-on-Hadoop의 쿼리 엔진은 MPP 데이터베이스처럼 데이터를 내부로 가져오지 않아도 텍스트 파일을 그대로 집계할 수 있다.

SELECT status, count(*) cnt FROM access_log_csv GROUP BY status LIMIT 2;

만약 위와 같이 쿼리를 실행하면 외부 테이블로 지정한 경로에 포함된 모든 CSV 파일이 로드되고 집계된다.

이렇게 하면 애드 혹 데이터를 분석하기에 유용하다. 하지만 아무래도 텍스트 파일을 읽어들이기 때문에 빠르지 않다. 이렇게 된다면 열 지향 스토리지로 변환한다.

열 지향 스토리지로의 변환: 데이터 집계의 고속화

CREATE TABLE access_log_org STORED AS ORC AS  
SELECT cast(time AS timestamp) time, 
       request, 
       status, 
       cast(bytes AS bigint) bytes  
FROM access_log_csv;

테이블을 열 지향 스토리지 형식인 ORC 형식으로 변환한다. ORC 형식으로의 변환에는 다소 시간이 걸리지만, 변환 후 테이블 집계는 엄청나게 단축된다. 하지만 시간이 걸리는 프로세스이므로, Hive와 같은 배치형의 쿼리 엔진에서 실행하는 데 적합하다.

Hive로 비정규화 테이블 작성하기

데이터 구조화가 완료되면 데이터 마트 구축을 해야한다. 즉, 테이블을 결합 및 집약해서 비정규화 테이블을 만든다. 이때 Presto 같은 대화형 쿼리 엔진을 사용할 것인지, Hive 같은 배치형 쿼리 엔진을 사용할 것인지에 따라 달라진다.

 

비정규화 테이블을 만드는 데 오랜 시간이 걸리는 것은 흔하기 때문에, 가능한 효율적인 쿼리를 작성해야 한다. 여기에는 서브 쿼리 안에서 레코드 수를 줄이는 방법과 데이터의 편향을 방지하는 방법이 있다.

 

① 서브 쿼리 안에서 레코드 수를 줄이는 방법은 초기에 팩트 테이블을 작게 하는 것이다. 팩트 테이블을 서브 쿼리를 통해 최대한 필터링하여 효율화하는 것이 중요하다.

② 분산 시스템의 성능을 발휘하기 위해서는 데이터의 편차를 최대한 없애고, 모든 노드에 데이터가 균등하게 분산되도록 해야한다. 이것도 최초에 중복을 제거함으로써 부하를 잘 분산시켜 데이터의 양을 줄이는 것으로 해결할 수 있다.

대화형 쿼리 엔진 Presto의 구조

Hive는 대규모 데이터 처리에는 적합하지만, 작은 쿼리를 여러 번 실행하는 대화형 데이터 처리에는 적합하지 않다. 이에 적합한 것이 대화형 쿼리 엔진이다.

플러그인 가능한 스토리지

Presto의 특징 중 하나는 플러그인 가능한 스토리지 설계이다. Presto는 전용 스토리지를 가지고 있지 않아, 다양한 데이터 소스에서 직접 데이터를 읽어 들인다.

 

Presto는 Hive 메타 스토어에 등록된 테이블을 가져올 수 있다. 따라서 Hive에서 만든 구조화 데이터를 좀 더 집계하는 등의 목적에 적합하다. Presto가 그 성능을 최대한 발휘하려면 원래 스토리지가 열 지향 데이터 구조로 되어 있어야 한다.

CPU 처리의 최적화

Presto는 메모리와 CPU 리소스만 충분하다면 데이터의 읽기 속도가 쿼리의 실행 시간을 결정하게 된다. Presto는 리소스 관리자를 사용할 수 없기 때문에(스케줄링 개발이 진행되고 있다고 한다), 너무 큰 쿼리를 실행해서는 안된다.

인 메모리 처리에 의한 고속화

Hive와 달리 Presto는 쿼리의 실행 과정에서 디스크에 쓰기를 하지 않는다. 모든 데이터 처리를 메모리상에서 실시하고 메모리가 부족하면 여유가 생길 때까지 기다리거나 오류로 실패한다. 이 경우는 메모리 할당을 늘리거나, 쿼리를 재작성해서 메모리 소비를 줄여야 한다.

 

메모리상에서 할 수 있는 것은 메모리에서 실행하고, 대규모 배치 처리와 거대한 테이블끼리의 결합 등에는 디스크를 활용해야 한다. 따라서 Presto는 단시간 쿼리 실행과 같은 대화형 쿼리 엔진로써 사용한다고 할 수 있다.

분산 결합과 브로드캐스트 결합

테이블 결합은(특히 팩트 테이블) 대량의 메모리를 소비한다. Presto는 기본적으로 분산 결합을 실시하며,

같은 키를 갖는 데이터는 동일한 노드에 모인다.

 

분산 결합에서는 노드 간의 데이터 전송을 위한 네트워크 통신이 발생하기 때문에 쿼리의 지연을 초래한다.

따라서 한쪽 테이블이 충분히 작은 경우에는 브로드캐스트 결합을 사용하여 속도를 높인다.

열 지향 스토리지 집계

Presto에서는 열 지향 스토리지의 집계를 매우 빠르게 실행할 수 있다. 실제로 수백만 레코드 정도를 집계해보면 1초 미만으로 매우 빠름을 알 수 있다.

SELECT status, count(*) cnt
FROM access_log_orc GROUP BY status LIMIT 2;

status |  cnt
-------+--------
200    | 1701534
302    |   46573

데이터 분석 프레임워크 선택하기

실제 운용에서는 아마 클라우드를 이용하여 시스템을 구축하는 경우가 많을 것이다. 많은 옵션 중 무엇을 선택해야 할까?

MPP 데이터베이스: 완성한 비정규화 테이블의 고속 집계

구조화 데이터를 SQL로 집계만 한다면 기존의 데이터 웨어하우스와 클라우드를 이용하는 것이 좋다. MPP 데이터베이스는 처음에 ETL로 데이터를 가져오는 절차가 필요하다. 그 부분만 완성하면, 그 다음은 SQL만으로 집계할 수 있다.

 

확장성, 유연성 등의 측면에서는 분산 시스템이 유리하다. 대량의 텍스트나, 데이터 처리를 프로그래밍하고 싶은 경우, 또는 NoSQL에 저장된 데이터를 집계하고 싶은 경우는 분산 시스템의 프레임워크를 결합해야 한다.

 

하지만 시각화를 위한 데이터 마트를 생각하면, MPP 데이터베이스와 BI 도구를 조합하여 비정규화 테이블을 고속으로 집계하는 데 최적이다.

Hive: 데이터양에 좌우되지 않는 쿼리 엔진

Hive는 대규모 배치 처리를 꾸준히 실행하는데 유용하다. 텍스트 데이터를 가공하거나 열 지향 스토리지를 만드는 등의 무거운 처리에 적합하다.

Presto: 대화식으로 특화된 쿼리 엔진

Presto는 Hive와는 반대로 속도가 빠른 대신 안정성은 낮다. 대화식 쿼리의 실행에 특화되어 있기 때문에 텍스트 처리가 중심이 되는 ETL이나 데이터 구조화에는 적합하지 않다. 단시간에 대량의 리소스를 소비하기 때문에 대규모 배치 처리는 Hive에게 맡기거나 클러스터를 나누는 등으로 여유를 갖게 하는 것이 좋다.

Spark: 분산 시스템을 사용한 프로그래밍 환경

Spark는 SQL에 특화된 쿼리 엔진은 아니지만, 인 메모리 데이터 처리가 중심이며, ETL에서 SQL까지 일련의 흐름을 하나의 데이터 파이프라인으로 기술할 수 있다. Hive의 데이터 구조화와 Presto의 SQL 실행을 동시에 하나의 스크립트에서 실행할 수 있다는 것이다.

 

Spark에서는 메모리 관리가 중요하므로 자주 사용하는 데이터는 캐시에 올려놓거나, 디스크에 스왑시킴으로써 메모리를 해제하는 등 프로그래머가 잘 제어해야 한다.

 

데이터 처리를 위한 실행 환경을 구성하고 싶다면 Spark가 가장 최적이라고 할 수 있다.

데이터 마트 구축

팩트 테이블

빅데이터 분석은 데이터를 구조화하는 것부터 시작하는데, 그 중 많은 부분을 차지하는 것이 팩트 테이블이다. 팩트 테이블이 크면 열 지향 스토리지에서 데이터를 압축해야 빠른 집계를 할 수 있다.

 

테이블 작성에는 추가와 치환이 있다. 추가는 새로 도착한 데이터만을 증분으로 추가하는 반면, 치환은 과거의 데이터를 포함하여 테이블 전체를 치환한다.

테이블 파티셔닝

효율만을 생각하면 추가가 압도적으로 유리하지만, 여러가지 잠재적 문제가 있다. 이를 줄이기 위해 테이블 파티셔닝을 사용한다. 이는 하나의 테이블을 여러 물리적인 파티션으로 나눔으로써 파티션 단위로 정리하여 데이터를 쓰거나 삭제할 수 있도록 한 것이다.

 

1일 1회 혹은 1시간 1회라는 식으로 새 파티션을 만들고 그것을 매번 교체하거나 덮어써 데이터가 중복될 가능성을 배제한다.

데이터 마트의 치환

테이블 파티셔닝은 DW를 구축하는데 유용하다. 팩트 테이블 전체를 치환하는 것은 많은 장점이 있다. 데이터가 중복되거나 유실될 가능성이 거의 없고, 스키마 변경이나 데이터 마트 무제한적 확대도 없다. 유일하게 우려되는 것은 처리 시간인데, MPP 데이터베이스라면 쓰기를 병렬화하여 속도를 높일 수도 있고 그래도 오래 걸린다면 주의 깊게 모니터링해야 한다.

집계 테이블

팩트 테이블을 어느 정도 모아서 집계하면 데이터의 양이 크게 줄어든다. 이를 집계 테이블이라고 한다. 특히 데이터를 1일 단위로 집계한 일일 집계를 잘 만들면 데이터 마트가 그다지 커지지 않는다. 집계 테이블을 작게 하려면 카디널리티를 줄여야 한다.

스냅샷 테이블: 마스터의 상태를 기록하기

마스터 테이블처럼 업데이트될 가능성이 있는 데이블에 대해서는 두 가지 방안이 있다. 정기적으로 테이블을 통째로 저장하는 스냅샷 테이블과 변경 내용만을 저장하는 이력 테이블이 있다.

이력 테이블: 마스터의 변화 기록하기

정기적으로 모든 데이터를 스냅샷 하는 것이 아니라 변경된 데이터만을 증분으로 스냅샷 하거나 변경이 있을 때마다 그 내용을 기록하는 이력 테이블이 만들어지는 경우도 있다.

디멘젼을 추가하여 비정규화 테이블을 완성시키기

마지막은 팩트 테이블과 디멘전 테이블을 결합하여 비정규화 테이블을 만든다. 디멘전 테이블로는 스냅샷이나 각종 중간 테이블을 사용한다.

데이터 집계의 기본형

데이터를 집계하는 전형적인 쿼리를 정리하면 다음과 같다. 먼저 팩트 테이블에서 필요한 데이터를 꺼낸다. 그리고 그를 디멘전 테이블과 결합하여 데이터 마트에 저장할 칼럼을 선택한다. 이 두 과정에서 중요한 것은 최대한 카디널리티를 작게하는 것이다. 시각화 프로세스에서 이용하고 싶은 디멘전만을 추가하도록 한다.

마지막으로 그룹화하여 측정값을 집계한다. 충분히 작은 비정규화 테이블이 만들어졌다면 그 결과를 데이터 마트나 CSV 파일로 내보내면 완료된다.