Spark 내부동작 (Execution Plan)

 

 

Contents

1. Spark 파일 포맷

Parquet, Avro, csv, json

 

2. Execution Plan

Spark이 우리가 만든 코드를 어떻게 실행할 지 보겠습니다.

Action과 Transformation의 차이를 살펴봅니다.

하나의 액션이 하나의 잡이고 하나의 잡이 다수의 스테이지를 만들 수 있고 하나의 스테이지가 다수의 태스크를 만들 수 있습니다.

 

그리고 이런 Execution Plan을 Spark Web UI로 살펴볼 수 있습니다.

 

3. Bucketing과 Partitioning

데이터를 처리하기 쉬운 형태로 HDFS에 저장하는 방식에 대해 살펴보겠습니다.

 

 

 

 

 

 

 

 

 

 

 

Execution Plan

Spark은 개발자가 만든 코드를 어떻게 변환하여 실행하는가?

 

 

 

 

 

 

 

 

다음 데이터 프레임 연산을 자세히 보자

spark.read.option("header", True).csv(“test.csv”). \
 where("gender <> 'F'"). \
 select("name", "gender"). \
 groupby("gender"). \
 count(). \
 show()

 

where에서 gender가 F가 아닌 것들로 필터링 했습니다.

name, gender 2개의 필드만 select했습니다.

gender로 그룹핑하고

count 했습니다.

최종적으로 드라이버쪽으로 20개의 레코드를 디스플레이합니다.

 

 

 

 

Transformation

 

로딩이 된 순간 데이터 프레임은 하나 혹은 그 이상의 파티션으로 구성됩니다. 그 이후에 WHERE와 SELECT를 수행하는데 이 때 필터링인 WHERE와 SELECT 작업은 해당 파티션에서 가능한 작업(Narrow Transformation)입니다. 셔플링이 필요가 없습니다. executor에서 진행되는 task들이 독립적으로 작업을 수행할 수 있습니다. 

 

그런데 GROUPBY를 호출하는 순간 GROUPBT의 키에 맞는 같은 값을 갖는 레코드들이 같은 파티션으로 재정렬되어야 하기 때문에 셔플링이 필요하고 발생(Wide Transformation)합니다.

 

COUNT는 해당 파티션에서 셔플링 없이 가능한 작업입니다.

 

 

Action

 

SHOW는 Action이라고 합니다. 앞에 있던 데이터 프레임 관계된 작업들을 실제로 수행시키는 역할을 합니다. 컨셉은 Spark의 Lazy Execution이며 WHERE ~ COUNT 까지 실제 작업들이 진행되는 것이 아닌 Action이라는 특정 작업들이 수행될 때 그제서야 그 앞의 연산들이 수행됩니다. SHOW는 대표적인 Action입니다.

 

Write과 Collect 역시 Action입니다.

 

 

 

 

 

 

 

 

 

 

Transformations and Actions

 

Transformations

Narrow Dependencies: 독립적인 Partition level 작업

    1. select, filter, map 등등

Wide Dependencies: Shuffling이 필요한 작업

    1. groupby, reduceby, partitionby, repartition 등등

 

Actions

Read, Write, Show, Collect -> Job을 실행시킴 (실제 코드가 실행됨)
Lazy Execution
    1. 어떤 장점이 존재할까?
        • 더 많은 오퍼레이션을 볼 수 있기에 최적화를 더 잘할 수 있음. 그래서 SQL이 더 선호됨

 

Action을 통해 실행되는 연산들을 Job이라고 부릅니다. 내가 작성하는 Spark 코드는 하나 혹은 그 이상의 Job으로 구성되는 것이고 하나의 Job은 다수의 Transformation으로 구성됩니다. 그런데 Transformation들을 Narrow냐 Wide냐로 나누어 스테이지라는 것을 다시 구성합니다.

 

하나의 Job은 하나 혹은 그 이상의 스테이지라는 것으로 구성되고 각 스테이지는 하나 혹은 그 이상의 Transformation으로 구성됩니다.

 

한 스테이지는 셔플링 없이 독립적, 병렬적으로 가능한 연산입니다. 만일 내 Job안에서 셔플링이 생겼다면 2개의 스테이지로 구성되었다는 얘기입니다.

 

하나의 스테이지는 다수의 task로 구성되고 각 task는 하나의 Narrow Dependency Transformation을 실행하는 겁니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Transformations and Actions 시각화

다음 DataFrame 조작 코드에 적용해보자

spark.read.option("header", True).csv(“test.csv”). \
 where("gender <> 'F'"). \
 select("name", "gender"). \
 groupby("gender"). \
 count(). \
 show()

 

 

SHOW라는 Action이 만든 Job은 2개의 스테이지로 구성됩니다.

 

첫 번째 스테이지는 WHERE, SELECT

두 번째 스테이지는 GROUPBY뒤에 COUNT하는 것

 

각 스테이지는 Narrow Dependency를 수행하는 task들로 구성됩니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Jobs, Stages, Tasks

 

Action -> Job -> 1+ Stages -> 1+ Tasks

 

Action

Job을 하나 만들어내고 코드가 실제로 실행됨

 

Job

하나 혹은 그 이상의 Stage로 구성됨
Stage는 Shuffling이 발생하는 경우 새로 생김

 

Stage

DAG의 형태로 구성된 Task들 존재
여기 Task들은 병렬 실행이 가능

 

Task

가장 작은 실행 유닛으로 Executor에 의해 실행됨

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Transformations and Actions 시각화

spark.read.option("header", True). \
 csv(“test.csv”). \
 where("gender <> 'F'"). \
 select("name", "gender"). \
 groupby("gender"). \
 count(). \
 show()

 

.option("inferSchema", True)가 추가되면 JOB이 하나 더 추가됨

 

 

header가 True이므로 csv파일을 읽어봐야 하므로 read operation이 Job 0을 생성

 

show로 Job 1이 생성됩니다.

 

 

execution plan을 보며 최적화할 포인트가 더 있는지 어디가 병목(Bottleneck)인지 확인할 수 있습니다.