[HP-Spark] 키/값 데이터로 작업하기

키/값 데이터로 작업하기

스파크는 자체적으로 튜플을 키/값 기반으로 RDD에 쓸수 있도록 구성된 함수들의 클래스 PairRDDFunctions 제공
- PairRDDFunctions 클래슨는 암묵적 변환을 통해 사용 가능
  - RDD[(K,V)] 형태로 만들면 PairRDDFunctions의 함수들을 쓸 수 있음
- 조인과 집ㄱ볘 연산에 대한 메서도도 있음
OrderedRDDFunctions 클래스는 정렬과 관련된 메서드 지원
- 튜플의 RDD에서 첫번 쩨 아이템이 키가 되고 내부적으로 순서가 암묵적으로 정의된 타입이라면 사용가능
  - 숫자형이나 문자열
키/값 연산은 성능 이슈를 야기할 수 있음
1. 드라이버 Out-Of-Memory
2. 이규제큐터 노드에서의 메모리 부족
3. 셔플 실패
4. 연산이 특히 느린 ‘뒤쳐진 태스크’나 파티션
드라이버의 OOM은 대개 액션에 의해 발생하고
나머지 3가지는 PairRDDFunctions이나 OrderedRDDFunctions 클래스의 넓은 트랜스포머과 연계된 셔플에서 발생
셔플과 관련된 성능 문제를 위해 주요 두 테크닛이 있음
1. 더 적은 셔플
2. 더 나은 셔플

더 적은 셔플
- 셔플 횟수를 최소화하는 기법
  1. 트랜스포메이션에서 파티셔닝 정보 보존
    연산에서 셔플링 횟수를 줄이기 위한 방법은 데이터가 다시 셔플을 방지하기 위한 ‘파티셔닝 상태 유지’
  2. 공동 그룹화 및 공존하는 RDD와 파티셔닝된 RDD활용하기
    조인에서의 셔플링을 피하고 여러 번의 넓은 트랜스포메이션을 연산하기 위한 셔플링 횟수를 줄일 수 있음
  3. 사용자 파티셔닝
    후속 연산들을 위해 데이터를 효과적으로 분산시킬 수 있는 파티셔너를 자체 제작하는 방법
  4. 보조 정렬과 repartitionAndSortWithinPartitions
    복잡한 연산을 더 효율적으로 처리할 수 있도록 연산 작업을 셔플 스테이지에 어떻게 집어넣을 것인가
더 나은 셔플
- 떄로는 셔플 없이 완료할 수 없는 연산이 있으나 모든 넓은 트랜스포메이션과 셔플이 모두 고비용이거나 실패하기 쉬운 것으 아님
  1. groupByKey 함수는 왜 위험한가 및 집계연산에서 메모리 부족 오류 막ㄱ;
    reduceByKey OR aggregateByKey
    
    집계 연산 종류를 위해서, 맵 사이드에서의 리듀스를 수행(컴바이너처럼), 키와 연관된 레코드를 모두 메모리에 올릴 필요없는 방법을 써서
    
    이그제큐터의 메모리 오류를 막고 넓은 트랜스포메이션의 속도를 향상 시킬 수 있음
  2. 뒤처지는 작업 감지와 균형이 맞지 않은 데이터
    키에 따른 레코드 비율이 균등하게 분배된 데이터나 중복되지 않은 키의 비율이 높은 데이터를 셔플하는 것은 이그제규터의 메모리 오류와, 뒤쳐지는 태스트 방지 가능

PairRDDFunctions과 OrderedRDDFunctions의 사용법

스파크 RDD 클래스는 스칼라의 implicit를 쓰고 있으며 PairRDDFunctions은 (K,V) 타입을 가진 모든 RDD에서 사용 가능
PairRDDFunctions은 K,V은 아무타입이나 상관없음
- OrderedRDDFunctions에 대해서는 K가 순서를 가질 수 있는 타입
- 숫자, 문자열 대신 임의의 타입을 사용하려면 직접 순서를 정의해야함
- PairRDD나 OrderedRDD 타입으로 변환하는 요구사항을 만족하기 위해 암묵적변환을 사용함

키/값 쌍의 액션들

countByKey, countByValue, lookUp, collectAsMap
countByKey는 각 키마다 데이터 리턴, 단일 키 종류가 많다면 메모리 에러 가능성이 있음
lookUp은 한 키에 대한 모든 값을 리턴
- lookUp은 등록된 파티셔너가 없으면 셔플을 발생시키기 때문에 고비용 연산
키/값 디자인 방법
- 값은 최소한 키의 분포 비율에 따라 잘 분배되어야함
- 각 키에는 개별 이그제큐터의 메모리 이상 레코드가 들어지 않게 하는 것이 좋음
키/값 트랜스포메이션은 하나의 키와 관련된 모든 데이터가 한 파티션 내에서 메모리에 유지되어야하는 경우
이그제큐터 메모리 오류 발생할 수 있음

집계연산 선택하기

대부분 combineByKey 위에서 구축되어있지만 각자 성능에 있어서는 큰 차이가 남

1. groupByKey

목적
- 동일한 키를 가지는 값들을 그룹지어 하나의 반복자를 넣음
제약
- 기본 HashPartitioner로는 배열 키를 가질 수 없어, 직접 파티셔너를 만들어야함
메모리가 부족한 상황
- 단일 키와 관련된 모든 레코드가 하나의 이그제큐터에서 메모리로 읽어 들이기 많을 때
느려지는 상황
- 알려진 파티셔너가 아니면 셔플이 발생
- 단일 키 개수가 많거나, 키 당 레코드 개수가 많거나, 레코드가 키별로 균등하게 배포되지 않을 수 록 비쌈
결과 파티셔너
- default : HashPartitioner

2. combineByKey

목적
- 동일한 키에 대한 값들을 다른 결과타입으로 써서 하나로 합칠 경우
제약
- 기본 HashPartitioner로는 배열 키를 가질 수 없어, 직접 파티셔너를 만들어야함 (동일)
메모리가 부족한 상황
- CombineBy 가 메모리를 너무 많이 쓰거나 많은 GC오버헤드를 일으키거나, 한 키에 대해 레코드가 너무 많을 때
느려지는 상황
- groupByKey와 비슷
- 하지만 연산이 리듀싱 종류라면 groupbyKey보다는 빠를 수 있음
결과 파티셔너
- default : HashPartitioner

3. aggregateByKey

목적
- 동일한 키에 대한 값들을 다른 결과타입으로 써서 하나로 합칠 경우
- 모든 누적 연산에 대해 제로값을 적용해서 시작
제약
- 기본 HashPartitioner로는 배열 키를 가질 수 없어, 직접 파티셔너를 만들어야함 (동일)
- 객체 생성을 줄여, 객체 재사용 지원하여 combineByKey보다는 저렴
메모리가 부족한 상황
- combineByKey와 동일
느려지는 상황
- combineByKey와 비슷
- 하지만 보내지 전에 맵사이드에서 병합하므로 combineByKey보다는 빠를 수 있음
결과 파티셔너
- default : HashPartitioner

4. reduceByKey

목적
- 동일한 키에 대해 모든 값을 합침
- 결과가 원래의 값들과 동일한 타입이어야만 함
제약
- 기본 HashPartitioner로는 배열 키를 가질 수 없어, 직접 파티셔너를 만들어야함 (동일)
- 객체 생성을 줄여, 객체 재사용 지원하여 combineByKey보다는 저렴
메모리가 부족한 상황
- combineByKey와 동일
- 타입 제한이 메모리 에러를 일으키지는 않음
- 컬렉션 타입이 아닌 이상 콤바인 함수에서 리듀싱이 일어남
- 추가적인 누적 객체를 만들지 않으므로 GC도 aggregateByKey보다 적음
느려지는 상황
- combineByKey와 비슷
결과 파티셔너
- default : HashPartitioner

5. foldByKey

목적
- 동일한 키에 대해 모든 값을 하나의 결합 함수오와 제로값을 사용해 합침
- 제로값이 결과에 여러번 더해질 수 있고, 기존값에 제로값이 존재한다면 reduceByKey사용할 것
제약
- 기본 HashPartitioner로는 배열 키를 가질 수 없어, 직접 파티셔너를 만들어야함 (동일)
- 객체 생성을 줄여, 객체 재사용 지원하여 combineByKey보다는 저렴
메모리가 부족한 상황
- reduceByKey와 동일
느려지는 상황
- reduceByKey와 비슷
결과 파티셔너
- default : HashPartitioner