Apache Spark — 느린 잡의 답은 실행 계획과 이벤트 로그에 있다 · 쏠림 · 이론

태스크 하나만 끝나지 않는 이유는 키 하나에 있다

한 줄 요약

셔플은 같은 키를 같은 태스크로 보내므로, 키 하나가 자료의 큰 몫을 차지하면 그 키를 받은 태스크 하나가 스테이지 전체를 붙잡는다. 이것이 쏠림(skew) 이고, 푸는 길은 AQE 쏠림 조인·솔팅·뜨거운 키 떼어 내기 셋이다.

개념 지도: 쏠림(skew) · 가장 늦은 태스크가 끝나야 · 같은 키는 반드시 같은 태스크로 간다. · 태스크별 분포

왜 태스크 하나가 문제인가

스테이지는 가장 늦은 태스크가 끝나야 끝난다. 태스크 199개가 2초 만에 끝나도 하나가 3분을 쓰면 스테이지는 3분이다. 그동안 나머지 코어는 논다. 실행기를 두 배로 늘려도 소용이 없다 — 느린 태스크 하나는 여전히 코어 하나에서 돈다.

왜 하나만 느린가. [RDD 프로그래밍 가이드](https://spark.apache.org/docs/4.2.0/rdd-programming-guide.html)는 셔플을 한 키의 값을 모두 한곳에 모으기 위해 모든 파티션을 읽는 all-to-all 연산으로 설명한다. 조인이든 집계든 키의 해시로 목적지 파티션을 정하므로, 같은 키는 반드시 같은 태스크로 간다. 이것이 셔플의 정확성을 보장하는 규칙이면서 동시에 쏠림의 원인이다. 클릭 24만 건 가운데 봇 한 명이 9만 6천 건을 만들었다면, 그 사용자 ID 를 받은 파티션 하나에 40% 가 몰린다. 파티션 수를 늘려도 해결되지 않는다. 키 하나는 쪼개지지 않기 때문이다.

어떻게 알아보나

쏠림은 평균으로는 보이지 않는다. 스테이지 전체의 읽은 바이트는 멀쩡하다. 봐야 할 것은 태스크별 분포다. [웹 UI 문서](https://spark.apache.org/docs/4.2.0/web-ui.html)가 설명하는 스테이지 상세 화면에는 모든 태스크의 요약 지표가 있고, 그 가운데 Duration 과 Shuffle Read Size / Records 가 최소·중앙값·최대로 나온다. 최대가 중앙값의 몇 배인가가 쏠림의 크기다. 수십 배면 태스크 하나가 일을 거의 혼자 하고 있다는 뜻이다.

이벤트 로그에도 같은 숫자가 있다. 태스크가 끝날 때마다 남는 기록에 그 태스크가 셔플로 읽은 레코드 수가 들어 있어, 로그 한 파일만 있으면 UI 없이도 최대와 중앙값을 직접 셀 수 있다.

첫 번째 길 — AQE 쏠림 조인

[성능 조정 문서](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html)의 쏠림 조인 최적화는 정렬 병합 조인의 쏠린 파티션을 비슷한 크기의 여러 태스크로 쪼개고, 반대쪽의 짝 파티션은 필요한 만큼 복제한다. spark.sql.adaptive.enabled 와 spark.sql.adaptive.skewJoin.enabled 가 둘 다 켜져 있어야 하고, 둘 다 기본값이 true 다.

어떤 파티션이 쏠렸는지는 두 조건을 모두 만족할 때 정해진다. 크기가 중앙값의 skewedPartitionFactor 배(기본 5.0)보다 크고, 동시에 skewedPartitionThresholdInBytes(기본 256MB)보다 커야 한다. 두 번째 조건 때문에 작은 실습 자료에서는 아무리 쏠려도 기본 설정으로는 아무 일도 일어나지 않는다. 실습에서 임계값을 낮추는 이유다. 쪼갤 때 목표로 삼는 크기는 advisoryPartitionSizeInBytes(기본 64MB)다.

동작하면 최종 계획에 SortMergeJoin(skew=true) 가 나타나고, 그 아래 셔플 읽기가 AQEShuffleRead … coalesced and skewed 로 바뀐다. 한 가지 단서가 있다. 쪼개기가 셔플을 하나 더 만들어야 하는 모양이면 AQE 는 기본적으로 적용하지 않는다. 그래도 하고 싶을 때 켜는 것이 spark.sql.adaptive.forceOptimizeSkewedJoin(기본 false)이다.

두 번째 길 — 솔팅

AQE 가 없거나 조인이 아닌 곳의 쏠림은 사람이 푼다. 솔팅은 큰 쪽의 키에 0부터 N-1 사이의 무작위 숫자(소금)를 붙여 뜨거운 키 하나를 N 개의 서로 다른 키로 만든다. 그러면 해시가 그것들을 N 개의 파티션으로 흩는다. 대신 작은 쪽은 짝을 잃지 않도록 모든 소금 값만큼 복제해야 한다.

from pyspark.sql import functions as FN = 8clicks_s = clicks.withColumn("salt", (F.rand(7) * N).cast("int"))users_s = users.crossJoin(spark.range(N).withColumnRenamed("id", "salt"))joined = clicks_s.join(users_s, ["user_id", "salt"]).drop("salt")

값은 분명하다. 작은 쪽이 N 배가 된다. 그래서 N 은 쏠림을 풀 만큼만 잡고, 작은 쪽이 정말 작을 때 쓴다.

세 번째 길 — 뜨거운 키 떼어 내기

뜨거운 키가 몇 개로 정해져 있으면 더 단순한 길이 있다. 그 키의 행만 걸러 따로 처리하고, 나머지는 평소대로 조인한 뒤 둘을 union 으로 붙인다. 뜨거운 키 쪽은 반대편 표에서 그 키에 해당하는 행이 몇 줄뿐이므로 브로드캐스트로 조인하면 셔플이 아예 없다. 나머지는 쏠림이 사라진 고른 자료다. 뜨거운 키를 찾는 일은 키별 개수를 세어 상위 몇 개를 보는 것으로 충분하다.

셋 가운데 무엇을 고르나

순서는 대개 이렇다. 먼저 AQE 가 이미 풀고 있는지 최종 계획에서 확인한다. 정렬 병합 조인이고 쏠린 파티션이 두 조건을 넘으면 설정 하나 바꾸지 않아도 풀린다. AQE 가 쪼개는 모양을 떠올려 보면 한계도 보인다. 쏠린 쪽 파티션을 여러 조각으로 나누고, 조각마다 반대쪽 짝 파티션을 통째로 붙여 준다. 반대쪽 짝 파티션도 크면 복제 비용이 커지고, 양쪽이 같은 키로 함께 쏠려 있으면 조각을 나눠도 짝의 수 자체가 줄지 않는다.

AQE 로 안 풀리면 뜨거운 키가 몇 개로 정해져 있는지 본다. 봇 계정 하나, 대형 고객 셋처럼 이름을 댈 수 있으면 떼어 내기가 가장 단순하고 결과를 설명하기도 쉽다. 뜨거운 키가 날마다 바뀌거나 수십 개면 솔팅이 낫다. 어떤 키가 뜨거운지 몰라도 모든 키를 고르게 흩기 때문이다. 어느 길이든 끝나면 태스크별 최대와 중앙값을 다시 재어 비율이 실제로 줄었는지 확인한다.

쏠림이 가려지는 경우 — 부분 집계

같은 봇 자료로 groupBy("user_id").count() 를 돌리면 이상하게도 쏠림이 거의 보이지 않는다. 계획을 보면 이유가 있다. 셔플 앞에 HashAggregate(partial_count) 가 있어 각 맵 태스크가 키마다 한 줄의 부분 합계만 보낸다. 봇의 9만 6천 건은 셔플 전에 맵 태스크 수만큼의 숫자 몇 개로 줄어 있다. RDD 가이드가 키별 합계나 평균에는 groupByKey 대신 reduceByKey·aggregateByKey 를 쓰라고 권하는 것도 같은 이유다.

그래서 쏠림은 조인에서, 그리고 부분 집계가 없는 연산에서 드러난다. 창 함수는 파티션 키의 모든 행을 한 태스크에 모아야 하고, 목록을 모으는 집계는 부분 결과 자체가 원래 행만큼 크다. 집계가 괜찮았다고 조인도 괜찮으리라 짐작하면 안 된다.

현장에서 만나는 모습

첫째, 진행 막대가 99% 에서 멈춘다. 태스크 하나만 남아 수십 분을 돈다. 그 태스크의 셔플 읽기 레코드가 다른 것의 수십 배면 쏠림이다.

둘째, null 키가 뜨거운 키다. 값이 빠진 칼럼에 null 이 수백만 개 쌓이면 null 도 해시로는 하나의 키라서 한 파티션에 모인다. 조인 조건에서 null 은 어떤 값과도 같지 않아 짝이 생기지 않는다. 그런데 외부 조인은 짝이 없는 행도 결과에 남겨야 하므로 그 행들이 그대로 셔플된다. null 키 행을 먼저 떼어 두었다가 조인 뒤 union 으로 붙이면 결과는 같고 쏠림은 사라진다.

셋째, 쏠림은 자라난다. 봇이나 대형 고객 하나가 생기는 날부터 어제까지 멀쩡하던 잡이 느려진다. 코드는 그대로이므로 자료 분포를 보지 않으면 원인을 찾지 못한다. 조인 키별 상위 몇 개의 행 수를 날마다 기록해 두면 뜨거운 키가 자라는 조짐을 잡이 느려지기 전에 본다.

실무에서 진짜 중요한 것

평균이 아니라 태스크별 최대와 중앙값을 본다. 그 비율이 쏠림의 크기다.
파티션 수를 늘려도 키 하나는 쪼개지지 않는다.
AQE 쏠림 조인은 정렬 병합 조인에만, 두 조건을 모두 넘을 때만 일한다.
솔팅은 작은 쪽을 N 배로 불리는 값을 치른다. N 은 필요한 만큼만.
부분 집계가 있는 집계는 쏠림을 가린다. 조인과 창 함수에서 다시 확인한다.

다음 실습에서 할 것

봇 하나가 클릭의 40% 를 만든 자료에서 뜨거운 사용자를 먼저 찾는다. 브로드캐스트와 AQE 를 끄고 정렬 병합으로 조인한 뒤, 이벤트 로그에서 셔플을 읽은 태스크별 레코드 수를 꺼내 최댓값과 중앙값을 계산한다. 그다음 AQE 쏠림 조인을 켜되 작은 실습 자료에 맞게 쏠림 임계값을 낮춰, 최종 계획에 쏠림 표식이 나타나는 것을 확인한다. 솔팅과 뜨거운 키 떼어 내기로 같은 조인을 다시 풀어 세 처방의 답이 같은지 견주고, 마지막으로 사용자별 groupBy 집계에서는 부분 집계 덕분에 쏠림이 거의 없다는 것을 같은 두 숫자로 확인한다.