[DATA] Spark Shuffle

Spark에서 Shuffle은 데이터 처리의 중요한 개념입니다. Shuffle이란 데이터의 재분배를 의미하며, 주로 Wide Transformation에서 발생합니다. Wide Transformation은 데이터의 구조를 변경하는 연산으로, 예를 들어 `groupByKey`와 `reduceByKey`가 있습니다. 이 두 연산은 데이터의 재구성을 필요로 하며, 이 과정에서 Shuffle이 발생합니다. 

`groupByKey`는 모든 데이터를 한 곳으로 모으는 방식으로, 데이터의 양이 많을 경우 성능 저하를 초래할 수 있습니다. 반면, `reduceByKey`는 데이터를 미리 로컬에서 합친 후 Shuffle을 수행하기 때문에 더 효율적입니다. 이러한 차이로 인해, 데이터 처리 시 어떤 연산을 선택하는지가 성능에 큰 영향을 미칩니다.

Shuffle이 발생하는 이유는 데이터의 분산 처리와 관련이 깊습니다. Spark는 데이터를 여러 파티션으로 나누어 처리하는데, 이 과정에서 데이터의 재분배가 필요하게 됩니다. 이때 발생하는 Shuffle은 디스크 I/O, 네트워크 대역폭, 직렬화 등의 병목 현상을 초래할 수 있습니다. 

Spark의 Shuffle 단계가 병목이 되는 이유는 여러 가지가 있습니다. 첫째, 디스크 I/O는 Shuffle 과정에서 가장 큰 성능 저하 요인 중 하나입니다. 데이터가 디스크에 쓰여지고 다시 읽혀야 하므로, 이 과정에서 시간이 소요됩니다. 둘째, 네트워크 대역폭도 중요한 요소입니다. Shuffle 과정에서 데이터가 여러 노드 간에 전송되기 때문에, 네트워크 속도가 느리면 성능이 저하됩니다. 셋째, 직렬화 과정도 무시할 수 없습니다. 데이터가 네트워크를 통해 전송되기 전에 직렬화되어야 하며, 이 과정에서 추가적인 시간이 소요됩니다.

Shuffle과 관련된 주요 설정 파라미터를 조정하는 것도 성능 개선에 큰 도움이 됩니다. 예를 들어, `spark.sql.shuffle.partitions`는 Shuffle Partition의 수를 조정하는 파라미터입니다. 일반적으로 이 값은 100~200MB 정도로 설정하는 것이 좋습니다. 또한, `spark.reducer.maxSizeInFlight`는 한 번에 전송할 수 있는 데이터의 최대 크기를 설정하며, 이 값을 적절히 조정하면 네트워크 대역폭을 효율적으로 사용할 수 있습니다. 마지막으로, `spark.shuffle.file.buffer`는 Shuffle 파일을 버퍼링하는 크기를 설정하는 파라미터로, 이 값을 조정하여 디스크 I/O 성능을 개선할 수 있습니다.

실행계획을 통해 Shuffle 발생 위치를 확인하는 방법도 있습니다. Spark에서는 `explain()` 메서드를 사용하여 실행계획을 확인할 수 있으며, Spark UI의 DAG(Directed Acyclic Graph)에서도 Shuffle이 발생하는 위치를 시각적으로 확인할 수 있습니다. 이를 통해 어떤 연산이 Shuffle을 유발하는지 파악하고, 필요한 경우 최적화할 수 있습니다.

마지막으로, 실제 튜닝 사례를 살펴보겠습니다. 예를 들어, 한 사용자가 10분 걸리던 작업을 Shuffle 최적화를 통해 2분으로 단축한 사례가 있습니다. 이 사용자는 `spark.sql.shuffle.partitions` 값을 조정하고, `reduceByKey`를 사용하여 데이터의 로컬 합산을 통해 Shuffle을 최소화했습니다. 이러한 최적화는 성능을 크게 향상시킬 수 있습니다.

이처럼 Spark의 Shuffle을 이해하고 최적화하는 것은 데이터 처리 성능을 높이는 데 매우 중요합니다. Shuffle의 발생 원인과 병목 현상, 주요 설정 파라미터를 잘 이해하고 활용하면, 더 효율적인 데이터 처리가 가능해집니다.

 데이터 처리에서의 두 가지 조인 방식: Broadcast join과 Shuffle join을 비교한 이미지입니다. 

 Spark의 JVM Heap Memory 구조

 데이터 파티셔닝

 Sort-merge join과 Bucket join