[Apache Spark] 스파크의 셔플(Shuffle)에 대하여

1. Spark에서 Shuffle에 대해서 

Apache Spark에서 셔플 작업은 일반적으로 데이터 그룹화 또는 집계와 같은 후속 처리 단계를 준비하기 위해 클러스터의 노드 전체에 데이터를 재분배하는 데 사용됩니다. 여기에는 데이터를 더 작은 청크로 분할하고, 네트워크 전체에서 데이터를 섞은 다음, 데이터를 새로운 청크 집합으로 다시 분할하는 작업이 포함됩니다. 셔플 작업은 Spark에서 가장 비용이 많이 드는 작업 중 하나이며 성능에 상당한 영향을 미칠 수 있습니다.

Apache Spark에는 두 가지 주요 유형의 셔플이 있습니다:

• Map-side 셔플: 이것은 네트워크를 통해 데이터를 전송하기 전에 단일 노드 내에서 데이터를 셔플 하는 것을 포함합니다. 이는 특히 데이터가 이미 분할되어 있거나 데이터의 일부만 셔플해야 하는 경우 전체 셔플보다 더 효율적일 수 있습니다.
• Reduce-side 셔플: 네트워크를 통해 다른 노드로 데이터를 셔플한 다음 추가 처리를 위해 데이터를 다시 파티셔닝 하는 방식입니다. 일반적으로 데이터가 너무 커서 단일 노드에 맞지 않거나 특정 방식으로 데이터를 그룹화하거나 집계해야 할 때 사용됩니다.

 

 예를 들어

Scala에서 shuffle은 Apache Spark에서 RDD(Resilient Distributed Dataset)의 요소를 새로운 RDD에 무작위로 배포하는 데 일반적으로 사용되는 기능입니다. 셔플 기능은 RDD의 각 요소가 키-값 쌍으로 변환되는 맵 단계와 각 키의 값이 결합되는 리듀스 단계의 두 단계로 수행됩니다.

 

Scala에서 shuffle 함수가 작동하는 방식을 보여주는 예제 코드 스니펫:

val data = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
val mappedData = data.map(x => (x % 2, x)) // Map step
val reducedData = mappedData.reduceByKey((x, y) => x + y) // Reduce step

결과 RDD인 맵데이터에는 (0, 2), (1, 1), (0, 4), (1, 3), (0, 6), (1, 5), (0, 8), (1, 7), (0, 10), (1, 9) 등 5개의 키-값 쌍이 포함됩니다.

다음으로, reduceByKey 함수를 사용하여 각 키의 모든 값을 그룹화하고 값에 reduce 함수를 적용합니다. 이 예제에서 reduce 함수는 단순히 값을 더하기만 합니다. 이것이 셔플의 축소 단계입니다.

결과 RDD인 reducedData에는 (0, 30) 및 (1, 25)라는 두 개의 키-값 쌍이 포함됩니다.

reduceByKey 함수는 함수를 인수로 받는 고차 함수라는 점에 유의하세요. 이 함수는 각 키의 값을 결합하는 데 사용됩니다. 이 예에서 함수는 (x, y) => x + y이며, 단순히 값을 더합니다.

셔플 연산은 계산 및 네트워크 I/O 측면에서 비용이 많이 드는 연산이라는 점에 유의해야 합니다. 따라서 성능을 최적화하려면 스파크 프로그램에서 수행되는 셔플 횟수를 최소화하는 것이 좋습니다.

 

2. Shuffle이 발생되는 function들은? 

 

Apache Spark에서 셔플 작업은 클러스터의 파티션 또는 노드 간에 데이터를 재분배하는 비용이 많이 드는 작업으로, 상당한 성능 오버헤드를 초래할 수 있습니다. Apache Spark에서 셔플을 유발할 수 있는 몇 가지 작업이 있습니다:

 

• groupByKey: 이 연산은 RDD의 각 키에 대한 값을 그룹화하므로 셔플이 필요합니다. 이 작업은 특히 키에 대한 데이터가 여러 파티션에 분산되어 있는 경우 비효율적일 수 있습니다.
• reduceByKey: 이 연산은 RDD의 각 키 값에 감소 함수를 적용하며 각 키의 값을 결합하기 위해 셔플이 필요합니다. groupByKey와 달리 reduceByKey는 데이터를 셔플 하기 전에 각 파티션에 로컬로 reduce 함수를 적용하므로 더 효율적일 수 있습니다.
• aggregateByKey: 이 연산은 RDD의 각 키 값에 집계 함수를 적용하고 각 키의 값을 결합하기 위해 셔플이 필요합니다. reduceByKey와 마찬가지로 aggregateByKey는 데이터를 셔플하기 전에 각 파티션에 로컬로 집계 함수를 적용합니다.
• sortByKey: 이 작업은 RDD의 요소를 키별로 정렬하는 작업으로, 데이터를 키별로 그룹화한 다음 정렬하기 위해 셔플이 필요합니다.
• join: 이 작업은 키를 기준으로 두 개의 RDD를 조인하며, 각 키의 데이터를 한데 모으기 위해 셔플이 필요합니다. 이 작업은 셔플 오버헤드 측면에서 가장 비용이 많이 드는 작업입니다.
• cogroup: 이 작업은 키를 기준으로 여러 RDD의 값을 그룹화하며, 셔플이 필요합니다.
• distinct: 이 작업은 RDD에서 중복을 제거하며, 데이터를 키별로 그룹화한 다음 중복을 제거하기 위해 셔플이 필요합니다.
• repartition: 이 작업은 RDD의 데이터를 지정된 수의 파티션으로 재분배하며, 이 작업에는 셔플이 필요할 수 있습니다.
• zip: 이 작업은 길이가 같은 두 개의 RDD에서 요소 쌍을 생성하며, 동일한 인덱스를 가진 요소에서 쌍이 생성되도록 셔플이 필요합니다.
• coalesce: 이 작업은 RDD의 파티션 수를 줄이며, 파티션 수를 늘리는 경우 셔플이 필요할 수 있습니다.

 

셔플을 유발하는 연산을 염두에 두고 신중하게 사용하여 성능 오버헤드를 최소화하는 것이 중요합니다. 셔플 성능을 최적화하는 한 가지 일반적인 접근 방식은 groupByKey 대신 reduceByKey 및 aggregateByKey와 같은 연산을 사용하여 보다 효율적인 셔플을 수행하는 것입니다. 또한 파티션 수, 셔플에 사용되는 메모리 양, 네트워크 대역폭을 적절히 구성하면 셔플 성능을 개선할 수 있습니다.

 

3. Shuffle 발생 시 유의사항 

 

Apache Spark에서 셔플 작업이 발생하면 애플리케이션 성능에 상당한 영향을 미칠 수 있습니다. 셔플이 애플리케이션 성능에 미치는 영향을 최소화하기 위해 고려해야 할 몇 가지 요소가 있습니다.

 

1. Partitioning: 파티션 수는 셔플 성능에 영향을 미칠 수 있습니다. 데이터가 잘 분할되지 않으면 셔플링이 효율적으로 발생하지 않을 수 있습니다. 데이터 크기, 클러스터 구성 및 사용 가능한 메모리 양을 기반으로 파티션 수를 신중하게 선택하는 것이 중요합니다.
2. Memory: 사용 가능한 메모리 양은 또한 임팩트 셔플 성능. Spark는 사용 가능한 메모리를 활용하여 디스크 I/O 양을 줄일 수 있는 메모리 기반 셔플 알고리즘을 사용합니다. 따라서 셔플에 사용할 수 있는 메모리 양을 늘리면 성능이 향상될 수 있습니다.
3. Serialization: 직렬화는 데이터를 네트워크를 통해 전송할 수 있는 형식으로 변환하는 프로세스입니다. 올바른 직렬화 형식을 선택하면 셔플 성능에 영향을 미칠 수 있습니다.
4. Network: 네트워크 대역폭과 대기 시간은 셔플 작업의 성능에 영향을 미칠 수 있습니다. 셔플 되는 데이터의 양을 처리할 수 있도록 네트워크를 적절하게 구성하는 것이 중요합니다.
5. Configuration: Spark에는 셔플 성능을 최적화하기 위해 조정할 수 있는 여러 구성 매개변수가 있습니다. 예를 들어 spark.shuffle.compress를 true로 설정하면 셔플 단계에서 데이터를 압축하여 네트워크 I/O의 양을 줄일 수 있습니다.

 

4. Shuffle 최적화 방법은?

 

Apache Spark의 셔플 작업은 특히 대규모 데이터 세트를 처리할 때 성능 문제 및 속도 저하를 유발할 수 있습니다. 다음은 셔플로 인해 발생하는 문제를 해결하기 위한 몇 가지 접근 방식입니다.

 

1. 구성 매개변수 조정: Apache Spark는 셔플 성능을 최적화하기 위해 조정할 수 있는 여러 구성 매개변수(예: 셔플에 사용되는 메모리, 리듀서 수, 사용되는 직렬화 형식. 이러한 매개변수를 조정하면 성능이 향상되고 셔플의 영향을 줄일 수 있습니다.
2. 데이터 크기 줄이기: 셔플이 성능 문제를 일으키는 경우 데이터 양을 줄이는 방법을 모색할 가치가 있습니다. 그것은 셔플 해야 합니다. 한 가지 접근 방식은 필요한 데이터만 처리되도록 섞기 전에 데이터를 필터링하는 것입니다. 또 다른 접근 방식은 섞기 전에 데이터를 미리 집계하여 섞을 데이터를 줄이는 것입니다.
3. 데이터 다시 분할: 경우에 따라 데이터를 다시 분할하면 데이터를 다시 분할할 수 있습니다. 셔플 성능을 향상합니다. 여기에는 데이터 크기 및 클러스터 구성과 일치하도록 파티션 수를 늘리거나 줄이는 작업이 포함됩니다. 데이터 크기와 사용 가능한 리소스를 기반으로 최적의 파티션 수를 선택하는 것이 중요합니다.
4. 적절한 데이터 구조 사용: 적절한 데이터 구조를 사용하면 데이터 양을 줄이는 데 도움이 될 수 있습니다. 그것은 셔플 해야 합니다. 예를 들어 브로드캐스트 변수를 사용하면 노드 간에 전송해야 하는 데이터의 양을 줄이는 데 도움이 될 수 있습니다.
5. 적절한 API 사용: Apache Spark는 다음과 같이 설계된 여러 API를 제공합니다. 셔플의 영향을 최소화합니다. 예를 들어 reduceByKey 및 aggregateByKey API는 셔플해야 하는 데이터의 양을 줄일 수 있는 반면 mapPartitions API는 처리를 통해 성능을 향상할 수 있습니다. 데이터를 일괄 처리합니다.

 

 

References:

1. "Shuffle Operations" in the Apache Spark documentation: https://spark.apache.org/docs/latest/rdd-programming-guide.html#shuffle-operations
2. "Understanding Apache Spark Shuffle" by Jitendra Yadav: https://towardsdatascience.com/understanding-apache-spark-shuffle-37ce38efba67
3. "Apache Spark Shuffling Performance Tuning" by Tomasz Drabas: https://databricks.com/blog/2015/06/22/understanding-your-spark-application-through-visualization.html
4. "Performance Tuning Apache Spark" by Holden Karau: This book provides an overview of Apache Spark and includes a chapter on optimizing shuffle performance.
5. "Apache Spark Shuffling Performance Tuning" by Tomasz Drabas: This blog post provides a detailed overview of shuffle performance tuning in Spark and includes recommendations for tuning specific configuration parameters.
6. "Optimizing Apache Spark Shuffle Performance" by Databricks: This blog post provides a detailed overview of how to optimize shuffle performance in Spark, including best practices for configuration, partitioning, and serialization.