Spark Stage 알아보기

- 목차

 

소개.

Spark Program 은 작성된 RDD 의 또는 DataFrame 의 데이터 처리 흐름을 기반으로 실행 계획을 생성합니다.

실행 계획 또한 Logical Execution Plan 과 Physical Execution Plan 으로 나뉘는데요.

Task 와 Stage 는 Physical Execution Plan 에 해당합니다.

이번 글에서 Spark 의 Task 와 Stage 는 무엇이며, 어떻게 실행되는지에 대해서 상세히 알아보려고 합니다.

 

Task.

Task 는 Spark 의 가장 작은 실행 단위입니다.

Task 는 Executor 에서 실행이 되구요.

Executor 는 단순히 JVM Process 라고 생각하시면 됩니다.

그리고 Task 는 Executor 에서 실행되는 JVM Thread 입니다.

 

RDD 와 Transformation 그리고 Action 이 조합되어 하나의 Task 가 됩니다.

아래 예시와 같이 Source 에 해당하는 RDD 는 Map & Filter Transformation 와 함께 하나의 Task 가 됩니다.

List<Integer> source = Arrays.asList(1,2,3,4,5);
JavaRDD<Integer> sourceRDD = context.parallelize(source);
JavaRDD<Integer> mapperRDD = sourceRDD.map(x -> x * 1);
JavaRDD<Integer> filterRDD = mapperRDD.filter(x -> x > 4);
List<Integer> result = filterRDD.collect();

 

위 Spark RDD 처리 코드는 아래와 같은 DAG 로 표현됩니다.

parallelize -> map -> filter 로 이어지는 이 흐름은 하나의 Stage 로 표현됩니다.

그리고 parallelize 로 분리되는 Partition 갯수 만큼의 Task 가 생성됩니다.

 

만약 위 케이스의 Partition 이 N 이라면 1개의 Stage 와 그 내부의 N개의 Task 로 표현됩니다.

 

Stage.

Partition 의 수만큼 Task 가 생성이 되며, Task 의 집합이 바로 Stage 가 됩니다.

그럼 Stage 를 나누는 기준을 무엇일까요 ?

그 기준은 Partition 의 Shuffling 입니다.

다른 표현으로 Wide Dependency Transformation 이라고도 불리는데요.

Partition 과 그 다음 단계의 Partition 사이에 Data Shuffling 이 발생하게 되면 새로운 Stage 가 생겨납니다.

 

코드 예시와 이미지 자료를 통해서 한번 알아보겠습니다.

바로 위 코드 예시에서 Wide Dependency Transformation 의 한 가지 종류인 SortBy 를 사용하였습니다.

List<Integer> source = Arrays.asList(1,2,3,4,5);
JavaRDD<Integer> sourceRDD = context.parallelize(source);
JavaRDD<Integer> mapperRDD = sourceRDD.map(x -> x * 2);
JavaRDD<Integer> filterRDD = mapperRDD.filter(x -> x > 4);
JavaRDD<Integer> sortedRDD = filterRDD.sortBy(x -> x, true, 4);
List<Integer> result = sortedRDD.collect();

 

SortBy Transformation 을 사용함으로써 2개의 Stage 가 생성되었습니다.

첫번째 Stage 에서 각 Partition 은 두번째 Stage 로 Shuffling 과 Repartitioning 이 됩니다.

 

또한 SortBy, GroupBy 등의 Wide Dependency Transformation 을 사용하지 않더라도,

repartition 를 통해서도 Stage 가 구별됩니다.

왜냐하면 이들은 Repartition 을 구현하는 Transformation 들이기 때문입니다.

반면 coalesce 는 Narrow Dependency Transformation 이기 때문에 Stage 구분에 영향을 주진 않습니다.

List<Integer> source = Arrays.asList(1,2,3,4,5);
JavaRDD<Integer> sourceRDD = context.parallelize(source, 5);
JavaRDD<Integer> repartitionRDD = sourceRDD.repartition(10);
JavaRDD<Integer> coalesceRDD = repartitionRDD.coalesce(5);
coalesceRDD.collect();