빅데이터를 지탱하는 기술 CHAPTER 6 : 빅데이터 분석 기반의 구축 (6-2)

6장에서는 몇 가지 오픈 소스 소프트웨어를 사용하여 실제로 데이터를 처리하는 예제 코드를 실행한다.

 

6.2절에서는 6.1절과 같은 것을 'Hive'와 'Presto'를 사용하여 재구축한다. 하나하나의 데이터 처리를 멱등한 태스크로 구현함으로써 나중에 워크플로에 포함할 수 있도록 한다.

 

Hadoop에 의한 데이터 파이프라인

ETL 프로세스로부터 데이터 마트의 작성에 이르는 '데이터 파이프라인'은 매일매일의 데이터 처리의 중심이 되는 프로세스다. 이 절에서는 Hive와 Presto를 사용한 배치형의 데이터 처리에 관해서 설명한다.

 

6.2절에서는 다음의 소프트웨어에 대해 설명한다.

벌크 형 데이터 전송 - Embulk 0.8.17

분산 시스템 - Hadoop 2.8.0

데이터 구조화 - Hive 2.1.1

쿼리 엔진 - Presto 0.181

 

일일 배치 처리를 태스크화하기

정기적으로 데이터를 전송하고 그것을 집계한 후 데이터 마트를 만드는 전형적인 데이터 파이프라인을 고려한다.

데이터 소스는 MongoDB를 이용한다. 장기적인 데이터 분석을 위해 Hive로 열 지향 스토리지를 만들고, 그것을 Presto로 집계한다. 큰 흐름으로는 밑의 그림과 같이 3개의 태스크로 구성한다.

데이터양이 늘어도 집계 효율이 떨어지지 않도록 1일 단위로 파티셔닝된 시계열 테이블을 만든다.

[태스크 1] Embulk에 의한 데이터 추출

MongoDB로부터 데이터를 추출하기 위해 여기서는 오픈 소스의 벌크 전송 도구인 'Embulk'를 사용한다.

실제로는 Embulk를 사용하지 않아도 MongoDB 전용의 커맨드라인 도구를 사용하거나, Hive로부터 MongoDB에 직접 접속해도 상관없지만, 여기서는 어디까지나 한 예로 범용적인 데이터 전송의 절차를 고려한다.

 

Embulk를 설치하고, 밑의 쉘 스크립트를 실핸한다. 여기서는 추출한 데이터를 JSON 형식으로 로컬 디스크에 쓰고 있다.

벌크 형의 데이터 전송은 정기적으로 실행되므로, 태스크의 파라미터로 시간의 범위를 부여하도록 한다.

여기서는 명령어의 인수로 2000-01-01부터 2100-01-01의 범위를 건네고 있는데, 실제로는 1일씩의 범위로 실행한다.

이 파라미터는 나중에 워크플로 관리 도구에서 설정한다.

 

지정된 기간의 데이터를 추출하고, 그것을 지정된 장소로 써냄으로써 이 태스크의 실행은 멱등하게 된다.

실제로 스크립트를 실행해보면 파라미터만 같다면 몇 번을 실행해도 결과가 덮어쓰여지는 것을 알 수 있다.

워크플로에서는 되도록 이렇게 각 태스크를 멱등하게 구현하도록 주의한다.

 

여기서는 추출한 데이터를 로컬 디스크에 써넣고 있는데, 실제 환경에서는 별도의 분산 스토리지에서 전송하거나 Hive로부터 안정적으로 읽어낸다.

 

[태스크 2] Hive에 의한 데이터 구조화

다음으로 Hive를 셋업한다. 참고로 초기 설정에는 로컬 디스크의 /user/hive/warehouse에 Hive의 테이블이 만들어지므로, 미리 디렉토리를 작성해둔다. 나중에 Presto로 집계할 수 있도록 Hive 메타 스토어 서비스를 가동한 채로 둔다.

 

태스크 1에서 작성한 JSON 파일을 변환해보자. Hive로 시계열 테이블을 만들려면, 밑의 리스트 6.5처럼 파티셔닝을 유효로 한 테이블을 만든다. 그리고 INSERT OVERWRITE TABLE 문으로 파티션을 덮어씀으로써 태스크가 멱등하게 된다.

쿼리에 건넬 파라미터(여기서는 'START')를 바꿈으로써 매일 새로운 파티션이 만들어져 그에 따라 시간으로 파티션이 분할된 시계열 테이블이 완성된다. 출력 테이블의 'time' 컬럼에는 Twitter의 타임 스템프(timestamp_ms)를 보관하여 마지막으로 'ORDER BY'로 정렬한다. 이것이 이벤트 시간에 의해 데이터가 정렬되어서 조건절(Predicate) 푸쉬다운에 의한 최적화를 기대할 수 있다.

 

[태스크 3] Presto에 의한 데이터 집계

마지막으로 Presto를 셋업한다. Presto는 서버/클라이언트 형의 시스템이므로 처음에는 Presto 서버를 가동해 둔다.

Hive 메타 스토어 서비스가 동작하고 있다면, Presto에서 Hive로 작성한 테이블로 접속할 수 있다.

 

설정에 문제가 없다면, 다음과 같이 해서 Presto의 쿼리를 실행할 수 있다. 1대의 컴퓨터로도 1,000만 건이 넘는 테이블을 수초 만에 집계할 수 있다.

Presto는 애드 혹 데이터 분석에서 사용되는 대화식 쿼리 엔진을 워크플로 안에서 호출해도 문제없다.

이 6.1절에서 Spark로 실행한 것과 같은 쿼리를 Presto용으로 고치면 리스트 6.7과 같다.

Presto에 의한 데이터 집계는 열 지향 스토리지를 만드는 것과 비교해 훨씬 고속이므로, 장기간의 집계를 다시 실행해도 그다지 시간이 걸리지 않는다. 따라서, 증분으로 데이터를 집계하는 것이 아니라, 매일 데이터 마트를 다시 만드는 편이 간단하다. 예를 들어 과거의 30일간의 추이를 보고 싶다면, 매일 30일만큼의 데이터를 읽어 들여 결과를 치환함으로써 태스크가 멱등이 된다.

 

리스트 6.7에서는 CSV 파일을 만드는 부분에서 태스크를 종료하고 있는데, 실제로는 그것을 분산 스토리지에 전송하거나 BI 도구에 읽어 들임으로써 최종적인 데이터 파이프라인이 완성된다. 이것으로 이후에는 파라미터를 교체해서 매일 실행하도록 하면 자동화가 완성된다.