[Infra] 데이터 인프라 - Ingestion&Transformation(Spark Platform, Python Lib, Batch Query Engine)

🔊 해당 포스팅은 YouTube GeekNews 채널의 데이터 인프라 강의 내용, 하용호님의 SlideShare 슬라이드에 기반해 제가 재구성하여 작성함을 알려드립니다.

 

이번 포스팅에서는 저번 포스팅에서 살펴본 Ingestion & Transformation 단계의 또 다른 도구인 Apache Spark Platform에 대해 알아보려고 한다. PySpark에 대해 공식 문서를 읽고 어느정도 공부한 내용을 저번에 포스팅한 적이 있다. 그런데 역시 공식 문서만으로 이해한 것이 모든 걸 내가 다 이해한 것이 아니었다.. 그래서 2가지 매체를 통해 더 공부를 하고 정리한 내용을 포스팅하려 한다.

 

데이터 인프라의 큰 구조

 

위 빨간색 네모칸의 요소들을 각각 소개하기 전에 이 요소들인 Spark Platform, Python Libs, Batch Query Engine은 파이프라인 순서상으로 이전 포스팅 마지막에 배웠던 Workflow Manager보다 앞선 위치에 존재한다. 왜냐하면 앞으로 소개할 Spark Platform, Python Libs, Batch Query Engine은 데이터 분석을 코드로 짜고 분석해보기 위해 사용되는 것이고 Workflow Manager들이 앞서 분석하기 위해 작성한 코드들을 실질적인 task로서 실행, 수행하는 것이기 때문이다. 따라서 본 포스팅에서 소개될 컨텐츠들은 Workflow Manager 다음으로 게시되지만 실제 파이프라인 순서는 [Workflow Manager ➡️ Spark Platform, Python Libs, Batch Query Engine] 임을 명심하자.

---

Spark Platform

아파치 스파크는 대용량의 데이터 또는 task들을 여러 대의 서버에 분산처리를 하면서 빠른 처리를 가능하도록 해주는 프레임워크를 의미한다. Spark가 출시되기 이전에 그나마 분산처리를 적용할 수 있었던 것이 Hadoop의 MapReduce이다. 우선 이 둘의 차이점부터 표로 살펴보자.

 

Spark VS MapReduce

 

읽어보면 알겠지만 속도적인 측면에서 Spark가 압도적으로 빠르다. 또한 데이터 처리 측면에서 맵리듀스는 Batch Processing(task들을 묶어서 여러 대의 서버로 할당하고 처리한 후 다시 합치는 과정)만을 제공하는 반면 스파크는 실시간(real-time), 그래프(graph) 등 여러가지 기능을 제공한다. 또 편의상으로 사용하기 쉽다고 한다. 마지막으로 가장 중요한 특징이 데이터를 메모리에 캐싱하는 스파크의 특징인데, 이것은 스파크가 어떤 식으로 동작하는지 이해하기 위한 근간이 된다. 이에 대한 자세한 내용은 아래에서 살펴보자.

1. MapReduce는 뭐가 문제였을까?

스파크의 RDD에 대해 이해하기 위해서는 하용호님의 슬라이드를 매우 참고했다. 이해하기 너무 쉽게 설명해주셔서 정말 감사하다! 스파크를 이해하기 전에 맵리듀스가 갖는 한계점부터 살펴보자.

 

• 머신러닝, 그래프와 같은 복잡하고 Multi-Stage한 처리를 잘 하지 못함
• Interactive한 쿼리와 ad-hoc한 쿼리를 잘 처리하지 못함

여기서 필자는 Interactive, ad-hoc한 쿼리에 대한 정의를 잘 몰라서 찾아보고 이해한 내용을 같이 소개하려 한다.

 

• Interactive한 쿼리 : Stream Processing을 위한 새로운 쿼리 형태. Stream Processing이란, 병렬처리를 어플리케이션이 쉽게 이용할 수 있도록 해주는 프로그래밍을 의미한다.
  • Interactive한 쿼리는 Stream Processing을 가볍게 해준다. 즉, Stream Processing을 하도록 쿼리를 하는 것인데, 이 때 외부 DB나 스토리지로 구체화할 필요 없이 embedded된 DB처럼 빠르고 효율적으로 가볍게 처리하는 것이다.
• ad-hoc한 쿼리 : 비정형적인 쿼리를 말한다. 즉석에서 또는 단발성으로 필요한 쿼리를 의미한다.

또한 맵리듀스는 iteration 과정을 HDFS라는 파일시스템에 넣었다 뺐다가를 반복하기 때문에 느려지게 된다. 이 때 사람들은 그러면 최근에 상대적으로 저렴해진 RAM(RAM이란, 주기억장치로서 보조기억장치에 있는 데이터를 가져와서 CPU가 작업할 수 있게 만들어주는 공간을 의미)을 이용하기로 했다.

2. RAM을 어떻게 사용할까?

그렇다면 RAM을 왜 사용할까? 바로 RAM에 한 번 저장해놓고 저장한 것이 필요할 때마다 빼서 쓰는 원리를 이용할 것이기 때문이다. 이는 얼마 전 알고리즘 공부를 하면서 Dynamic Programming 원리와 매우 비슷하다고 느꼈다. 하지만 RAM 자체를 사용하다가 Fault(고장이나 에러)가 발생했을 땐 문제가 발생한다. 

 

만약 하나가 고장이 났을 때 다른 것들이라도 살리기 위해서 복사를 해놓은 Replicating을 하게되면 이 Replicating을 하는 동안에는 데이터 처리를 멈추어야만 하고 결국 속도가 저하되는 문제가 발생한다. 또, Replicating만으로는 위함할 것아 데이터를 디스크에 저장하는 Checkpointing을 한다면 속도가 매우 느려지는 문제가 발생한다. 따라서 사람들은 Fault에 잘 견디고 효율적인 이른바 Fault-tolerant & Efficient 한 RAM 스토리지를 어떻게 만들지 계속적으로 고민했다.

3. RAM인데... 읽기만 가능한 RAM?

제목 그대로다. RAM이긴 한데 Read-only, 즉, 수정이 불가하고 읽기만 가능한 RAM을 만드는 것이다. 이렇게 읽기만 가능한 RAM인 스파크의 RDD(Resilient Distributed Dataset)가 탄생한다.

4. Spark = RDD + Interface (Interface는 거들 뿐..)

참조 슬라이드에서 하용호님은 스파크와 RDD를 거의 동일시하면서 RDD에 대한 이해의 중요성을 강조하신다. 그러 RDD의 특징들부터 알아보자.

 

• 수정이 불가(Immutable), 파티션된 레코드들의 집합(Partitioned collections of records)
• 스토리지에서 RDD로 또는 RDD에서 RDD로만 변환이 가능

위에서 살펴보았지만 RDD의 특성은 수정이 불가한 즉, 읽기만 가능한 Read-only 성격이다. 그러면 데이터를 수정할 수가 없는 걸까? 아니다. RDD는 변형되는 모습들을 계속 기록해서 계보(Lineage)를 만들어서 저장한다. 이렇게 함으로써 Fault-tolerant 가 가능해지게 된다. RDD가 계보(Lineage)를 만든다는 것이 어떤 의미일까? 하단의 그림을 참고해보자.

 

출처 : 하용호님 Slideshare

 

어디서 본 적이 있지 않은가? 바로 Workflow Manager에서 살펴보았던 DAG(Directed Acyclic Graph)이다. 즉 위 그림처럼 특정 데이터에 대해 Map한 결과, Union한 결과, Reduce한 결과 등등을 모두 계속적으로 기록해서 일종의 계보(Lineage)를 만드는 것이다. 물론 이도 DAG 구조이기 때문에 자기 자신으로 돌아오는 순환 Edge는 절대 없다.

5. RDD의 2가지 Operator

RDD에는 2가지 오퍼레이터가 존재한다. 바로 Transformations와 Actions이다. 이에 대해선 예전에 PySpark 포스팅할 때도 잠깐 소개했던 기억이 난다. 다시 개념에 대해 알아보자.

 

• Transformations : 어떤 '변환' 또는 어떤 '변환들'을 취할지 계속 준비해놓는다. 여기서 계보(Lineage)가 생성된다.
• Actions : 여러개의 Transformations로 구성된 Lineage를 실실적으로 수행하는 것을 의미한다.

Transformations 와 Actions 간의 관계를 도식화해보면 다음과 같다고 할 수 있다.

 

Transformations 와 Actions의 관계

 

위 도식화를 보면 알겠지만 Actions를 선언하기 전까지 모든 Transformation들이 대기(?)상태로 되어 있다. 이러한 특성으로 인해 스파크가 Lazy-Execution 특성이 있다고 한다. 이는 자원이 배치되고 또 배치될 상황을 미리 고려해서 최적의 루트로 갈 수 있도록 해준다. 다시 말해, 아무리 복잡한 일을 많이 시켜도 안 하고 아껴두다가 마지막에 최적의 방법을 알아서 찾아서 실행하는 장점이 된다.

6. RDD의 Type Dependency

RDD의 Transformations Operator들에는 map, filter, reduce, groupbykey, join 등 여러가지가 있다. 사용되는 Transformation에 따라 Type Dependency가 Narrow, Wide로 나누어진다.

 

Transformation Operato 종류에 따른 Dependency type

 

그림 속에서는 Dependency가 Transformation으로 되어 있는데 동일한 의미로 해석하면 될 듯 하다. 그러면 이제 두 개의 차이점에 대해 살펴보자. 큰 특징은 특정 Transformation을 수행하면서 데이터의 Shuffle이 이루어지는 아닌지의 차이이다.

 

• Narrow Dependency : 현재 서버(스파크 상에서 '노드'라고 불림)에서 처리할 수 있는 모든 일은 다 처리한다. Shuffle을 사용하지 않고 메모리의 속도로 동작하기 때문에 매우 빠르다. 
• Wide Dependency :  모든 서버(스파크 상에서 '노드'라고 불림)에서 처리할 수 있는 일을 다 가져와 처리한다. Shuffle을 사용하고 네트워크의 속도로 동작하기 때문에 느리다. 이러한 Transformation들은 추후에 디스크에 따로 저장하는 Checkpointing을 할 필요가 있다고 한다.

7. Spark가 돌아가는 과정

이제 RDD에 대해 알아보았으니 Spark를 좀 안 듯(?)한 느낌이 난다. 이제 Spark가 클러스터라는 것을 생성하고 내부적으로 어떻게 동작하는지 도식화, 설명 자료를 통해 살펴보자.

 

Spark가 돌아가는 과정

 

그런데 어쨌거나 스파크가 RAM을 사용해서 돌아간다고 했다. RAM도 메모리이기 때문에 메모리 부족 문제가 발생할 수 있다. 이럴 땐 어떻게 해결할까? 스파크는 이러한 문제가 발생했을 때 LRU(Least Recently Used) 알고리즘, 즉, 가장 오랫동안 참조되지 않은 페이지를 교체하는 알고리즘을 사용해 특정 메모리를 삭제하게 된다. 잠깐 LRU에 관해서 나와서 다른 페이지 교체 알고리즘을 잠깐만 설명하고 넘어가겠다. 이에 대해 아시는 분들은 다음 내용으로 넘어가도 좋을 것 같다.

 

• FIFO(First-In-First-Out): 큐와 동일한 원리인데, 페이지가 주기억장치에 적재된 시간을 기준으로 교체될 페이지를 선정한다. 즉, 가장 오래된 시간인 페이지가 가장 우선적인 삭제 대상이다. 
  • 단점 : 가장 오래된 페이지가 계속 사용되어야 한다면 문제가 발생한다.
• LFU(Least Frequently Used): 가장 적은 횟수를 참조하는 페이지를 교체
  • 단점 : 단순히 '횟수'에 기반하므로 최근에 사용된 페이지를 교체시킬 가능성이 있고 해당 횟수를 증가시키면서 오버헤드가 발생한다.
• LRU(Least Recently Used): 가장 오랫동안 참조되지 않은 페이지를 교체
  • 단점 : 프로세스가 주기억장치에 접근할 때마다 참조된 페이지에 대한 시간을 기록해 큰 오버헤드가 발생한다.

스파크의 중요한 특징으로 고장에 잘 견디는 Fault-tolerant라고 했다. 그런데 정말 Fault가 발생했을 때, Recovery는 어떻게 할까? 스파크에서 사용되는 Lineage는 용량이 작기 때문에 부담도 없고 잘 로깅이 되며 특정 파티션(분할된 데이터)에 문제가 생기면 다른 노드(서버)를 가져와 대신 실행하게 된다.

8. Spark를 제공하는 플랫폼

스파크도 직접 다루기보다 솔루션으로서 편하게 만들어주는 플랫폼이 역시 존재한다. 스파크를 만든 사람들이 창업한 DataBricks가 있다. 보통 DataBricks는 작은 인스턴스일 때 주로 사용된다고 한다. 또 다른 플랫폼으로 AWS EMR(Elastic MapReduce)이 있다. 이는 DataBricks와 비슷한 기능을 제공하지만 보통 큰 인스턴스일 때 주로 사용된다고 한다.

---

Python Libs

다음은 Spark Platform 이외의 여러가지 Python Library에 대해 소개하려 한다.

1. Pandas

'파이썬으로 데이터 분석?' 하면 당연히 누구나 들어봤을 법한 라이브러리다. 주로 2차원 Tabular data로 되어 있는 데이터프레임을 다룬다.

2. Boto

Python으로 AWS에 접근을 가능하게 해주는 SDK(Software Developer Kit)이다.

3.DASK

Python으로 병렬처리를 가능하게 해준다. Pandas, Numpy, Scikit-learn API를 모두 병렬처리가 가능하도록 해준다. 주로 방대한 데이터프레임을 분할하거나 전처리를 하는 등과 같이 빅데이터를 분산처리하는 데 자주 사용된다. 하단에서 소개할 Ray와는 다르게 Centralized 스케쥴링을 사용한다.

4.Ray

DASK와 마찬가지로 파이썬으로 병렬처리를 가능하게 해준다. 주로 여러개의 머신러닝 모델을 분산처리할 때 사용된다. DASK와는 다르게 Bottom-Up 스케줄링을 사용한다. 여기서 Bottom-Up 스케줄링이란, task를 각 로컬 스케줄러에게 전달하고 각각의 로컬 스케줄러가 각 Worker들에게 다시 task를 assign, forward 하는 것을 의미한다. 그래서 Centralized 스케줄링을 사용하는 DASK에 비해 빠르고 task 지연(latency) 문제와 task throughput(task throughput이란, 단위 시간 당 완료되는 task의 숫자 즉, task 완료율(rate)를 의미한다)을 개선시키는 장점이 있다.

---

Batch Query Engine

Apache Hive는 Hadoop위에 올라와 있는 HDFS(Hadoop File System)의 데이터를 처리하기 위한 엔진이다. 그러면 Hive의 장점은 무엇일까? 하둡을 통해 데이터를 쿼리하려면 MapReduce(맵리듀스) 잡을 처리해주어야 한다. 하지만 Hive를 사용해서 SQL 쿼리를 작성한다면 맵리듀스 잡으로 자동으로 변환해주고 실행되고 있는 하둡의 데이터를 읽어와 처리해주는 편의성을 제공한다.

 

그런데 Spark Platform의 Spark SQL의 쿼리 기능이 있는데 굳이 Hive를 사용하는 이유는 뭘까? Hive와 Spark는 서로 상호보완 작용을 한다고 볼 수 있다. 우선 Hive가 Spark에게 제공하는 장점은 다음과 같다.

 

• Hive ➡️ Spark : Hive 덕분에 다양한 데이터에 접근이 가능

반면에, Spark가 Hive에게 주는 장점은 무엇일까? 우선 하단의 그림을 보자.

 

출처 : 긱뉴스님 유투브 강의영상 속 자료

 

• Spark ➡️ Hive : Hive 쿼리만을 사용해 Spark 쿼리의 효과를 볼 수 있다.

위 그림을 보다시피 Hive는 여러가지(Thrift, JDBC, ODBC) Driver, 즉, 여러가지 Application과 연결되어 있다. 따라서 이 여러가지 어플리케이션이 있다보니 이 어플리케이션이 갖는 DB들이 있다. 만약 Hive가 Spark와 연결된다면 Hive 실행 엔진으로 Spark를 사용하게 되고 단순히 Hive 쿼리만을 사용해 Hive와 연결되어 있는 어플리케이션들의 DB의 데이터들을 쉽게 처리하고 빠르게 사용이 가능해진다.