PySpark 데이터 스큐 완전 정복 — 99%에서 멈추는 잡 살리기

Spark 잡이 "99%에서 멈춘 것처럼 보이는" 경험은 데이터 엔지니어라면 누구나 합니다. 200개 태스크 중 199개는 몇 초 만에 끝났는데 마지막 하나가 30분째 돌고 있습니다. 거의 항상 원인은 하나 — 데이터 스큐(Data Skew) 입니다.

이 글은 스큐가 왜 생기는지, Spark UI 로 어떻게 식별하는지, 그리고 AQE Skew Join 부터 Salting 까지 실전 해결 패턴을 코드와 함께 정리합니다.

1. 스큐란 무엇인가 — 파티션 불균형

Spark 는 데이터를 파티션으로 나눠 태스크에 분배합니다. 셔플(조인·groupBy)이 일어나면 같은 키가 같은 파티션으로 모입니다. 특정 키에 데이터가 몰려 있으면, 그 키를 담당하는 파티션만 거대해집니다.

정상:   [50MB][52MB][48MB][51MB] ...  → 모든 태스크 비슷한 시간
스큐:   [50MB][48MB][9GB!!][51MB] ...  → 한 태스크만 30분, 나머지는 끝남

증상:

• 한두 개 태스크만 유독 오래 걸림(straggler)
• 그 태스크에서 spill 폭증 또는 OOM
• 전체 잡 시간이 "가장 느린 태스크 하나"에 묶임

2. 스큐는 어디서 생기나

가장 흔한 범인은 NULL 또는 기본값(0, '', 'unknown') 키입니다. 의미 없는 값이 수억 건 쌓여 한 파티션으로 몰립니다.

3. 진단 — Spark UI 로 스큐 확인

추측하지 말고 Spark UI 의 Stage 상세에서 Task 분포를 봅니다.

• Summary Metrics 의 Duration, Shuffle Read Size, Spill 에서 Max 가 Median(중앙값)의 수십~수백 배라면 스큐입니다.
• 75th percentile 과 Max 의 격차가 크면 소수 태스크가 전체를 끌고 있다는 신호입니다.

# 어떤 키가 몰렸는지 직접 확인
(df.groupBy("join_key")
   .count()
   .orderBy(F.desc("count"))
   .show(20, truncate=False))

상위 몇 개 키가 전체의 상당 비율을 차지하면 핫 키 스큐가 확정됩니다.

4. 해결 1 — AQE Skew Join (먼저 시도할 것)

Spark 3.0+ 의 Adaptive Query Execution(AQE) 은 런타임에 스큐 파티션을 감지해 자동으로 더 작은 서브파티션으로 쪼갭니다. 가장 먼저, 가장 적은 노력으로 시도할 방법입니다.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
 
# 스큐 판정 임계값 (기본값 — 필요 시 조정)
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

판정 규칙(대략): 한 파티션이 중앙값의 skewedPartitionFactor(기본 5)배 보다 크고 thresholdInBytes(기본 256MB) 도 넘으면 스큐로 보고 분할합니다.

AQE 만으로 해결되면 더 손댈 필요 없습니다. 부족하면 아래 기법을 더합니다.

5. 해결 2 — Salting (소금 뿌리기)

극단적 핫 키는 인위적으로 키를 분산시킵니다. 핫 키에 랜덤 접미사(salt)를 붙여 여러 파티션으로 퍼뜨리고, 작은 쪽 테이블은 salt 만큼 복제합니다.

from pyspark.sql import functions as F
 
N = 16  # salt 개수
 
# 큰(스큐) 테이블: 조인 키에 랜덤 salt 부여
big_salted = big.withColumn("salt", (F.rand() * N).cast("int"))
 
# 작은 테이블: 0..N-1 로 복제(explode)하여 모든 salt 와 매칭되게
small_salted = (small
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)]))))
 
# salt 를 조인 키에 포함
joined = big_salted.join(small_salted, ["join_key", "salt"]).drop("salt")

핵심 아이디어: 한 핫 키 → key#0 ~ key#15 로 16조각으로 나눠 16개 태스크가 나눠 처리. 작은 테이블을 16배 복제하는 비용이 들지만, 스큐로 멈추는 것보다 훨씬 낫습니다.

핫 키만 골라 Salting (최적화)

전체에 salt 를 주면 셔플이 늘어납니다. 핫 키에만 salt 를 적용하고 나머지는 일반 조인하면 효율적입니다.

hot_keys = [0, None]  # 진단으로 찾은 핫 키
is_hot = F.col("join_key").isin(hot_keys)
 
# 핫 키 행만 salt, 나머지는 그대로 → union 후 조인하는 패턴

6. 해결 3 — Broadcast Join

조인 상대가 충분히 작으면(수백 MB 이하), 작은 쪽을 모든 익스큐터에 복제해서 셔플 자체를 없앱니다. 셔플이 없으니 스큐도 없습니다.

from pyspark.sql.functions import broadcast
 
joined = big.join(broadcast(small_dim), "join_key")
 
# 자동 브로드캐스트 임계값 조정 (기본 10MB)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100MB")

차원 테이블처럼 한쪽이 작은 별-스키마 조인에서 스큐의 근본 해법입니다. 단, 너무 큰 테이블을 broadcast 하면 익스큐터 OOM 이 나므로 크기를 확인하세요.

7. 해결 4 — NULL/기본값 키 분리

NULL 끼리는 조인되지 않는데도 한 파티션에 다 모여 스큐를 만듭니다. 조인 전에 분리합니다.

# NULL 키는 어차피 inner join 에서 매칭 안 되므로 미리 제외
matched = big.filter(F.col("join_key").isNotNull()).join(small, "join_key")
 
# NULL 행이 결과에 필요하면(outer) 따로 처리 후 union
null_rows = big.filter(F.col("join_key").isNull())
result = matched.unionByName(null_rows, allowMissingColumns=True)

기본값(0, 'unknown')도 같은 원리로, 의미 없는 값이면 분리하거나 별도 처리합니다.

8. 해결 5 — 사전 집계로 조인 데이터 줄이기

groupBy 스큐라면, 조인 전에 미리 집계해 조인에 들어가는 데이터량 자체를 줄입니다. 큰 fact 를 키별로 먼저 집계하면 조인 대상이 작아져 스큐가 완화됩니다.

# fact 를 먼저 집계 → 작아진 결과를 차원과 조인
agg = fact.groupBy("seller_id").agg(F.sum("amount").alias("total"))
result = agg.join(dim_seller, "seller_id")

9. 기법 선택 가이드

대개 AQE 켜기 → 그래도 남으면 broadcast/salting 순서로 접근합니다.

10. 흔한 함정

repartition 은 파티션 개수를 바꿀 뿐, 특정 키가 한 파티션에 몰리는 건 못 막습니다. 키 스큐에는 salt/broadcast 가 정답입니다.

11. 정리

데이터 스큐 해결의 출발점은 "추측하지 말고 Spark UI 에서 태스크 분포를 보는 것"입니다. Max 가 Median 의 수십 배라면 스큐를 의심하고, 어떤 키가 몰렸는지 확인한 뒤 AQE → broadcast → salting 순으로 대응하세요. 99%에서 멈추던 잡이 균등하게 끝나는 순간, 스큐를 이해한 보람을 느끼게 됩니다.

이 글은 Spark 3.5 기준으로 작성되었습니다. 대규모 Spark 잡의 스큐·성능 문제 진단이 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀