Apache Spark에서 데이터 불균형(Data Skew) 이해하기: 느린 데이터 파이프라인의 숨겨진 원인

Apache Spark를 사용하여 2TB 규모의 이커머스 데이터를 처리하는 작업을 실행한다고 상상해 보세요.

클러스터를 200개의 executor로 구성했습니다. 코드는 올바르고, 클러스터 리소스도 충분합니다. 대부분의 작업은 몇 분 안에 완료됩니다.

하지만 하나의 작업이 계속 실행되고 있습니다.

5분이 지납니다.

10분이 지납니다.

30분이 지난 후에도 그 단일 작업은 여전히 실행 중이며, 클러스터의 대부분은 유휴 상태입니다.

executor 수를 200개에서 300개로 늘려봅니다.

아무것도 변하지 않습니다.

메모리를 늘립니다.

여전히 아무 변화가 없습니다.

결국 파이프라인은 완료되지만, 예상보다 몇 시간이나 더 걸립니다.

무엇이 잘못된 것일까요?

가능한 한 가지 이유는 데이터 불균형(Data Skew)입니다.

Illustration: How data skew can cause one Spark task to become a bottleneck while other tasks finish early.

그리고 Apache Spark를 다룬다면, 이 개념을 이해하는 것이 느린 파이프라인을 디버깅하는 방식을 근본적으로 바꿀 수 있습니다.

1. 먼저, Spark가 어떻게 작업을 분산하는지 이해하기

Apache Spark는 데이터를 파티션(partitions)으로 나누어 병렬로 처리합니다.

한 회사에서 백만 건의 고객 주문을 처리해야 한다고 생각해 보세요.

직원 한 명에게 모든 주문 처리를 맡기는 대신, 회사는 작업을 100명의 직원들에게 분산합니다.

모든 직원이 약 10,000건의 주문을 받는다면 작업 부하가 균형 잡힙니다.

하지만 만약 한 직원에게 70만 건의 주문이 주어지고 나머지 99명의 직원이 단지 30만 건만을 나누어 가진다고 상상해 보세요.

99명의 직원이 일찍 끝내더라도, 회사는 과부하가 걸린 직원이 작업을 완료할 때까지 기다려야 합니다.

분산 컴퓨팅도 이와 유사하게 작동합니다.

Spark는 파티션을 작업(tasks)에 분산하고, 작업은 사용 가능한 executor 리소스에서 실행됩니다.

병렬성은 작업이 합리적으로 잘 분산될 때만 도움이 됩니다.

executor가 더 많다고 해서 작업 부하가 자동으로 균형 잡힌 것을 의미하지는 않습니다.

1. 데이터 불균형(Data Skew)이란 정확히 무엇인가요?

데이터 불균형(Data skew)은 데이터가 파티션에 고르게 분포되지 않아 일부 태스크가 다른 태스크보다 훨씬 많은 데이터를 처리하게 만들 때 발생합니다.

간단한 예시를 들어보겠습니다.

어떤 이커머스 회사가 1억 건의 주문 기록을 가지고 있다고 가정해 봅시다.

이 회사는 각 판매자별 총매출액을 계산하고 싶습니다.

간소화된 데이터셋은 다음과 같습니다:

Seller A 혼자서 전체 주문의 70%를 차지합니다.

Spark가 seller_id를 기준으로 샤플(shuffle) 기반 집계를 수행한다면, 주어진 판매자에 대한 모든 기록은 일반적으로 해당 집계 단계로 모여야 합니다.

Seller A를 담당하는 파티션은 더 작은 판매자를 처리하는 파티션보다 훨씬 큰 워크로드(workload)를 받을 수 있습니다.

그 결과는 무엇일까요?

한 태스크가 수천만 건의 기록을 처리하는 동안 다른 태스크들은 빠르게 완료됩니다.

이것이 바로 데이터 불균형의 실제 영향입니다.

중요한 구별점: 빈번하게 나타나는 값(frequent value)이 모든 Spark 작업에서 자동으로 성능 문제를 일으키는 것은 아닙니다. 그 영향은 작업, 파티셔닝 전략, 데이터 볼륨 및 실행 계획에 따라 달라집니다.

1. 더 많은 Executor를 추가하는 것이 반드시 도움이 되지 않는 이유

많은 엔지니어들이 여기서 값비싼 실수를 저지릅니다.

그들은 느린 Spark 작업을 보고 즉시 클러스터 리소스를 늘립니다.

하지만 Spark가 키(key)별로 데이터를 샤플할 때 무슨 일이 일어나는지 고려해 보세요.

만약 7천만 건의 기록이 같은 판매자에게 속한다면, 단순히 샤플 파티션 수를 늘린다고 해서 그 판매자의 기록이 반드시 모든 파티션에 분산되는 것은 아닙니다.

표준 집계(aggregation)나 조인(join)의 경우, 동일한 키를 가진 기록은 여전히 특정 단계에서 함께 처리되어야 할 수 있습니다.

Executor를 추가하는 것은 사용 가능한 컴퓨팅 용량을 늘릴 수는 있지만, 알고리즘적 병목 현상(algorithmic bottleneck)을 자동으로 제거할 수는 없습니다.

클러스터를 확장하기 전에 다음 질문을 해보세요:

• 소수의 키가 대부분의 레코드를 담당하나요?
• 일부 작업이 다른 작업보다 훨씬 많은 데이터를 처리하고 있나요?
• 병목 현상이 스큐(skew), 메모리 부족, 디스크 스필링 때문인가요, 아니면 다른 원인 때문인가요?
• 어떤 연산이 비용이 많이 드는 셔플(shuffle)을 유발하나요?

Spark UI가 이러한 조사를 시작하기에 가장 좋은 장소인 경우가 많습니다.

1. 프로덕션 환경에서 데이터 스큐를 감지하는 방법은 무엇인가요?

느린 파이프라인을 디버깅할 때, 다음 신호들을 조사합니다.

A. 작업 지속 시간 (Task duration)
같은 스테이지 내의 작업 실행 시간을 비교해 보세요.
대부분의 작업이 20초 만에 완료되지만 몇몇 작업이 15분이 걸린다면, 그 이유를 조사해야 합니다.
지속 시간이 불균형하다는 것만으로는 스큐를 증명하지 못하지만, 유용한 신호입니다.

B. 셔플 읽기 크기 (Shuffle read size)
각 작업이 얼마나 많은 셔플 데이터를 읽는지 확인합니다.
소수의 작업이 나머지 작업보다 상당히 더 많은 데이터를 읽는다면, 스큐가 병목 현상에 기여하고 있을 수 있습니다.

C. 메모리 및 디스크 스필링 (Memory and disk spilling)
큰 파티션은 상당한 메모리 압박을 일으킬 수 있습니다. Spark는 중간 데이터를 디스크로 스필(spill)할 수 있으며, 이는 실행 시간을 증가시킵니다.

D. 데이터 자체 (The data itself)
중요한 조인 및 집계 키에 대해 빈도 분석(frequency analysis)을 수행합니다.
예를 들어:

SELECT
seller_id,
COUNT(*) AS order_count
FROM orders
GROUP BY seller_id
ORDER BY order_count DESC
LIMIT 20;

이 쿼리는 비정상적으로 빈번한 판매자 ID를 식별하는 데 도움이 됩니다.

하지만, 키의 빈도만으로는 진단의 일부일 뿐입니다. 워크로드가 실제로 어떻게 분산되는지 이해하려면 물리적 실행 계획(physical execution plan)과 Spark UI도 검사해야 합니다.

1. 데이터 스큐를 처리하는 네 가지 방법
   모든 워크로드에 적용되는 단 하나의 해결책은 없습니다. 올바른 접근 방식은 스큐가 조인, 집계 또는 다른 셔플이 많은 연산 중 어느 시점에 발생하는지에 따라 달라집니다.

테크닉 1: 브로드캐스트 조인 (Broadcast Join)
만약 다음과 같은 상황이라고 가정해 봅시다:

• 5억 개의 행을 포함하는 orders 테이블
• 단지 20,000개의 행만을 포함하는 seller 차원(dimension) 테이블

더 작은 테이블이 사용 가능한 메모리 내에 안전하게 들어간다면, 브로드캐스트 조인은 이를 더 큰 데이터셋을 처리하는 실행기(executor)에 분산할 수 있습니다.

예시:

from pyspark.sql.functions import broadcast
result = orders.join(
broadcast(sellers),
on="seller_id",
how="left" )

이렇게 하면 대용량 orders 데이터셋을 조인하기 위해 셔플링하는 것을 방지할 수 있습니다.

하지만 중요한 제한 사항이 있습니다. 너무 큰 테이블을 브로드캐스팅하면 메모리 압박을 일으키거나 심지어 executor 실패를 유발할 수 있습니다. 항상 테이블 크기와 클러스터의 사용 가능한 리소스를 검증해야 합니다.

브로드캐스트 조인은 특히 조인의 작은 쪽(smaller side)이 실제로 브로드캐스트하기에 충분히 작을 때 유용합니다.

기술 2: 적응형 쿼리 실행 (Adaptive Query Execution, AQE)

최신 Spark 버전은 적응형 쿼리 실행(AQE)을 지원하며, 이는 런타임 통계를 사용하여 물리적 쿼리 계획의 일부를 조정할 수 있습니다.

데이터 불균형이 발생한 정렬-병합 조인(sort-merge joins)의 경우, Spark는 특정 불균형 파티션을 식별하고 이를 더 작은 조각들로 분할하여 다른 조인 쪽의 복제된 조각들과 함께 처리할 수 있습니다.

일반적인 설정은 다음과 같습니다:

spark.conf.set(
"spark.sql.adaptive.enabled",
"true" )
spark.conf.set(
"spark.sql.adaptive.skewJoin.enabled",
"true" )

이 설정들은 AQE와 그 불균형 조인 최적화를 활성화합니다.

정확한 동작은 Spark 버전, 임계값, 쿼리 계획에 따라 달라집니다. AQE가 모든 불균형 집계(skewed aggregation)나 모든 느린 작업(slow task)에 대한 만능 해결책은 아닙니다.

기술 3: 솔팅 (Salting)

솔팅은 소수의 키가 워크로드를 지배할 때 도움이 될 수 있는 수동 기술입니다.

예를 들어, Seller A의 주문이 7천만 건이라고 가정해 봅시다.

Seller A의 모든 레코드를 하나의 키 아래에서 처리하는 대신, 여러 개의 임시 키를 생성할 수 있습니다:

• Seller A + Salt 0
• Seller A + Salt 1
• Seller A + Salt 2
• Seller A + Salt 3
• Seller A + Salt 4

이렇게 하면 병렬 처리를 위한 기회가 더 많이 생깁니다.

SUM과 같은 집계의 경우, 각 솔트된 키에 대한 부분 합계를 계산한 다음, 이 총합들을 결합하여 최종 결과를 얻을 수 있습니다.

조인의 경우, 불균형한 쪽(skewed side)에 솔팅을 적용하고 작은 쪽(smaller side)에서 해당 솔트 값들에 걸쳐 대응하는 행들을 복제할 수 있습니다.

하지만 salting은 추가적인 작업과 복잡성을 도입합니다. 결과의 정확성을 유지하기 위해 신중하게 설계되어야 합니다.

테크닉 4: 데이터 모델 재검토

때로는 최적화가 Spark 설정 변경이 아닐 때가 있습니다.

왜 특정 키가 워크로드의 그렇게 큰 부분을 차지하는지 질문해 보세요.

비즈니스 프로세스가 더 균형 잡힌 키를 생성할 수 있을까요? 집계(aggregation)를 부분 단계와 최종 단계로 분리할 수 있을까요? 데이터를 더 일찍 압축하거나 필터링할 수 있을까요? 조인(join)이 현재의 단위(grain)에서 실제로 필요한가요?

최적의 해결책은 단순히 하드웨어를 추가하는 것보다 계산 자체를 변경하는 데서 나올 수 있습니다.

1. 일반적인 면접 함정: 재분할(Repartitioning)이 항상 불균형을 해결하나요?

아닙니다.

다음 것을 고려해 보세요:

df.repartition(200, "seller_id")

이는 seller_id를 사용하여 데이터를 재분할하지만, 결과 파티션 전반에 걸쳐 레코드가 고르게 분포됨을 보장하지 않습니다.

만약 Seller A가 데이터셋을 지배한다면, 해당 레코드들은 여전히 하나의 파티션에 집중될 수 있습니다.

파티션 수를 늘린다고 해서 매우 빈번한 키가 그 파티션들 전반에 걸쳐 분할됨을 보장하지는 않습니다.

이것이 엔지니어들이 두 가지 개념을 구별해야 하는 이유입니다:

파티셔닝(Partitioning): 데이터가 어떻게 구성되고 파티션 전반에 걸쳐 분포되는가.
불균형(Skew): 워크로드 또는 데이터가 해당 파티션들 전반에 걸쳐 얼마나 고르지 않게 분포되는가.

이 둘은 관련되어 있지만, 같은 것은 아닙니다.

1. 프로덕션 인시던트 발생 시 제가 할 일

만약 중요한 Spark 파이프라인이 갑자기 훨씬 느려진다면, 저는 구조화된 조사를 따를 것입니다:

1. Spark UI에서 느린 단계를 식별합니다.
2. 태스크 지속 시간(task durations), 셔플 읽기(shuffle reads), 그리고 스필 메트릭(spill metrics)을 비교합니다.
3. 몇몇 파티션이 불균형적으로 큰지 확인합니다.
4. 조인 키(join keys)와 집계 키(aggregation keys)에 매우 빈번한 값이 있는지 검사합니다.
5. 물리적 실행 계획(physical execution plan)을 검토하고 AQE가 활성화되어 있는지 확인합니다.
6. 브로드캐스트 조인(broadcast join), 스큐-조인 처리(skew-join handling), 또는 2단계 집계와 같은 가장 작은 적절한 최적화를 테스트합니다.
7. 변경 전후의 결과 정확성 및 실행 메트릭을 비교합니다.

저는 병목 현상을 먼저 식별하지 않고는 즉시 executor 개수를 늘리지 않을 것입니다.

클러스터가 더 크면 전체적인 용량 부족에는 도움이 될 수 있지만, 하나의 스큐된 태스크가 완료 시간을 결정하는 경우 개선 효과가 미미할 수 있습니다.

1. 데이터 엔지니어를 위한 더 큰 교훈

데이터 엔지니어링은 단순히 올바른 출력을 반환하는 변환(transformation)을 작성하는 것에 관한 것이 아닙니다.

이는 데이터가 어떻게 이동하고, 어디서 계산이 발생하며, 리소스가 어떻게 소비되고, 파이프라인이 프로덕션 규모에서 왜 다르게 동작하는지를 이해하는 것입니다.

데이터 스큐는 여러 필수 기술들을 연결합니다:

• 분산 컴퓨팅(Distributed computing)
• SQL 조인 및 집계(SQL joins and aggregations)
• Spark 실행 계획(Spark execution plans)
• 셔플 메커니즘(Shuffle mechanics)
• 메모리 및 디스크 관리(Memory and disk management)
• 데이터 모델링(Data modeling)
• 프로덕션 디버깅(Production debugging)

이러한 관계를 이해하면, Spark 최적화는 무작위로 설정을 변경하는 것이 아니라 증거를 진단하는 과정이 됩니다.

마지막 생각

만약 Spark 작업이 느리다면, 다음 질문만 하지 마세요:

제가 몇 개의 executor가 필요할까요?

더 유용한 질문을 하세요:

제 데이터가 해당 executor들이 효율적으로 작동하도록 분산되어 있나요?

그 단 하나의 질문이 실제 병목 현상으로 안내할 수 있습니다.

PrepNPlaced에서는 예비 및 실무 데이터 엔지니어들이 실제 시나리오, 기술적 깊이, 그리고 프로덕션 지향적인 문제 해결을 통해 개념을 이해하도록 돕는 데 중점을 두고 있습니다.

저는 Durgesh Yadav이며, 목표는 단순히 Spark, Databricks 또는 Hadoop 같은 도구를 배우는 것이 아니라고 생각합니다. 중요한 것은 이들이 왜 그렇게 작동하는지—그리고 어떻게 하면 이를 대규모로 신뢰할 수 있게 만들 수 있는지 이해하는 것입니다.

데이터 엔지니어링을 공부하고 있다면, 이 글을 저장해 두었다가 다음에 느린 Spark 작업을 문제 해결할 때 예시들을 다시 살펴보세요.

더 많은 학습 자료: https://www.prepnplaced.com/