01_08. Shuffling & Partitioning

Shuffling 

• 그룹핑시 데이터를 한 노드에서 다른 노드로 옮길 때 사용한다.
• 성능을 (많이) 저하시킨다.

✅ groupByKey를 할 때도 발생하는 Shuffling

여러 노드에서 데이터를 주고받게 됨 (네트워크 연산)

• key를 기준으로 연산을 하며, 데이터가 이리저리 섞이기 때문에 통신을 많이 하게 된다.
• 여러 네트워크 연산을 일으키기 때문에 연산 코스트가 큰 연산이다.

✅ 통신 속도

메모리 > 디스크 > 네트워크

✅ Shuffle을 일으킬 수 있는 작업들

• Join, leftOuterJoin, rightOuterJoin
• GroupByKey
• ReduceByKey
• ComebineByKey
• Distinct
• Intersection
• Repartition
• Coalesce

→ 위의 함수들이 결과로 나온 RDD를 원본 RDD의 다른 요소를 참조하거나, 다른 RDD를 참조하여 Shuffle을 일으킴

Partition

✅ 예제1

👎 GroupByKeys + Reduce 

• groupByKeys를 한 후 리듀스를 하게 되면 이미 키값들이 여러개인 상태에서 groupByKeys를 하기 때문에 통신 연산이 늘어나게된다.

👍 ReduceByKey 

• 따라서, 미리 reduce를 하여 통신이 일어나는 횟수를 줄이도록 하는 함수인 reducebykey를 사용하는 것이 좋다.

✅ Shuffle을 최소화 하려면

• 미리 파티션을 만들어 두고 캐싱 후 reduceByKey 실행
• 미리 파티션을 만들어 두고 캐싱 후 Join 실행
• 둘다 파티션과 캐싱을 조합해서 최대한 로컬 환경에서 연산이 실행되도록 하는 방식
  • local = 각각의 파티션에서 실행한다는 뜻

→ 셔플을 최소화하여 10배의 성능 향상 가능

✅ 예제2

GroupByKey vs ReduceByKey 

• 둘의 결과는 같지만 성능은 다르다.
• textRDD에 긴 텍스트가 들어았다고 가정한다.
• GroupByKey는 네트워크를 통해 전송되는 양이 많아질 뿐만 아니라 out of disk 문제가 발생할 수도 있다.
• 셔플은 기본적으로 비용이 큰 연산이므로 reduceByKey로 대체 가능하다.

✅ Partition은 어떻게 결정될까?

목적

• 데이터를 최대한 균일하게 퍼트리고, 쿼리가 같이 되는 데이터를 최대한 옆에 두어 검색 성능을 향상시키는 것

특징

• RDD는 쪼개져서 여러 파티션에 저장됨
• 하나의 파티션은 하나의 노드 (서버)에
• 하나의 노드는 여러개의 파티션을 가질 수 있음
• 파티션의 크기와 배치는 자유롭게 설정 가능하며 성능에 큰 영향을 미침
  • 데이터 찾기가 쉬워져서
• Key-Value RDD를 사용할 때만 의미가 있다.

스파크의 파티셔닝 == 일반 프로그래밍에서 자료구조를 선택하는 것

✅ Partition의 종류

• Hash Partitioning
• Range Partitioning

*️⃣ Hash Partitioning

• 데이터를 여러 파티션에 균일하게 분배하는 방식
• 파이썬의 dictionary와 같다.
• key값을 hash function에 넣어 해시값으로 만들어 해시값에 해당되는 buckets에 데이터를 저장한다.

[극단적인 예] 2개의 파티션이 있는 상황에서:

• 짝수의 Key만 있는 데이터셋에 Hash 함수가 (x % 2) 라면?
• 한쪽 파티션만 사용하게 될 것

=> 데이터 셋과 hash function의 성질을 파악하는 것이 중요하다.

*️⃣ Range Partitioning

• 순서가 있는, 정렬된 파티셔닝
  • 키의 순서에 따라
  • 키의 집합의 순서에 따라
• 서비스의 쿼리 패턴이 날짜 위주면 일별 Range Partition 고려

✅ Partition의 종류

• 디스크에서 파티션하기
  • partitionBy()
• 메모리에서 파티션하기
  • repartition()
  • coalesce()

*️⃣ 디스크에서 파티션하기

• 사용자가 지정한 파티션을 가지는 RDD를 생성하는 함수
  • partitionBy()
  • 파티션을 만든 후엔 persist()하지 않으면, 다음 연산에 불릴때마다 반복하게 된다. (셔플링이 반복적으로 일어난다.)

• partitionBy()는 transformations 중 하나라 바로 실행되지 않는다.
• glom()은 파티션까지 다 보여주는 transformations이다.

*️⃣ Repartition & Coalesce

• 둘 다 파티션의 갯수를 조절하는데 사용
• 둘 다 shuffling을 동반하여 매우 비싼 작업
• Repartition: 파티션의 크기를 줄이거나 늘리는데 사용됨
• Clalesce: 파티션의 크기를 줄이는데 사용됨

✅ 연산 중 Partition을 만드는 작업들

• Join, leftOuterJoin, rightOuterJoin
• groupByKey
• reduceByKey
• foldByKey
• foldByKey
• Sort
• mapValues (parent)
  • parent: 위에서 파티션이 정의되어 있으면 그대로 사용
• flatMapValues (parent)
• filter (parent)
• 등등

map vs mapValues
flatMap vs flatMapValues

=> map과 flatMap의 경우 Key의 변형이 가능하기 때문에 연산 중 새로운 파티션을 만들지 않는다. (key가 변경되는 경우 파티션이 망가지기 때문)