Spark 데이터 처리

 

 

Contents

 

1. Spark 데이터 처리

 

2. Spark 데이터 구조: RDD, DataFrame, Dataset

 

3. 프로그램 구조

 

4. 개발/실습 환경 소개

 

5. Spark DataFrame 실습

 

 

 

 

 

 

 

 

 

 

 

 

Spark 데이터 처리

Spark 데이터 처리의 기본이 되는 파티션에 대해 알아보자

 

 

 

 

 

 

 

 

 

 

 

 

 

Spark 데이터 시스템 아키텍처

 

 

 

 

 

 

 

 

 

 

 

 

데이터 병렬처리가 가능하려면?

 

데이터가 먼저 분산되어야함

하둡 맵의 데이터 처리 단위는 디스크에 있는 데이터 블록 (128MB)
    - hdfs-site.xml에 있는 dfs.block.size 프로퍼티가 결정
Spark에서는 이를 파티션 (Partition)이라 부름. 파티션의 기본크기도 128MB
    - spark.sql.files.maxPartitionBytes: HDFS등에 있는 파일을 읽어올 때만 적용됨

 

다음으로 나눠진 데이터를 각각 따로 동시 처리

맵리듀스에서 N개의 데이터 블록으로 구성된 파일 처리시 N개의 Map 태스크가 실행
Spark에서는 파티션 단위로 메모리로 로드되어 Executor가 배정됨

 

 

 

 

 

 

 

 

 

 

 

처리 데이터를 나누기 -> 파티션 -> 병렬처리

예를 들어 Executor가 2개이고 각 Executor의 CPU가 1개라면 2x1=2로 한번에 2개씩 병렬 처리

 

 

 

 

 

 

 

 

 

 

Spark 데이터 처리 흐름

 

데이터프레임은 작은 파티션들로 구성됨

데이터프레임은 한번 만들어지면 수정 불가 (Immutable)

수정하려면 새로 데이터프레임을 생성

 

입력 데이터프레임을 원하는 결과 도출까지 다른 데이터 프레임으로 계속 변환

sort, group by, filter, map, join, …

 

 

모든 작업이 파티션간에 데이터 이동 없이 이루어지지는 않습니다. map, filter라면 몰라도 group by나 sort 등은 새로 파티션을 만들어야 하며 이는 파티션간의 데이터 이동이 있어야 합니다. 이 때 네트워크를 통해 데이터 전송이 발생하고 새로운 파티션이 만들어질텐데 새로 만들어진 파티션 간의 데이터 크기가 균등하지 않을 수 있습니다.

 

파티션간에 데이터 이동이 필요한 경우를 셔플링이라고 합니다.

 

 

 

 

 

 

 

 

 

 

 

셔플링: 파티션간에 데이터 이동이 필요한 경우 발생

셔플링이 발생하는 경우는?

명시적 파티션을 새롭게 하는 경우 (예: 파티션 수를 줄이기)
시스템에 의해 이뤄지는 셔플링
     - 예를 들면 그룹핑 등의 aggregation이나 sorting

 

셔플링이 발생할 때 네트웍을 타고 데이터가 이동하게 됨

몇 개의 파티션이 결과로 만들어질까?
    - spark.sql.shuffle.partitions이 결정
     - 기본값은 200이며 이는 최대 파티션 수
     - 오퍼레이션에 따라 파티션 수가 결정됨
     - random, hashing partition, range partition 등등
     - sorting의 경우 range partition을 사용함
또한 이때 Data Skew 발생 가능! 

 

 

 

 

 

 

 

 

 

셔플링: hashing partition

Aggregation 오퍼레이션

 

 

 

 

 

 

 

 

 

 

 

 

 

Data Skewness

Data partitioning은 데이터 처리에 병렬성을 주지만 단점도 존재

이는 데이터가 균등하게 분포하지 않는 경우
    - 주로 데이터 셔플링 후에 발생
셔플링을 최소화하는 것이 중요하고 파티션 최적화를 하는 것이 중요.