Apache Spark — 느린 잡의 답은 실행 계획과 이벤트 로그에 있다 · 셔플과 파티션 수 · 讲解

셔플은 가장 비싼 한 걸음이고, 파티션 200개는 자료를 모르고 정한 숫자다

한 줄 요약

셔플은 같은 키를 한곳에 모으려고 모든 파티션이 모든 파티션에 자료를 보내는 일이고, 디스크 쓰기·직렬화·네트워크를 한꺼번에 치른다. 셔플 뒤 파티션 수는 기본 200 으로, 자료 크기와 상관없이 정해진 숫자다. AQE 가 실행 중에 작은 파티션을 합쳐 주지만, 무엇이 합쳐졌는지와 repartition·coalesce 의 차이는 직접 확인해야 한다.

概念图: 모든 파티션이 모든 파티션에 자료를 보내는 일 · 200 · 이웃한 작은 셔플 파티션을 합친다. · 셔플이 생긴다.

왜 셔플이 문제인가

filter 나 withColumn 은 파티션 하나를 받아 파티션 하나를 내놓는다. 이웃 파티션을 볼 필요가 없으니 태스크 하나 안에서 끝난다. 이런 것을 좁은 변환이라 한다. groupBy·join·distinct·orderBy 는 다르다. 같은 키가 어느 파티션에 흩어져 있을지 모르므로, 계산하려면 모든 파티션을 뒤져 같은 키끼리 모아야 한다.

[RDD 가이드의 셔플 절](https://spark.apache.org/docs/4.2.0/rdd-programming-guide.html#shuffle-operations)은 이것을 파티션 사이에 자료를 다시 나누는 Spark 의 메커니즘이라 부르고, 비용이 큰 이유를 디스크 입출력·자료 직렬화·네트워크 입출력 세 가지로 든다. 셔플을 준비하는 쪽은 맵 태스크, 모아서 계산하는 쪽은 리듀스 태스크라는 이름을 쓰는데, 문서는 이 이름이 MapReduce 에서 왔을 뿐 Spark 의 map·reduce 연산과 직접 관련은 없다고 덧붙인다.

동작은 이렇다. 맵 쪽 태스크는 결과를 메모리에 모으다가 넘치면 목적지 파티션 순으로 정렬해 파일로 쓴다. 리듀스 쪽 태스크는 모든 맵 태스크가 쓴 파일에서 자기 몫의 블록만 골라 읽는다. 문서는 셔플이 디스크에 중간 파일을 많이 만들고, 계보를 다시 계산할 때를 대비해 그 파일을 참조가 사라질 때까지 남겨 둔다고 적는다. 그래서 셔플이 많은 긴 잡은 디스크를 꽤 먹는다.

어떻게 동작하나 — 200 이라는 숫자

셔플 뒤 파티션을 몇 개로 나눌지는 누군가 정해야 한다. DataFrame 과 SQL 에서는 그 값이 [성능 조정 문서](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html#tuning-partitions)의 spark.sql.shuffle.partitions 이고, 조인이나 집계에서 셔플할 때 쓰는 파티션 수로 기본값은 200 이다.

200 은 자료를 보고 고른 숫자가 아니다. 1TB 에게는 너무 적어 파티션 하나가 5GB 안팎이 되고 메모리가 넘쳐 디스크로 흘린다. 30만 줄에게는 너무 많아 한 파티션에 수백 줄씩 들어가고, 태스크 200개를 띄우고 거두는 고정 비용이 일보다 커진다. 그 결과를 그대로 파일로 쓰면 쓰기 태스크마다 파일을 쓰므로 작은 파일이 쏟아진다.

spark.conf.set("spark.sql.adaptive.enabled", "false")    # AQE 를 끄고 날것을 본다by_day = orders.groupBy("day").agg(F.sum("qty"))by_day.write.mode("overwrite").parquet("/root/spk/out/by_day")   # 셔플 뒤 태스크 200개spark.conf.set("spark.sql.shuffle.partitions", "8")        # 자료에 맞춰 줄인다

AQE 의 파티션 합치기

손으로 숫자를 맞추기 어려운 이유는 셔플 뒤의 크기를 셔플하기 전에는 모른다는 데 있다. [셔플 뒤 파티션 합치기 절](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html#coalescing-post-shuffle-partitions)이 이 문제를 푼다. AQE 와 spark.sql.adaptive.coalescePartitions.enabled 가 둘 다 켜져 있으면(둘 다 기본 true) 맵 쪽 출력 통계를 보고 이웃한 작은 셔플 파티션을 합친다. 시작 파티션 수 initialPartitionNum 을 안 주면 spark.sql.shuffle.partitions 와 같다.

어디까지 합치는지에는 세부가 있다. 목표 크기 advisoryPartitionSizeInBytes 의 기본값은 64MB 지만, parallelismFirst 가 기본 true 라서 이 목표 크기를 무시하고 최소 크기 minPartitionSize(기본 1MB)만 지키며 병렬성을 최대로 잡는다. 문서는 바쁜 클러스터에서는 작은 태스크가 많아지지 않도록 이 값을 false 로 두라고 권한다. 즉 기본 설정의 AQE 는 "64MB 로 뭉친다" 가 아니다. 합쳐진 결과는 최종 계획에 AQEShuffleRead ... coalesced 로 나타난다.

groupBy 로 센 잡 하나가 스테이지 둘로 갈리는 그림. 첫 스테이지의 태스크들이 셔플 파일을 쓰고, 둘째 스테이지의 태스크가 모든 셔플 파일에서 자기 몫의 키만 읽는다. 둘째 스테이지의 태스크 수는 입력 파티션 수가 아니라 셔플 파티션 수가 정한다

repartition 과 coalesce

파티션 수를 직접 바꾸는 방법은 둘이고, 비용이 전혀 다르다. [repartition](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.repartition.html)은 해시 파티셔닝된 새 DataFrame 을 만든다. 모든 행을 다시 나누므로 셔플이 생긴다. 그 대가로 파티션 크기가 고르게 되고, 칼럼을 주면 같은 값이 같은 파티션에 모인다.

[coalesce](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.coalesce.html)는 좁은 의존이다. 문서의 예로 1000 개에서 100 개로 줄이면 셔플 없이 새 파티션 하나가 기존 파티션 10개를 맡는다. 늘리라고 하면 지금 수에 머문다. 대신 문서는 coalesce(1) 같은 급격한 축소가 계산 자체를 적은 노드에서 돌게 만들 수 있다고 경고한다. 셔플이 없으니 앞 단계까지 한 태스크로 합쳐지기 때문이다. 그럴 때는 셔플 한 번을 치르더라도 repartition 이 앞 단계를 병렬로 돌게 한다.

RDD 가이드는 셔플을 일으킬 수 있는 연산에 coalesce 도 넣어 두었는데, 이는 RDD 의 coalesce 가 셔플 여부를 인자로 받기 때문이다. DataFrame 의 coalesce 는 위 설명대로 셔플이 없다. SQL 에서는 [파티션 힌트](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html#coalesce-hints) COALESCE·REPARTITION·REBALANCE 로 같은 일을 하고, 문서도 이것을 출력 파일 수를 줄이는 도구로 소개한다.

현장에서 만나는 모습

작은 자료에 200 개 파일. 개발 자료로 돌린 잡이 결과 폴더에 수십 바이트짜리 파일을 잔뜩 만든다. 설정 하나만 봐도 원인이 보인다. AQE 가 켜져 있으면 많이 줄지만, 쓰기 직전에 coalesce 로 파일 수를 정하는 편이 확실하다.

셔플 양은 로그와 UI 에 있다. [웹 UI 의 스테이지 탭](https://spark.apache.org/docs/4.2.0/web-ui.html#stages-tab)은 스테이지마다 Shuffle read 와 Shuffle write 를 바이트와 레코드로 보여 주고, 이벤트 로그의 태스크 끝 이벤트에도 같은 지표가 들어 있다. "셔플이 크다" 는 느낌 대신 숫자로 말한다.

좁은 변환만 있는 잡은 스테이지가 하나다. 읽고 거르고 쓰기만 하는 잡은 셔플이 없으니 스테이지가 하나다. 스테이지가 둘 이상이면 어딘가에 셔플이 있다는 증거다.

실무에서 진짜 중요한 것

셔플은 디스크·직렬화·네트워크를 한꺼번에 치른다. 스테이지 경계가 곧 셔플이다.
shuffle.partitions 200 은 자료를 모르는 기본값이다. 자료 크기에 맞춰 정하거나 AQE 에 맡기되 결과를 확인한다.
기본 AQE 는 64MB 로 뭉치지 않는다. parallelismFirst 가 true 라 병렬성을 먼저 지킨다.
줄일 때는 coalesce, 고르게 나눌 때는 repartition. 앞은 셔플이 없고 뒤는 셔플이 있다.
coalesce(1) 은 앞 단계까지 한 태스크로 만든다. 큰 자료에서는 repartition 을 쓴다.
셔플 바이트는 UI 와 이벤트 로그에서 숫자로 읽는다.

다음 실습에서 할 것

AQE 를 끈 채로 고객별 매출을 집계해 쓰고 셔플 파티션 200 개가 만든 결과 파일 수를 센 뒤, 셔플 파티션을 8 로 줄여 차이를 본다. 설정을 기본값으로 되돌려 AQE 가 합친 뒤 실제로 셔플을 읽은 태스크 수를 이벤트 로그에서 세고, 클릭 자료에 repartition 과 coalesce 를 각각 걸어 결과 파일 수를 견준다. 셔플로 쓴 바이트를 로그에서 직접 합산하고, 좁은 변환만 있는 잡에는 셔플이 없음을 확인해 보고서로 정리한다.