Apache Spark의 기본, RDD

Yuni·2022년 9월 12일

DE 찍먹을 위한 Apache Spark 알아보기

목록 보기

3/6

RDD 탄력적 분산 데이터셋?

Resilient Distributed Dataset RDD

from pyspark import SparkContext
sc = SparkContext(conf=conf)

데이터 추상화! 데이터는 클러스터에 흩어져 있지만 하나의 파일처럼 사용 가능
탄력적이고 불변 -> 여러 군데서 분산처리를 진행하는데 여러 노드 중 하나가 망가진다면?
- immutable : 노드1에 문제가 생기면 다시 복원 후 노드2에서 진행
  변환을 거칠때마다 연산의 기록을 남긴다.
  (e.g. RDD1이 변환을 거치면 RDD1에 수정이 일어나는게 아니라 RDD2가 생성된다.)
- RDD의 변환 과정은 하나의 DAG
탈 type (*컴파일 시 타입 판별 가능!) -> 개발자 친화적
구조적(RDB, Dataframe) / 비구조적(로그, 자연어) 데이터 모두 다룰 수 있음
Lazy Evaluation -> 결과가 필요할때까지 연산하지 않음
- 액션(A)할때까지 변환(T)하지 않는다.

Spark Operation = transform + action
Spark는 액션(action)을 할때까지 변환(transform)하지 않는다.

transformations	actions
결과값으로 새로운 RDD 반환	결과값을 연산하여 출력하거나 저장
지연실행(Lazy-execution)	즉시실행(Eagar-execution)
map() filter() distinct() groupby() orderBy()	collect() count() reduce() foreach()

Why RDD?

유연함
짧은 코드로 할 수 있는게 많음
개발할때 ‘무엇’보다 ‘어떻게’에 대해 더 생각하게 한다 (how-to)
: 게으른 연산 덕에 데이터가 어떻게 변환될지 생각하게 함

Key-Value RDD

Key와 Value 쌍을 가지는 Key-Value RDD pairs rdd

pairs = rdd.map(lambda x: (x, 1))

파티션에 대한 이해가 중요함

파티션

Spark Partitioning == 일반 프로그래밍에서 자료구조를 선택하는 것

input partition, output partition, shuffle partition이 있는데 shuffle partition이 가장 중요함

Hash 방식	Range 방식
데이터를 여러 파티션에 균일 분배하는 것	키의 순서와 집합에 따라 정렬된 파티셔닝

key-value RDD를 사용할때만 의미가 있음
파티션의 크기와 배치는 자유롭게 설정 가능하며 성능에 큰 영향을 미침
하나의 노드는 여러 개의 파티션을 가질 수 있음
RDD는 쪼개져서 여러 파티션에 저장됨

파티션 방법

디스크에서 파티션 하기 partitionBy() - transformation : persist를 해주면 더 효과적으로 활용할 수 있음
메모리에서 파티션 하기 repartition() coalesce() : 셔플링을 동반하여 매우 비싼 작업

# 새로운 파티션을 만들 수 있는 함수
join()
sort()
partitionBy()
filter()

파티션 팁

최적화를 위해 아래와 같은 규칙을 기억하면 좋음

쿼리를 최대한 groupBy로 집계한 후에 join을 수행한다.
그 후에 partition 수를 조절하고,
그래도 안된다면 Core 당 메모리를 증가시키자.

Spark Operations

병렬처리를 하기 어려운 경우 Spark를 쓰는 의미가 퇴색되므로 Operation 특징을 잘 알고 써야 그 기능을 극대화할 수 있음

Transformation

_Narrow transformation

# Narrow Transformation
filter()
map()
flatMap()
sample()
union()

1:1 변환을 진행하므로 1열을 조작하기 위해 다른 열/파티션의 데이터를 쓸 필요가 없음
정렬이 필요하지 않은 경우 수행됨. 다른 파티션을 참고하지 않음, 즉 통신이 필요 없음

_Wide transformation

# Narrow Transformation
join()
distinct()
orderBy()

Shuffling을 진행하여 통신이 필요함
아웃풋 RDD의 파티션에 다른 파티션의 데이터가 들어갈 수 있음

Action

대부분의 Action은 Reduction 임
근접하는 요소들을 모아서 하나의 결과로 만드는 일

(예외) collect()

# reduce()
rdd.reduce(<function>)

# fold()
rdd.fold(zeroValue, <function>)

# groupby()
rdd.groupBy(<function>)

# aggregate()
rdd.aggregate(zeroValue, seqOp, combOp)