groupBy는 셔플이다 — spark.sql.shuffle.partitions 200과 AQE의 파티션 재조정
PySpark Tutorial: Build a Real Pipeline and Read the Spark UI — Darshil Parmar PySpark Tutorial: Build a Real Pipeline and Read the Spark UI — "One groupBy, 200 tasks, and 151 files nobody wants" 구간 (17~22페이지)
이 모듈을 다 읽으면
- groupBy 같은 wide 트랜스포메이션이 왜 셔플을 유발하는지 설명할 수 있다
- spark.sql.shuffle.partitions 기본값 200이 결과 행 수와 무관하게 셔플 출력 파티션 개수를 결정한다는 점과 그로 인한 오버헤드를 설명할 수 있다
- AQE(Adaptive Query Execution)가 정확히 언제부터 기본 활성화되었는지를 블로그의 서술과 대조해 정정할 수 있다
도시·일자별 배송 완료 매출을 구하는 groupBy("city", "order_date") 집계는 wide 트랜스포메이션이라 같은 그룹의 행이 서로 다른 파티션에 흩어져 있으면 반드시 셔플이 일어난다. 이 셔플의 출력은 spark.sql.shuffle.partitions 설정값(기본 200)만큼의 파티션으로 쪼개지는데, 이는 결과가 단 257행이라는 사실과 무관하게 고정된 값이다. Stages 탭을 보면 200개 태스크 각각의 중앙값 실행 시간은 12ms에 불과하지만, 200개 태스크를 실행·스케줄링·종료하는 오버헤드 때문에 전체 스테이지는 0.5초가 걸린다. AQE(Adaptive Query Execution)를 켜면 런타임에 실제 셔플 출력이 678행뿐임을 감지해 AQEShuffleRead 노드가 파티션을 자동으로 병합하고, 다음 스테이지는 200개가 아닌 1개 태스크로 실행된다. 다만 AQE가 Spark 3과 4 전체의 기본값이라는 블로그의 서술은 부정확하며, 정확히는 Spark 3.2부터 기본 활성화되었다.
wide 트랜스포메이션과 셔플: 200개 태스크로 257행을 만든다
daily = joined.withColumn(...).filter(...).groupBy("city", "order_date").agg(...)를 실행하면 257개의 city-day 행이 나온다. groupBy는 wide 트랜스포메이션이다 — 같은 city와 order_date를 가진 행들이 원래 서로 다른 파티션에 흩어져 있을 수 있으므로, Spark는 이들을 물리적으로 한데 모으는 셔플을 수행해야 한다. Stages 탭에서 이 스테이지를 열어 보면 200/200 태스크가 표시되는데, 각 태스크의 중앙값 실행 시간은 12ms이고 총 태스크 시간은 3초에 불과하지만 실제 스테이지는 0.5초 만에 끝난다. 이 격차는 거의 아무 일도 하지 않는 200개의 태스크를 실행·스케줄링·종료하는 순수한 오버헤드다.
핵심 포인트
- groupBy는 wide 트랜스포메이션이라 같은 그룹의 행을 모으기 위한 셔플을 반드시 유발한다
- 257행짜리 결과를 위해 200개 태스크가 실행되며, 태스크당 중앙값 실행 시간(12ms)보다 200개 태스크 자체의 실행/스케줄링 오버헤드가 병목이 된다
spark.sql.shuffle.partitions=200의 유래와 AQE의 파티션 재조정 — 정정
이 200이라는 숫자는 결과 행 수와 무관한 spark.sql.shuffle.partitions 설정값에서 온다. 모든 셔플은 이 설정값만큼의 파티션으로 출력을 나누며, 이 기본값 200은 Spark SQL 초기 버전부터 지금까지 변하지 않았다 — '작은 잡'의 기준이 지금과 크게 달랐던 시절에 정해진 값이 그대로 남아 있는 것이다. 블로그는 AQE를 끈 실행 계획과 켠 실행 계획을 나란히 비교하며 'AQE가 Spark 3과 4의 기본값'이라고 설명하는데, 이는 지나치게 뭉뚱그린 서술이다 — 정정하면, spark.sql.adaptive.enabled는 Spark 1.6~3.0 사이에는 실험적 기능으로 존재했고, 3.0에서 재설계를 거쳐 정확히 Spark 3.2부터 기본값 true로 바뀌었다. 즉 'Spark 3 전체'가 아니라 'Spark 3.2 이상과 4'가 정확한 범위다. AQE가 켜져 있으면 런타임에 실제 셔플 출력이 678행뿐임을 감지해 AQEShuffleRead라는 노드를 추가하고 파티션들을 자동으로 병합하는데, 그 결과 다음 스테이지는 200개가 아닌 1개의 태스크로 실행된다. 이 판단은 설정을 건드리지 않아도 실제 데이터를 보고 실행 계획을 스스로 고치는 AQE의 핵심 동작이다.
핵심 포인트
- spark.sql.shuffle.partitions 기본값 200은 결과 행 수와 무관하게 모든 셔플의 출력 파티션 개수를 정한다
- 정정: AQE(spark.sql.adaptive.enabled)는 '블로그가 말하는 Spark 3 전체'가 아니라 정확히 Spark 3.2부터 기본값 true로 바뀌었다
- AQE가 켜져 있으면 런타임에 실제 셔플 출력 행 수를 보고 AQEShuffleRead 노드로 파티션을 자동 병합해, 200개 태스크가 1개로 줄어든다