스파크는 무엇이고 왜 쓰는지? 스파크에 대해 알아보기

스파크는 무엇인가요?

 

기존의 애플리케이션들은 단일 프로세서에서만 실행되도록 설계되었다. 문제는 시대가 바뀌면서 데이터의 양은 기하급수적으로 늘어났지만, 물리적인 방열 한계 때문에 하드웨어의 성능 향상은 멈췄다는 것이다. 그래서 개발자들은 단일 프로세서의 성능을 향상하는 방법 대신 CPU 코어를 병렬로 더 많이 추가하는 방법을 선택했다. 지금도 데이터는 나날이 거대해져 가고 있고, 마침내 컴퓨터 한 대로는 도저히 처리할 수 없을만큼 거대해졌다.

 

스파크는 이 문제를 해결하기 위해 등장한 데이터 병렬 처리 오픈 소스 라이브러리 엔진이다. 스파크를 사용하면 클러스터 환경에서  데이터를 병렬로 처리할 수 있다. 단일 노트북 환경에서 실행할 수도 있지만, 수천 대의 서버로 구성된 엄청난 규모의 클러스터에서 실행할 수도 있고, 후자로 실행했을 때 빅데이터도 빠르게 처리할 수 있게 해준다.

 

---

 

스파크의 기본 이해

스파크의 기본 아키텍처

 

컴퓨터 클러스터는 여러 컴퓨터의 자원을 모아서 하나의 컴퓨터처럼 사용할 수 있게 만드는 것이다. 클러스터를 구성한 후에는, 클러스터들의 작업을 조율할 수 있는 프레임워크가 필요한데, 스파크는 바로 그 역할을 하는 프레임워크이다.

 

스파크 애플리케이션은 Driver 프로세스와 다수의 Executor 프로세스로 구성된다.

 

드라이버 프로세스는 클러스터 노드 중 하나에서 실행되며 main() 함수를 실행한다.

 

프로세스	하는 일
드라이버 프로세스	스파크 앱의 유지 관리, 사용자 프로그램이나 입력에 대한 응답, 배포, 스케줄링 등 필수적 작업
익스큐터 프로세스	드라이버 프로세스가 할당한 작업을 수행

 

Spark Standalone cluster

 

 

---

 

스파크의 언어 API

 

다양한 프로그래밍 언어로 스파크 코드를 실행할 수 있다.

 

개발 커뮤니티에서 가장 권장되는 언어는 스칼라와 파이썬이다. 

 

스파크는 스칼라로 개발되어 있기 때문에 기본 언어인 스칼라를 쓰는 것이 권장된다.

 

파이썬에서 UDF나 RDD를 사용할 경우 데이터 직렬화에서 큰 비용이 들고 안전성이 떨어지게 되므로,

파이썬에서 UDF, RDD를 사용하는 것은 반드시 피해야 한다.

하지만 파이썬은 데이터 과학에서 가장 범용적으로 쓰이는 언어이므로 DS, DE 들의 학습 비용이 적다는 장점이 있다.

 

파이썬 스파크는 스칼라 스파크가 지원하는 거의 모든 구조를 지원한다.

UDF, RDD를 사용하지 않는다면 스칼라와 큰 성능 차이는 없다.

 

---

 

스파크 엔진

똑똑한 GPT 챗봇에게 물어보았다.

Apache Spark에서 사용되는 Catalyst 엔진은 오픈소스 SQL 쿼리 엔진으로, 스파크의 구조적 API인 DataFrame과 Dataset의 최적화를 담당합니다.

Catalyst 엔진은 스파크의 DataFrame API를 사용하여 작성된 쿼리를 최적화하는 데 사용됩니다. 이 최적화는 스파크의 성능을 향상하고, 대규모 데이터셋에서 빠른 쿼리 처리를 가능하게 합니다.
Catalyst 엔진은 다음과 같은 최적화를 수행합니다.

- 필요한 컬럼만 선택하거나 조합하여 프로젝션 수행
- 필요한 데이터만 필터링하여 프레디케이트 푸시다운 수행
- 연속적인 조인을 하나의 스파크 작업으로 최적화하여 성능 향상
- 적절한 파티셔닝을 수행하여 셔플링 비용 최소화
- 쿼리 실행 계획의 최적화

이러한 최적화를 통해 Catalyst 엔진은 스파크의 성능과 확장성을 향상하고, 빠른 데이터 처리와 대규모 분산 데이터셋의 처리를 가능하게 합니다.

 

 

 

구조적 API의 실행 과정

작성한 스파크 API 쿼리가 사용자 코드에서 실제 실행 코드로 변환되는 과정

 

1. 코드 작성

2. 스파크가 코드를 논리적 실행 계획으로 변환

논리적 실행 계획단계에서는 클러스터 레벨을 배제하고, 추상적 트랜스포메이션만 표현해서 쿼리가 변환된다.

이 단계에서는 코드의 유효성, 테이블이나 컬럼의 존재 여부 등만 검증한다.

 

3. 스파크가 논리적 실행 계획에서 물리적 실행 계획으로 변환, 이 과정에서 최적화 이뤄짐

물리적 실행 계획은 클러스터 환경에서 실행하는 방법을 정의한 것이다.

클러스터 자원을 고려하여 여러가지 실행 전략을 생성한 후 각각을 비교하여 최적의 전략을 만들어 준다. (똑똑하네..!)

이렇게 최적전략으로 선택된 물리적 실행계획은 최종적으로 저수준 API 인 RDD로 컴파일된다.

 

4. 클러스터에서 물리적 실행 계획 실행됨

 

---

 

스파크 시작하기

SparkSession

스파크 세션은 위에서 설명한 드라이버 프로세스를 말한다.

스파크 세션은 사용자가 정의한 처리 명령을 클러스터에서 실행한다.

 

pyspark 로 spark를 실행해서 명령을 실행시켜 보았다.

 

spark

spark.range(1000)

 

 

column 1개에 1000개의 숫자가 들어간 DataFrame 이 생성되었다.

만약 이 명령을 클러스터 모드에서 실행했다면, 

이 데이터 프레임의 데이터는 쪼개져서 n개의 익스큐터에 각각 할당되게 된다.

 

 

---

 

 

Data Frame

위에서 생성한 데이터 프레임을 조금 더 들여다보자..

데이터 프레임은 테이블의 데이터를 로우와 칼럼으로 단순하게 표현한 것이다.

스파크 데이터프레임은 수천 대의 컴퓨터에 분산될 수 있다.

엄청나게 큰 빅데이터를 처리할 때에, 데이터를 클러스터의 익스큐터 프로세스에 나눠 저장한 후 처리하면, 한 대로만 처리하는 것보다 훨씬 빠를 것이다.

 

DataFrame 은 파이썬에 원래 존재하는 개념이다.

pandas 등의 라이브러리를 사용해 보았다면, 이미 DataFrame의 개념을 숙지하고 있을 것이다.

하지만 Py-spark로 구현하지 않은 Python의 데이터프레임은 여러 컴퓨터에서 분산처리가 되지 않고, 단일 컴퓨터에서만 존재할 수 있다.

 

비슷한 API로 자바와 스칼라에서 쓰이는 Dataset이라는 것도 있는데, 데이터 Row의 형식을 미리 정의해 두고 쓰는 분산 컬렉션이다.

Scala에서는 Dataset[T]로 작성한다. 

Dataset의 장점은 타입 안정성을 보장해 주는 것이다.

Dataset에서 불러온 데이터가 사전에 정의한 T 클래스와 형식이 다르면 에러가 발생한다.

 

 

---

 

Partition

 

스파크는 어떤 단위로 데이터 프레임을 각각의 익스큐터에게 분할해 줄까?

파티션이라고 하는 청크 단위로 데이터를 분할해 준다.

 

예를 들어 이런 데이터 프레임이 있을 때,

 

이름	성별	혈액형
이사라	여자	A
박연진	여자	O
최혜정	여자	A
전재준	남자	B

 

파티션을 '성별' 칼럼으로 설정한다면,

이 데이터프레임은 성별이 여자인 데이터셋 1개, 성별이 남자인 데이터셋 1개, 이렇게 2개의 데이터셋으로 쪼개진다.

파티션을 '혈액형' 칼럼으로 설정한다면,

이 데이터프레임은 혈액형이 A인 데이터셋 1개, 혈액형이 O인 데이터셋 1개, 혈액형이 B인 데이터셋 1개, 이렇게 3개의 데이터셋으로 쪼개진다.

 

 

근데 이 데이터의 범위가 많이 확대되어서... 우리가 전 국민 5000만 명의 데이터를 갖고 있다고 가정해 보자.

그런데 우리가 성별을 파티션으로 설정한다면, 파티션이 남자 여자 2개만 생성되어서 스파크에 수천 개의 익스큐터가 있더라도 2개의 익스큐터에만 데이터를 몰빵 해주게 될 것이다..

물론 반대로 수백 개의 파티션이 있어도 익스큐터가 한 개 밖에 없다면 데이터는 익스큐터 한 개에 들어갈 수밖에 없다.

 

 

스파크는 내부 로직을 사용해서 물리적 파티션을 구분하기 때문에, 사용자는 크게 신경을 쓰지 않아도 된다.

(하지만 성능 개선이 필요한 경우 repartition 등을 써서 리팩토링을 하기도 한다..)

 

---

 

Transformation

 

데이터프레임을 변경할 때에 스파크에 내리는 명령이 트랜스포메이션이다.

 

스파크에서 운영 코드를 작성할 때 꼭 이해해야 할 개념이

트랜스포메이션의 추상성인 것 같다.

 

myRange = spark.range(1000).toDF("number")
divisBy2 = myRange.where("number % 2 = 0") # Transformation executed

 

2행으로 트랜스포메이션 코드를 실행해도 결과는 출력되지 않는다.

왜냐... 아직 내 명령은 아직 실행되지 않았기 때문이다. 메모리에 실행 계획으로 착실히 쌓이기만 했을 뿐...

내가 액션을 호출하지 않으면 스파크는 트랜스포메이션을 실제로 수행하지 않는다.

 

액션은 뒤에서 설명할 텐데, 결과 계산 명령을 말한다. 대표적으로는 count()가 있다.

액션이 일어나기 전까지 스파크는 실행 계획 (쿼리)를 계속 쌓아간다.

액션이 일어나기 전 마지막 순간까지 계속 최적화된 실행 계획으로 만들어간다.

 

중요: *트랜스포메이션은 필요한 순간까지 실행되지 않는다!*

 

---

 

Action

 

사용자가 트랜스포메이션으로 실행 계획을 계속 쌓은 후, 실제 연산을 수행하려면 액션 명령을 내려야 한다.

액션은 결과가 실제로 필요한 순간에 내려지는 명령이다.

 

count() 를 실행했더니 그제서야 결과를 출력해주었당.

 

- Spark의 액션 -

 

row의 카운트 출력 - count()

콘솔에서 데이터를 보는 액션 - show()

출력, 저장 액션

각 언어로 된 네이티브 객체에 데이터를 모으는 액션

 

 

메모리에 DataFrame 캐싱 작업을 수행하는 용도로 액션을 쓰기도 한다. (*중요*)

 

실제로 count() 가 필요없더라도

'이 시점에서 불필요한 read 를 줄이기 위해 캐싱을 한 번 해야겠어...' 라고 생각한 개발자가

count() 를 코드에 집어 넣어 결과 반환을 트리거 하는 것이다.

결과 반환을 통해 데이터 프레임은 액션이 작성된 그 시점에 메모리에 캐싱된다.

 

---

 

스파크의 조인

1. 셔플 조인

2. 브로드캐스트 조인

 

셔플 조인은 큰 테이블을 다른 큰 테이블에 조인할 때에 발생한다. 셔플 조인은 전체 노드 간 통신을 발생시킨다.

 

브로드캐스트 조인은, 한 테이블이 단일 워커 노드의 메모리 크기에 적합할 정도로 작은 경우 사용할 수 있는 최적화 조인 기법이다. 이 경우 최초에만 노드 간 통신이 발생하고 그 이후로는 발생하지 않는다. 

 

-> 스파크에게 특수조인을 사용하도록 힌트를 줄수도 있고, 물리적 실행 계획 단계에서 스파크가 결정하도록 둘 수도 있다.

 

 

---

 

저수준 API RDD

 

RDD는 객체를 다루는 데 필요한 기본 기능 API를 담고 있는 라이브러리입니다.

실제로 스파크의 거의 모든 기능은 RDD를 기반으로 만들어졌습니다.

하지만 RDD는 저수준이고 이해하기 어렵기 때문에, 실제 코드에서는 스파크 사용자를 위해 제공되는 구조적 고수준 API 를 사용하는 것이 권장됩니다.

 

파티션과 같은 물리적 실행 특성을 변경하는 등의 세밀한 제어가 꼭 필요한 것이 아니라면,

RDD 는 권장되지 않습니다.

 

---

 

스파크 UI

 

로컬에서 스파크를 실행 후 http://localhost:4040으로 접속하면 스파크 잡의 상태, 환경 설정, 클러스터 상태 등의 정보를 확인할 수 있다.

 

 

 

 

---

스파크 다른 포스팅 보기

2023.02.26 - [[IT] 공부하는 개발자/Data engineering] - 스파크 API - 성능 최적화, 리팩토링 practice

---

 

Reference

스파크 완벽 가이드 (한빛 미디어)