[Spark] 스파크 이해하기

2013년에 생산된 데이터는 약 4.4ZB로 추정됩니다. 이는 44억 테라바이트입니다. 2020년에는 그 10배에 달하는 데이터가 생산될 것으로 예측됩니다. 데이터는 매초마다 증가하고 있으며, 이런 상황에 비춰 2004년 구글의 제프리 딘과 산자 제마왓은 'MapReduce: Simplified Data Processing on Large Clusters'라는 세미 논문을 발표했습니다. 그때부터 아파치 하둡과 같이 맵리듀스를 사용하는 개념이 매우 유명해지기 시작했습니다. 피그 하이브 머하웃과 같은 추상 층을 포함하는 하둡 에코시스템도 개발되었습니다. 이 모든 것이 간단한 맵리듀스 개념을 사용합니다.

 

매일 수 페타바이트를 다룰수 있으에도 불구하고, 맵리듀스는 상당히 제한적인 프레임워크입니다. 또한 대부분의 태스크가 디스크 읽기/쓰기를 필요로 합니다. 이러한 단점을 극복하기 위해 2009년 마태 자하리아는 그의 박사 과정 중에 스파크를 개발하기 시작했으며, 2012년 처음으로 배포했습니다. 스파크는 기본적으로 똑같은 맵리듀스 개념을 사용하지만, 스파크만의 데이터 처리방법과 태스크 정리 방법으로 인해 메모리 내의 연산 속도가 하둡보다 100배 가량 빨라졌습니다.

 

아파치 스파크(Apache Spark)는 마태 자하리아가 UC 버클리에서 박사 과정 논문의 일부로 개발한 강력한 오픈소스 처리 엔진입니다. 2012년에 처음 배포되었으며,이 후 스파크의 핵심 코드를 아파치 소프트웨어 재단에 기부했으며, 마침내 스파크는 아파치 소프트웨어 프로젝트의 대표 프로젝트가 되었습니다.

 

스파크는 다양한 종류의 데이터 관련 문제, 예를 들어 반구조(semi-structured), 구조, 스트리밍 또는 머신 러닝/데이터 과학 관련 문제를 해결하기 위해 쉽고 빠르게 쓸 수 있는 프레임워크입니다. 이번 포스팅에서는 스파크 잡(Spark Job)과 API를 뒷받침하는 개념을 설명한 후, 스파크 2.0의 구조를 소개하고 스파크 2.0의 여러 기능을 살펴볼 것입니다.

 

 

아파치 스파크는 무엇인가?

---

아파치 스파크는 오픈소스 분산 쿼리 및 처리 엔진입니다. 스파크는 유연성과 맵리듀스에 대한 확장성을 훨씬 빠른 속도로 제공합니다. 데이터가 메모리에 저장돼 있을 때는 아파치 하둡보다 100배 빠르며, 디스크에 저장돼 있을 때는 10배 빠릅니다.

 

스파크는 데이터를 읽고, 변형하고, 합계를 낼 수 있으며, 복잡한 통계 모델들을 쉽게 학습하고 배포할 수 있습니다. 스파크 API는 자바, 스칼라, 파이썬, R, SQL을 이용해 접근할 수 있습니다. 애플리케이션을 빌드하는데 씅리 수 있고, 여러 애플리케이션을 라이브러리로 묶어서 클러스터에 배포할 수도 있으며 파이썬 노트북을 통해(예를 들어 주피터, Spark-Notebook,  Apache Zepplin) 대화식으로 빠른 분석을 수행할 수 있습니다.

 

스파크는 파이썬 pandas 라이브러리와 R의 data.frames 또는 data.tables를 이용하는 데이터 분석가, 데이터 과학자 또는 연구우너들에게 적합한 여러 라이브러리를 제공합니다. 스파크의 데이터프레임은 pandas나 data.frames/data.tables와 유사하지만, 몇 가지 다른 부분도 있으므로 너무 큰 기대는 하지 않는게 좋습니다. SQL에 더 익숙한 사용자들은 데이터를 다지기 위해 쿼리를 사용할 수 있습니다. 또한 몇몇 선구현된, 튜닝된 알고리즘, 통계 모델과 프레임워크를 아파치 스파크로 사용 가능합니다. 머신러닝을 위한 MLib과 ML 라이브러리, 그래프 처리를 위한 GraphX와 그래프프레임, 스파크 스트리밍(DStream 스트림과 구조적 스트림)이 해당합니다. 스파크로 이러한 라이브러리들을 한 애플리케이션에서 균일하게 사용할 수 있습니다.

 

아파치 스파크는 개인 PC에서도 쉽게 동작하며, YARN과 아파치 메소스(또는 로컬 클러스터나 클라우드)에서 Standalone모드로 쉽게 사용할 수 있습니다. 스파크는 데이터를 HDFS, 아파치 카산드라, 아파치 HBase, S3와 같은 다양한 소스로부터 읽고 쓸 수 있습니다.

 

 

 

스파크 잡과 API

---

아파치 스파크 잡과 API에 대해 간략히 소개해보겠습니다. 이는 스파크 2.0의 구조에 대한 필수 기초가 됩니다.

 

* 실행 프로세스

모든 스파크 애플리케이션은 여러 개의 잡을 가질 수 있는 하나의 드라이버 프로세스를 마스터 노드에서 실행합니다. 드라이버 프로세스는 다음 그림에서처럼 실행 프로세스(Executor Process)들을 컨트롤합니다. 여러 태스크를 포함하고 있는 실행 프로세스들은 여러 개의 워커 노드로 태스크를 분산 시킵니다. 드라이버 프로세스는 태크스 프로세스의 개수와 구성을 결정합니다. 태스크 프로세스는 하나의 잡 그래프에 기반해 실행 노드에 의해 컨트롤됩니다. 모든 워커 노드는 서로 다른 잡으로부터 받은 태스크를 실행할 수 있습니다. 

 

 

스파크 잡에서의 객체 의존성은 비순환 방향성 그래프 형태로 표현할 수 있습니다. 비순환 방향성 그래프는 스파크 UI상에서 다음 그림과 같이 생성됩니다. 이 그래프를 기반으로 스파크는 스케줄링(scheduling)과 태스크 실행을 최적화할 수 있습니다.

 

 

 

* RDD(Resilient Distributed Datasets)

스파크는 RDD(Resilient Distributed Datasets)라고 불리는 이뮤터블(immutable) 자바 가상 머신(JVM) 객체들의 집합으로 만들어졌습니다. 이 강좌에서는 파이썬으로 작업하기 때문에 파이썬 데이터들이 이 JVM 객체들에 저장돼 있다는 것을 알아야 합니다. 이것에 대한 추가적인 내용은 다음 장의 RDD와 데이터프레임에서 다룰 예정입니다. 이 객체들은 잡들이 매우 빠르게 연산할 수 있게 합니다. RDD는 메모리상에서 캐시되고 저장되고 계산됩니다. 이러한 스키마는 하둡과 같은 다른 전형적인 분산 처리 시스템과 비교해서 빠른 계산 차수 연산을 가능케 합니다.

 

이와 동시에 RDD는 map(), reduce(), filter()와 같은 트랜스포메이션(transforamtion)을 제 하며, 다양한 연산을 수행하기 위한 하둡 플랫폼의 유연성과 확장성을 유지합니다. map(), reduce(), filter() 함수는 2장. 'RDD'에서 자세히 다룰 예정입니다. RDD는 병렬로 데이터에 트랜스포메이션을 적용하고 기록하는데, 이는 속도와 내구성을 향상시킵니다. RDD는 트랜스포메이션별로 데이터 흐름(data lineage: 각 중간 단계에서 그래프 형태의 트리계보)를 제공합니다. 이것으로 인해 RDD는 데이터 유실을 막을 수 있습니다. RDD 일부의 파티션이 없어지더라도, 추가적인 데이터 복구를 하지 않고 유실된 파티션을 재생산해낼 수 있는 충분한 데이터를 가지고 있습니다.

 

RDD에서 병렬 연산은 포인터를 새로운 RDD에 리턴하는 트랜스포메이션과 연산 후에 값을 리턴하는 액션(action) 이렇게 두 종류가 있습니다. (참고 사이트)

 

RDD 트랜스포메이션은 연산을 즉시 하지 않는다는 점에서 게으른 연산입니다. 트랜스포메이션은 액션이 실행되고 결과가 드라이버에 리턴돼야 할 때만 수행됩니다. 실행을 지연시키면 성능 측면에서 더욱 정교화된 쿼리를 수행할 수 잇습니다. 최적화는 아파치 스파크 DAG스케줄러(스테이지를 이용해 변형하는 스테이지 기반의 스케줄러)에서 시작됩니다. RDD는 트랜스포메이션과 액션이 분리돼 DAG 스케줄러의 경우 최적화를 쿼리상에서 수행할 수 있으며, 또한 데이터를 섞는 것 같은 상당한 리소스가 소모되는 작업을 피할 수 있습니다.

 

 

* 데이터프레임

RDD와 같이 데이터프레임은 클러스터상의 여러 노드에 분산된 이뮤터블 데이터 집합입니다. 그러나 RDD와는 달리 데이터프레임의 데이터는 칼럼명으로 이뤄져있습니다. 데이터프레임은 큰 데이ㅏ터셋을 쉽게 처리하기 위해 디자인됐습니다. 데이터프레임으로 데이터의 구조를 공식화할 수 있으며 상위 계층에 대한 추상화도 가능합니다. 이러한 점에서 데이터프레임은 관계형 데이터베이스의 테이블과 비슷합니다. 데이터프레임은 분산 데이터를 다루기 위해 그리고 전문 데이터 엔지니어뿐만 아니라 더 많은 사람들이 스파크에 접근할 수 있도록 하기 위해 각 주제별로 API들을 제공합니다.

 

스파크 엔진은 최초에 논리적인 실행 계획을 작성하고 물리적 플랜에 의해 생성된 코드를 실행합니다. 이 물리적 플랜은 비용 옵티마이저(cost optimizer)에 의해 결정됩니다. 이는 데이터프레임의 가장 큰 장점 중 하나입니다. 자바나 스칼라와 비교해, 파이썬에서는 상당히 느린 RDD와 다르게 데이터프레임은 모든 언어에 균일한 성능을 나타냅니다.

 

 

* 데이터셋

스파크 1.6에서 설명돼 있듯이, 스파크 데이터셋의 목적은 사용자가 도메인 객체에서 트랜스포메이션을 쉽게 표현할 수 있는 API를 제공하고, 또한 견고한 스파크 SQL 실행 엔진의 성능과 장점을 제공 하는 것입니다.

 

 

* 카탈리스트 옵티마이저

스파크 SQL은 SQL 쿼리와 데이터프레임 API를 모두 강화시키기 때문에 아파치 스파크에서 가장 기술적으로 발전되고 디자인된 구성 요소 중 하나입니다. 스파크 SQL 코어에는 카탈리스트 옵티마이저가 있습니다. 이 옵티마이저는 함수 프로그래밍 구조에 기반하고 있으며 두가지 목적을 위해 디자인되었습니다. 하나는 새로운 최적화 기술과 스파크 SQL 피처를 쉽게 추가하기 위함이고, 다른 하나는 외부 개발자들이 옵티마이저를 확장시킬 수 있도록 하기 위함입니다. 예를 들어, 특정 데이터 소스에 적용할 룰이나 새로운 데이터 타입에 대한 지원 등을 추가하는 것입니다. (참고 사이트)

 

 

* 프로젝트 텅스텐

텅스텐은 아파치 스파크 실행 엔진의 하위 프로젝트에 해당하는 코드네임입니다. 이 프로젝트는 현재 하드웨어들의 성능을 최대한 활용하도록 해서 스파크 알고리즘을 향상시키는데 목적을 둡니다. 그로 인해 알고리즘상에서 메모리와 CPU가 더 효율적으로 사용될 수 있도록 합니다. 그 외에 이 프로젝트는 다음 내용에 목적을 두고 있습니다.

 

 - JVM 객체 모델의 오버헤드와 가비지 컬렉션이 제거되도록 메모리를 직접적으로 관리

 - 알고리즘 디자인과 메모리 구조를 활용하는 데이터 구조 디자인

 - 애플리케이션이 현재의 컴파일러를 활용하고 CPU에 최적화되도록 코드를 실행 시간에 생성

 - 다중 CPU 호출을 줄일 수 있도록 가상함수 디스패치를 제거

 - 메모리 접근을 빠르게 하기 위해 로우 레벨 프로그래밍 활용(예를 들어, 즉각적으로 사용되는 데이터들을 CPU 레지스터에 로딩하는 등)

 - 간단한 루프를 효과적으로 컴파일하고 실행하기 위해 스파크 엔진 최적화

 

 

 

스파크 2.0의 구조

---

아파치 스파크 2.0은 지난 2년간의 플랫폼 개발로부터 나온 주요 결과물을 기반으로 한 아파치 스파크 프로젝트의 최신 배포판입니다. 아파치 스파크 2.0에서의 세 가지 오버라이딩은 텅스텐 페이스 2를 통한 성능 강화, 구조적 스트리밍, 그리고 데이터셋과 데이터프레임의 통합을 포함합니다. 데이터셋은 현재 스칼라와 자바로만 사용가능하지만, 스파크 2.0의 일부분이므로 데이터셋에 대해 소개하게 될 것입니다. 

 

* 데이터셋과 데이터프레임 통합

이전 절에서 데이터셋은 스칼라와 자바에서만 사용 가능핟고 했습니다. 그러나 스파크 2.0의 방향을 더 잘 설명하고자 다음의 내용을 설명합니다.

 

데이터셋은 아파치 스파크 1.6 버전의 일부로 2015년에 소개되었습니다.

 

데이터셋의 목적은 타입 관점에서 안전한 프로그래밍 인터페이스를 제공하기 위함입니다. 이렇게 되면 개발자들이 반구조 데이터(JSON이나 키-값 쌍과 같은)로 컴파일 시간에 타입 관점에서 안전하게 작업을 할 수 있습니다.(즉 애플리케이션이 실행되기 전에 에러를 검증할 수 있습니다.) 데이터셋 API가 파이썬으로 구현되지 않은 부분적인 이유는 파이썬에서의 변수 타입이 그다지 안전하지 않기 때문입니다.

 

데이터셋 API는 sum(), avg(), avg() 또는 group()과 같은 특정 도메인에 종속되는 하이레벨 언어 오퍼레이터를 포함합니다. 이 특성은 전통적인 스파크 RDD에 대한 유연함을 유지한 채 코드 또한 쉽게 표현하고 읽거나 쓸 수 있음을 의미하기도 합니다. 

 

데이터프레임과 비슷하게, 데이터셋도 데이터 필드와 표현을 쿼리 플래너에 노출하고 텅스텐의 빠른 인메모리(in-memory) 인코딩을 사용해 스파크 카탈리스트 옵티마이저의 장점을 취합니다.

 

스파크 API에 대한 역사는 RDD, 데이터프레임, 데이터셋 순서로 다음 다이어그램에 표시돼 있습니다.

 

 

데이터프레임과 데이터셋 API 통합은 이전 버전에 대해 브레이킹 체인지를 야기할 가능성이 있습니다. 이러한 변화로 인해 아파치 스파크 1.x 버전(브레이킹 체인지가 거의 발생하지 않음)이 아니라 2.0 버전으로 배포된 것입니다. 다음 다이어그램에서 볼수 있듯이 데이터프레임과 데이터셋 모두 아파치 스파크 2.0의 일부인 새로운 데이터셋 API에 속합니다.

 

 

이전에 언급했듯이, 데이터셋 API는 타입 관점에서 안전한 객체지향 프로그래밍 인터페이스를 제공합니다. 데이터셋은 스파크 카탈리스트 옵티마이저로 데이터 필드와 표현을 쿼리 플래너에 노출함으로써, 그리고 텅스텐의 빠른 인메모리 인코딩을 사용함으로써 장점을 취할 수 있습니다. 그러나 데이터프레임과 데이터셋이 아파치 2.0에서 통합돼, 데이터프레임은 데이터셋에서의 타입을 지정하지 않은 API(untyped API)로 새롭게 정의되었습니다.

DataFrame = Dataset[Row]

 

* 스파크세션에 대한 소개

 

예전에는 다양한 스파크 쿼리를 실행하기 위해 각각 SparkConf, SparkContext, SQLContext, HiveContext를 사용해야 했습니다. SparkSession은 StreamingContext를 포함한 이들 모두의 조합입니다.

 

한가지 예를 들어보겠습니다.

df = sqlContext.read.format('json').load('py/test/sql/people.json')

위와 같이 작성하는 대신에 다음과 같이 작성합니다.

df = spark.read.format('json').load('py/test/sql/people.json')

또는 다음과 같이 작성할 수 있습니다.

df = spark.read.json('py/test/sql/people.json')

 

SparkSession은 이제 데이터 읽기, 메타데이터 활용, 세션 세팅, 그리고 클러스터 리소스에 접근하기 위한 시작점입니다.

 

 

* 텅스턴 페이스 2

이 프로젝트를 시작했을때의 컴퓨터 하드웨어 영역에 대한 통상적인 사실은 CPU에 대한 가격 대비 성능과 RAM, 디스크, 네트워크 인터페이스에 대한 가격 대비 성능이 같지 않다는 것입니다. 비록 하드웨어 제조사가 각각 소켓에 더 많은 코어를 둘 수 있으나(병렬 처리를 통한 성능 향상), 실제 코어 스피드에는 큰 변화가 없었습니다. 

 

프로젝트 텅스텐은 2015년 소개되었는데, 스파크 엔진의 성능 향상에 그 목적을 두고 있습니다. 이러한 향상에 대한 첫번째 페이스는 다음 측면에 초점을 두고 있습니다.

 

 - 메모리 관리와 이진 프로세싱: 메모리를 명확하게 다루고 JVM 객체 모델과 가비지 컬렉션에 대한 오버헤드를 제거하기 위한 애플리케이션 시맨틱스를 활용하기.

 - 캐시 활용 연산: 메모리 계층을 활요하는 알고리즘과 데이터 구조

 - 코드 생성: 최신 컴파일러와 CPU를 활용하는 코드 생성

 

다음 다이어그램은 업데이트된 카탈리스트 엔진으로 데이터셋이 포함된 것도 보여주고 있습니다. 다이어그램의 오른쪽 부분에서 볼 수 있듯이(비용모델 오른쪽), 코드 생성이 기반 RDD를 생성하기 위해 선택된 물리적 플랜에 사용됩니다.

 

 

 

텅스텐 페이스 2의 일부로, 전체 단계 코드를 생성하기 위한 요청 단계가 있습니다. 즉, 스파크 엔진은 컴파일 시간에 특정 잡이나 태스크가 아닌, 전체 스파크 스테이지를 위한 바이트 코드를 생성합니다. 이러한 향상을 아우르는 주된 이유는 다음과 같습니다.

 

 - 가상 함수 디스패치 제거: 디스패치가 무수히 많이 발생했을 때 성능에 크게 영향을 주는 다중 CPU 호출을 줄인다.

 - 메모리상에서의 데이터 vs. CPU 레지스터: 텅스텐 페이스2는 연산 과정에 있는 데이터를 CPU 레지스터에 저장한다. 데이터를 메모리가 아닌 CPU 레지스터에서 얻기 위해 사이클 개수를 줄이는 명령인 셈이다.

 - 루프 언롤링과 SIMD: 스파크 실행 엔진을 현대 컴파일러의 장점을 극대화해 수행한다. 또한 간단한 for문을 효과적으로 실행하기 위한 CPU 최적화를 이용하기도 한다.

 

 

* 구조적 스트리밍

2016년 스파크 서밋 이스트(Spark Summit East)에서 레이놀드 신은 다음과 같이 말했습니다.

 

"스트리밍 분석을 하는 가장 간단한 방법은 스트리밍에 대한 이유를 갖지 않는 것이다."

이것이 구조적 스트리밍에 대한 기본 철학입니다. 스트리밍은 강력하지만, 빌드하고 유지하기 어렵다는 큰 문제점이 있습니다. 우버, 넷플릭스, 핀터레스트와 같은 회사들은 제품상에서 스파크 스트리밍 애플리케이션을 사용하는데, 그 회사들 역시 시스템이 제대로 잘 동작하도록 하기 위한 별도의 팀이 있습니다. 

 

이전에 암시했듯이 스파크 스트리밍을 실행할때(다른 스트리밍 시스템도 마찬가지로) 이벤트가 지연되거나, 최종 데이터 소스에 대한 결과가 불완전하거나, 오작동이 발생할 경우에 대한 상태 복구 또는 분산된 읽기/쓰기와 같이 오류가 날 수 있는 몇 가지 부분이 있습니다.

 

그러므로 스파크 스트리밍을 단순화하기 위해, 아파치 스파크 2.0에는 배치(batch)와 스트리밍을 모두 포함하는 단일 API가 있습니다. 더 간단히 말하면, 하이레벨 스트리밍 API가 아파치 스파크 SQL 엔진에 구축돼 있습니다. 하이레벨 스트리밍 API는 데이터셋과 데이터프레임으로부터 취할 수 있는 성능 및 최적화 관련 장점과 이벤트 시간, 윈도윙, 세션, 소스, 싱크와 같은 장점을 모두 제공합니다. 게다가 쿼리도 스파크 SQL 엔진에서 사용하는 것과 동일한 쿼리를 사용할 수 있습니다.

 

 

* 지속적 애플리케이션

아파치 스파크 2.0은 데이터프레임과 데이터셋을 통합했을 뿐만 아니라 스트리밍, 대화식, 배치 쿼리 역시 통합했습니다. 이로 인해 데이터를 스트림에서 합쳐서 기존의 JDBC/ODBC를 사용해 결과를 제공하는 기능, 실행 시간에 쿼리를 변경하는 기능 또는 많은 시나리오를 위한 ML 모델을 빌드하고 적용할 수 있는 기능이 포함된 새로운 유스케이스가 개발 되었습니다. 

 

 

이제 이 모든 것을 같이 사용해 엔드 투 엔드 지속적 애플리케이션을 빌드할 수 있습니다. 엔드 투 엔드 지속적 애플리케이션을 통해 실시간 데이터에서 사용하는 것과 같은 쿼리를 사용해 배치(BATCH) 처리를 할 수 있고, ETL 작업(Extract, Transform, Load)을 실행할 수 있으며, 보고서를 생성하고 스트리밍에서의 특정 데이터를 업데이트 및 추적할 수 있습니다.

 

 

요약

---

아파치 스파크가 무엇인지 알아봤고, 스파크 잡과 API에서 제공하는 핵심 기능을 설명했습니다. 또한 RDD, 데이터프레임, 데이터셋에 대한 핵심 기능도 설명핶고, 데이터프레임이 어떻게 스파크 SQL 엔진의 카탈리스트 옵티마이저와 프로젝트 텅스텐을 이용해 아파치 스파크에서 더 빠른 쿠 ㅓ리 성능을 나타내는지 이야기했습니다. 마지막으로 텅스텐 페이스 2, 구조적 스트리밍, 데이터프레임과 데이터셋 통합을 포함해 스파크 2.0의 하이레벨 구조도 설명했습니다.

 

다음 포스팅에서는 스파크의 가장 핵심적인 데이터 구조인 RDD를 다룰 예정입니다. 실제로 파이스파크를 시작할 수 있도록 RDD를 어떻게 생성하는지 설명하고, 이와 같은 스키마리스(schema-less) 데이터를 ㅌ트랜스포머와 액션을 사용해 어떻게 수정하는지 설명할 것입니다.

출처: https://12bme.tistory.com/305?category=737765 [길은 가면, 뒤에 있다.]