Spark

advisoryPartitionSizeInBytes — AQE가 공유하는 '목표 크기' 하나

spark.sql.adaptive.advisoryPartitionSizeInBytes(기본 64MB)는 AQE의 여러 기능이 공통으로 참조하는 '이상적인 파티션 크기' 기준값이다. 파티션 병합(coalesce)에서는 이 크기에 가깝도록 작은 파티션들을 합치고, 스큐 조인 처리에서는 이 크기를 기준으로 스큐 파티션을 몇 조각으로 쪼갤지 정한다. 즉 셔플 파티션 수를 몇 개로 할지 고민하는 대신, '파티션 하나가 대략 몇 MB였으면 좋겠는가'를 이 값으로 지정하고 AQE가 실행 중 실측 통계를 보고 개수를 알아서 맞추게 하는 방식이 spark.sql.shuffle.partitions를 고정값으로 튜닝하는 것보다 최신 실무 관행에 가깝다.

coalescePartitions — 병렬성이냐 크기냐

spark.sql.adaptive.coalescePartitions.enabled(기본 true)가 켜져 있으면 AQE는 spark.sql.adaptive.coalescePartitions.initialPartitionNum(기본값 미지정 시 spark.sql.shuffle.partitions 값을 그대로 씀)만큼 일단 넉넉하게 셔플 파티션을 만든 뒤, 실행 중 실제 크기를 보고 인접한 작은 파티션들을 합친다.

합칠 때 판단 기준이 두 가지 있는데 우선순위가 있다: spark.sql.adaptive.coalescePartitions.parallelismFirst(기본 true)가 켜져 있으면 목표 크기(advisoryPartitionSizeInBytes)를 무시하고 minPartitionSize(기본 1MB)만 넘으면 합치지 않는다 — 즉 기본 동작은 '크기를 맞추는 것'보다 '병렬성을 최대한 확보하는 것'을 우선한다는 뜻이다. 클러스터 자원이 넉넉해 병렬성보다 파티션당 처리 효율을 우선하고 싶다면 parallelismFirst를 false로 끄고 advisoryPartitionSizeInBytes 목표에 더 가깝게 병합하도록 유도할 수 있다.

스큐 조인 처리 — 두 조건이 동시에 필요하다

spark.sql.adaptive.skewJoin.enabled(기본 true)가 켜져 있으면 sort-merge join에서 스큐 파티션을 자동으로 쪼갠다. 어떤 파티션이 '스큐됐다'고 판정하려면 두 조건을 동시에 만족해야 한다: (1) 크기가 spark.sql.adaptive.skewJoin.skewedPartitionFactor(기본 5.0)배 × 중간값(median) 파티션 크기보다 크고, 동시에 (2) 절대 크기가 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes(기본 256MB)를 넘어야 한다. 배수 조건만 있으면 파티션들이 전체적으로 다 작을 때(예: median 1MB에 5배인 5MB) 불필요하게 스큐로 오판하는 것을 막고, 절대크기 조건만 있으면 원래 다 고르게 큰 워크로드에서 정상 파티션도 스큐로 오판하는 것을 막는다 — 두 조건을 곱해 쓰는 이유가 바로 이 균형이다.

spark.sql.adaptive.forceOptimizeSkewedJoin(기본 false)을 켜면 스큐 최적화가 추가 셔플을 유발하는 상황에서도 강제로 적용한다(기본은 추가 비용이 이득보다 크다고 판단되면 건너뛴다). rebalancePartitionsSmallPartitionFactor(기본 0.2)는 스큐 파티션을 쪼갤 때 너무 작게 쪼개진 조각(advisory 크기의 20% 미만)은 다시 합치도록 하는 보정값이다.

localShuffleReader와 조인 시점의 브로드캐스트 재판단

spark.sql.adaptive.localShuffleReader.enabled(기본 true)는 셔플 이후 재파티셔닝이 굳이 필요 없는 상황(대표적으로 브로드캐스트 조인으로 전환된 이후)에서, 그 쪽 셔플 데이터를 원격 네트워크로 다시 읽어오지 않고 각 executor가 자기 로컬에 이미 있는 셔플 파일을 그대로 읽게 해 네트워크 트래픽을 아낀다. AQE가 실행 중 한쪽 테이블이 실제로는 충분히 작다는 걸 발견해 sort-merge join을 broadcast join으로 바꿔치기했을 때 특히 효과가 크다.

이와 관련해 spark.sql.adaptive.autoBroadcastJoinThreshold는 spark.sql.autoBroadcastJoinThreshold(정적 플래닝 시점의 브로드캐스트 임계값, 기본 10MB)와 별개로 AQE의 런타임 재판단 시점에 적용할 임계값을 따로 지정할 수 있게 해준다 — 지정하지 않으면 정적 값을 그대로 물려받는다. spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold(기본 0, 즉 비활성)를 0보다 크게 설정하면, 파티션당 크기가 이 값 이하일 때 AQE가 sort-merge join 대신 shuffled hash join을 선호하도록 만들 수도 있다(정렬을 건너뛰어 더 빠를 수 있음).