파티션과 셔플링

Spark에서 데이터를 처리할 때 알아야 하는 중요한 개념 두 가지를 정리해보려고 한다.

파티션이란

분산 저장 단위, Block

빅데이터를 병렬로 처리하기 위해서 선행되어야 하는 조건은 데이터의 분산 저장이다. HDFS와 같은 분산 파일 시스템에서는 분산 저장되는 데이터의 최소 기본 단위를 블록이라 하고, 이 블록 단위로 데이터를 나눠서 보관한다.

• 이 블록의 기본 크기는 128MB이다.
• 데이터의 유실 방지를 위해 replication factor 수만큼 여러 노드에 분산 저장한다.

분산 처리 단위, Partition

이렇게 분산 저장되어 있는 데이터 위에서 데이터 처리 엔진인 Spark이 구동된다. Spark에서 사용하는 병렬 처리의 기본 단위를 파티션이라고 한다.

파일을 읽어올때, 파티션 하나가 포함하는 데이터의 기본 크기도 128MB이다.

파티션과 블록은 동일한 것?

처음에 헷갈렸던 부분은 이 블록과 파티션은 동일한 개념인가했는데, 그렇지는 않고 효율적인 병렬처리를 위한 기본값으로 이해하는 관점에서 블록과 파티션을 바라보면 될 것 같다. 초기 로딩 시 최적화된 상태로 부르기 위한 것이지, 파티션의 크기도 Spark 설정값(spark.sql.files.maxPartitionBytes)으로 조정할 수 있다.

이는 데이터의 소스에 따라 달라지는데, HDFS 같은 파티셔닝된 파일 형태의 데이터 소스에서는 Spark이 이 파티션별로 나누어서 병렬로 구성하지만,

RDBMS에서 JDBC를 통해 테이블을 실제로 로딩할 때는, 적절한 파티션 기준(어떤 컬럼으로 나눌지, 몇 개로 나눌지) 등을 설정해주지 않으면 하나의 큰 파티션으로 가져온다.

효과적인 병렬 처리를 위해서, 파티션은 한 스테이지 안에서 서로 의존성을 갖지 않게 설계된다. 의존성이 없기 때문에 독립적인 스레드 위에서 동시에 처리가 가능하고, 파티션을 더 많이 나눌수록 처리 속도가 올라간다.

그래서 Spark에서 사용하는 데이터 프레임은 사용자가 볼 때는 하나의 데이터처럼 보이지만, 이는 논리적으로 하나의 데이터 세트처럼 보이게 한 것이고 내부적으로는 여러 파티션으로 구성되어 있다.

파티션을 처리하는 주체, Executor

실제 Spark 태스크들을 수행하는 JVM 프로세스를 말한다. 이때 태스크는 파티션을 맡아서 처리한다.

Spark 코드는 Driver에서 실행된다. 실제 Action 연산 등에 의해서 Spark Job이 수행될 때, Driver는 해당 잡에 대한 실행 계획을 세우고, 이에 맞는 태스크들을 Spark 세션에 할당된 Executor들에 배분한다. 이때 이 태스크들은 파티션에 대응되어 실행된다.

태스크와 파티션의 대응.

이때 각각의 Executor가 동시에 처리할 수 있는 태스크 수는, 해당 Executor에 할당된 CPU 코어 수만큼이다.

따라서 Spark Job의 병렬성과 적절한 파티션 수를 계산할 때는, 실제로 얼마나 병렬 처리가 가능한지를 다음 식으로 확인할 수 있다.

병렬처리 가능한 태스크/파티션 수 = executor_num * cpu_core_num (executor 별)

Spark Job이란

작성한 Spark 프로그래밍에서 실행되는 과정에서 만들어지는 가장 큰 작업 실행 단위이다. Action 연산을 호출할 때, 하나의 스파크 잡이 만들어진다.

Job은 Stage -> Task 로 나눠진다.

• Job (잡): Action 연산에 의해 생성되는 가장 큰 단위이다.
• Stage (스테이지): 잡을 쪼갠 중간 단위. 데이터가 셔플링(Shuffling)되어야 하는 시점을 기준으로 잡이 나뉜다.
• Task (태스크): 가장 작은 실행 단위이다. 태스크 하나가 파티션 하나를 처리하게 되는 구조이다.

셔플링이란

기본적인 Spark의 데이터 처리 방식

Spark으로 데이터를 처리할 때는 보통 데이터 프레임을 기본 단위로 삼아, 여러 연산을 단계적으로 구성하면서 최종 산출물을 만들어 나간다.

예) 데이터 프레임 -> 필터링 -> map 변환 -> group by -> sort -> 최종 결과

이러한 처리와 변환 작업은 데이터 프레임 API를 활용하거나 Spark SQL을 통해 수행할 수 있다.

앞서 데이터 프레임은 내부적으로 다수의 파티션으로 구성된다고 했다. 각 파티션이 연산을 거칠 때, 파티션 내의 데이터가 변경되거나 여러 파티션이 하나의 파티션으로 통합되는 일이 발생한다.

이때 일어나는 데이터의 이동을 셔플링이라고 부른다.

Shuffling, 새로운 파티션이 생성되며 데이터 이동이 발생되는 현상

조금 더 구체적으로 살펴보면,

데이터 프레임에 사용하는 오퍼레이션에 따라서 병렬로 나눠서 처리되던 파티션들이 합쳐지면서 새로운 파티션으로 구성되는 경우가 있다.

데이터 프레임 연산들에 따라 나눠서 보면 다음과 같다.

파티션을 이루는 데이터가 단순히 변환되거나 사라지는 경우

주로 map, filter 등과 같은 연산들이 이에 해당된다.

셔플링이 일어나지 않는 연산 예시. (filter)

이 경우에는 파티션 데이터가 새로 만들어지지는 않으니, Executor 사이에서 데이터 이동이 불필요하다.

다른 파티션들을 이용해 새로운 파티션을 만들어야하는 경우

주로 group by, sort 등과 같은 연산들이 이에 해당된다.

셔플링이 일어나는 연산 예시. (group by)

group by를 기준으로 설명해보면, 특정 키를 중심으로 데이터를 집계해야 하기 때문에 여러 파티션에 흩어져 있는 데이터를 모두 확인해야만 한다.

이때 기존 파티션들은 자기 데이터를 키별로 나누어 저장해두고, 새로 만들어지는 파티션이 자기 키에 해당하는 데이터들을 가져온다. 그 결과 같은 키를 가진 데이터가 한 파티션으로 모이게 된다. 데이터가 fan-in 되는 구조라고 볼 수 있을 것 같다.

이때 이 연산을 위해 데이터가 한 곳으로 모이는, 즉 데이터가 이동되는 일들이 발생하는데, 이 과정을 셔플링이라고 한다.

셔플링은 데이터를 이동시키기 위해 디스크 I/O, 데이터 직렬화/역직렬화, 네트워크 트래픽 발생, 그리고 데이터를 정렬·구성하기 위한 메모리 사용 등을 동반하기 때문에 비용이 매우 많이 드는 무거운 작업에 속한다.

셔플링이 발생하는 경우

셔플링은 지금의 파티션 구성 그대로는 결과를 만들 수 없을 때 일어난다. 그런 상황은 둘로 나뉜다.

키를 기준으로 데이터를 다시 묶어야 하는 연산

• groupBy : 같은 키끼리 집계
• join : 양쪽에서 같은 키를 맞춤
• distinct : 같은 값끼리 모아 중복 제거
• orderBy, sort : 값의 범위대로 다시 나눔

파티션 수 자체를 바꾸는 연산

• repartition : 파티션 수를 늘리거나 줄임
• repartitionByRange : 값의 범위를 기준으로 다시 나눔

coalesce도 파티션 수를 줄이는 연산이지만 셔플링이 일어나지 않는다. 같은 Executor 안에 있는 파티션끼리만 합치기 때문이다. 파티션 수를 줄일 때 repartition 대신 coalesce를 쓰는 이유가 여기에 있다.

셔플링이 끝난 뒤 만들어지는 파티션 수는 spark.sql.shuffle.partitions 설정으로 정해지며, 기본값은 200이다.

데이터 스큐 현상

이때, 셔플링이 발생한 이후에 새로 생성되는 파티션은 데이터가 가지는 특성, 데이터 분포의 불균형이 있을 경우, 변환된 파티션들 중, 특정 파티션으로 데이터가 몰리게 된다. 이러한 현상을 데이터 스큐현상이라고 한다.

데이터 스큐 예시.

데이터 스큐가 문제가 되는 이유는 병렬 처리의 이점이 줄어들기 때문이다. 병렬 처리는 일을 균등하게 나눠 가질 때 가장 빠른데, 위 예시 그림처럼 특정 파티션으로만 데이터가 몰리게 되면 데이터간 균등 분배가 깨지게된다. 극단적인 경우에는 단일 노드로 처리하는 것보다 느려질 수도 있다. 때문에 이러한 데이터 스큐 문제를 적절하게 해결할 수 있게, 파티션을 생성하는 최적화가 중요해진다.