Spark(8) : 데이터 처리

Summary 

데이터 처리에서 중요한 개념 : Partition, Shuffling

데이터 구조 :  RDD, DataFrame, Dataset

Spark Session 생성과 설정

DataFrame 프로그래밍

---

Spark 데이터 시스템 아키텍처

( 종합선물세트 같은 )

외부데이터 :  NOSQL, RDMS

 

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

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

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

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

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

 

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

 

HDFS 에 데이터 파일이 있을 때 Data Block 으로 나눠져 있고 Data Block 수만큼 파티션이 생기면 Spark Cluster 메모리로 로딩이 된다. 예를 들어 4개의 Data Block으로 나눠져 있을 것 

Spark Cluster에 CPU 코어가 1개 씩 있다고 가정하면, 동시에 실행가는 최대 개수가 2개가 된다. 

 

적절한 파티션의 수 = Executor의 수 * Executor당 CPU의 수

병렬성 최대화

 

Spark 데이터 처리 흐름

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

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

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

• sort, group by, filter, map, join

 

파티션간에 데이터 이동없이 계속 변환이 가능할까? - 불가능

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

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

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

셔플링이 발생할 때 네트워크를 타고 데이터가 이동하게 됨

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

셔플링 : Hashing partition

Aggregation 오퍼레이션

Data Skewness

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

이는 데이터가 균등하게 분포하지 않는 경우 - 주로 데이터 셔플링 후에 발생

셔플링을 최소화하는 것이 중요하고 파티션 최적화를 하는 것이 중요