RDD Operations - Shuffle operations
24 August 2019
Spark 내의 특정 작업은 셔플이라고하는 이벤트를 트리거합니다.
셔플 (shuffle)은 데이터를 재분배하여 Spark간에 다르게 그룹화되는 Spark의 메커니즘입니다.
이것은 일반적으로 executor와 machine간에 데이터를 복사하고, 복잡하고 값비싼 작업인 shuffle을 만드는 것을 포함하고 있습니다.
Background
셔플을 하는 동안 발생하는 상황을 이해하기 위해 reduceByKey 작업의 예를 고려할 수 있습니다.
reduceByKey 작업은 단일 키의 모든 값이 튜플(키 및 해당 키와 연관된 모든 값에 대해 감소 기능을 실행 한 결과)로 결합되는 새 RDD를 생성합니다.
문제는 단일 키의 모든 값이 반드시 동일한 파티션 또는 동일한 시스템에 상주하지는 않지만 결과를 계산하려면 결국엔 함께 배치되어야한다는 것입니다.
Spark에서 데이터는 일반적으로 특정 작업에 꼭 필요한 위치에 있도록 파티션에 분산되지는 않습니다.
계산 중에 단일 파티션에서는 단일 작업만 작동하므로 하나의 reduceByKey 감소 작업을 실행하기 위해 모든 데이터를 구성하려면 Spark는 전체의 모든 작업을 수행해야합니다.
모든 키에 대한 모든 값을 찾으려면 모든 파티션에서 값을 읽은 다음, 파티션 간 값을 모아 각 키의 최종 결과를 계산해야합니다.
이를 셔플이라고합니다.
각 파티션에 대해서 각 요소의 데이터 세트가 구성되게 되면, 각 파티션 내에서 세트는 정돈되어 있을 수 있지만 전체 요소의 순서로 보면 그것은 정돈되어 있다고 볼 수는 없습니다.
셔플 이후에 예측 가능한 순서로 데이터를 원한다면 아래와 같은 방법이 있습니다.
- mapPartitions : 예를 들어 .sorted를 사용하여 각 파티션에 대해 정렬이 가능합니다.
- repartitionAndSortWithinPartitions : 파티션을 다시 진행하는 동안 동시에 파티션 sorting이 가능합니다.
- sortBy : 전체적으로 정렬된 RDD를 만들 수 있습니다.
셔플을 실행시키는 작업에는 repartition 및 coalesce과 같은 repartition 작업, groupByKey 및 reduceByKey와 ByKey 작업(counting 작업 제외) 및 cogroup 및 join과 같은 join 작업이 포함됩니다.
Performance Impact
셔플은 디스크 I/O, 데이터 직렬화 및 네트워크 I/O를 포함하므로 비용이 많이 드는 작업입니다.
셔플에 대한 데이터를 구성하기 위해 Spark는 task 집합(데이터를 구성시키는 map 작업, 데이터를 집계하는 reduce 작업)을 생성합니다.
이 명명법은 MapReduce에서 유래 한 것으로 Spark의 map과 reduce 작업과는 직접적인 연관이 없습니다.
내부적으로 개별 map의 결과는 적합하게 사용되기 전까지 memory 상에 보관됩니다.
그런 다음 대상 파티션을 기준으로 정렬되어 단일 파일에 기록됩니다.
reduce 측면에서는, task는 관련된 정렬된 block을 읽게 됩니다.
특정 셔플 작업은 메모리 내 데이터 구조를 사용하여 레코드를 전송하기 전후에 레코드를 구성하기 때문에 많은 양의 힙 메모리를 소비 할 수 있습니다.
특히, reduceByKey 및 aggregationByKey는 map 측에서 이러한 구조를 사용하고 ByKey 작업은 Reduce 측면에서 이를 생성합니다.
데이터가 메모리 사이즈에 맞지 않으면 Spark는 이러한 테이블을 디스크에 쏟아내고, 디스크 I/O의 추가 오버 헤드와 가비지 수집 증가를 야기시킬 수 있습니다.
셔플은 또한 디스크에 많은 수의 중간 파일을 생성합니다.
현재 Spark 1.3에서는 이러한 파일은 해당 RDD가 더 이상 사용되지 않고 가비지 수집될 때까지 보존됩니다.
만약 같은 데이터에 대해 다시 작업 진행이 필요할 경우, 추가 연산을 하지 않기 위해 이러한 파일들은 사용됩니다.
가비지 콜렉션은 애플리케이션이 이러한 RDD에 대한 참조를 보유하거나 GC가 자주 발생되지 않는 경우에는 오랜 시간 후에 발생할 수 있습니다.
이는 오랜 기간 실행되는 Spark 작업은 많은 디스크 공간을 소비 할 수 있음을 의미합니다.
이러한 임시 파일 디렉토리는 Spark 컨텍스트를 구성 할 때 spark.local.dir 변수에 의해 지정됩니다.
셔플 동작은 다양한 config 변수를 통해 조정할 수 있습니다.
Spark Configuration Guide의 'Shuffle Behavior'섹션을 참조하십시오.
참조 : https://spark.apache.org/docs/2.1.1/programming-guide.html#shuffle-operations