[Spark] Adaptive Query Execution(AQE)

Spark 2.x 에서는 rule-based/cost-based의 방식으로 쿼리를 최적화한다.

Spark 3.0부터는 런타임에 최적화할 수 있는 AQE가 도입되었고, Spark 3.2 부터는 AQE 활성화가 디폴트 버전이다.

 

AQE에서는 쿼리를 최적화하기 위해 크게 아래의 세 가지의 기능을 지원한다.

• Coalescing Post shuffle partitions
• Switching join strategies
• Optimizing Skew Join

Coalescing Shuffle Partitions

셔플 파티션의 수를 최적화해주는 기능이다. 

셔플은 스파크 어플리케이션의 성능에 매우 큰 역할을 차지하고, 기존에는 사용자가 셔플 크기 및 파티션을 직접 확인하며 수동으로 파티션 수를 조정해주어야 했다.

AQE에서는 런타임 중 생성되는 다양한 통계자료를 확인하여 파티션 수를 최적화한다.

 

파티션의 수는 ` spark.sql.adaptive.advisoryPartitionSizeInBytes`가 결정하는데, 기본값은 64m(64MB)이다.

평균적으로 advisoryPartitionSizeInBytes 값 근처에서 partition 크기가 결정될 수 있도록 파티션 수를 조정한다.

 

https://www.slideshare.net/slideshow/adaptive-query-execution-speeding-up-spark-sql-at-runtime/236693931

이미지와 같이 stage를 진행해가며 파티션의 크기를 모니터링하면서 COALESCE 를 통해 파티션 수를 조정하는 것을 볼 수 있다.

https://blog.cloudera.com/how-does-apache-spark-3-0-increase-the-performance-of-your-sql-workloads/

 

 

https://tech.kakao.com/2022/01/18/aqe-coalescing-post-shuffle-partitions/

 

• Coalescing Post Shuffle Partition은 shuffle 이후 coalesce를 자동으로 해준다.
• 이 기능은 너무 작고 많은 partition(task)의 생성을 방지한다.
• 잘 활용하려면 shuffle partition의 수를 처음에 충분히 큰 수를 설정해야 한다.

활성화 하는 방법은 아래와 같다

- `spark.sql.adaptive.enabled` 값 true로 설정

- `spark.sql.adaptive.coalescePartitions.enabled` 값 true로 설정

Switching Join Strategies

조인 전략을 최적화하는 기능이다.

Spark2.x에서는 수행 초기 시점에 예측한 데이터의 크기가 실제와 달라져 다른 조인 전략이 유리할 수 있음에도 런타임 최적화가 불가능하기 때문에 비효율적인 조인 전략으로 작업이 진행될 수 있다.

예측은 아래의 이유로 틀릴 수 있다:

• cardinality 또는 selectivity 예측을 위한 통계가 부정확할 수 있음.
• 대상 데이터가 복잡한 서브트리일 수 있음
• UDFs와 같이 블랙박스 predicates여서 초기 시점에 통계지 계산이 불가능할 수 있음

그러나 AQE 는 실행중에 수집한 데이터 정보를 바탕으로 중간에 Join 전략을 변경할 수 있다.

 

Converting Sort-Merge Join to Broadcast Join

 

https://jeevan-madhur22.medium.com/spark-3-0-aqe-dynamically-choosing-join-strategy-explained-part-2-66d0cfb28c42

Sort Merge Join을 Broadcast join으로 변경하는 예시를 살펴보면 이미지와 같다.

오른쪽 테이블 크기가 예를들어 25MB로 예측되었다고 가정하면, AQE가 없는 Spark2.x버전에서는 충분히 Broadcast를 할 수 있음에도 Sort merge join으로 수행이 될 것이다.

AQE enabled 시에는 일부 leaf stage 를 완료한 이후 오른쪽 테이블의 actual size가 8MB라는 것을 확인할 수 있다.

이후 최적화 과정이 일어나 Broadcast 조인으로 변경된다.

 

그러나 이미 Sort Merge Join이 시작되어버렸는데 다시 조인 유형을 변경하는 게 효율적인지 의문이 있을 수 있다.

참고: https://spark.apache.org/docs/latest/sql-performance-tuning.html#converting-sort-merge-join-to-broadcast-join

도큐먼트에서는 Sort Merge Join을 계속 끝까지 하는 것 보다는 낫다고 설명하고 있다. `spark.sql.adaptive.localShuffleReader.enabled` 값이 True라면 이미 셔플된 데이터는 로컬(executor인지?) 에 저장되고 해당 파일은 로컬에서 읽어 네트워크 트래픽을 절약할 수 있다.

 

 

Converting sort-merge join to shuffled hash join

소트머지조인: 일단 두 개의 테이블이 있을 때 일단 소팅을 하고 테이블간 record를 비교함 -> 느림

shuffled hash join: 네트워킹을 할 때 자동으로 데이터를 같은 partition에 놓음 -> 데이터 자체가 작을 때 자동으로 일어남

소팅이 없기 때문에 상대적으로 sort merge보단 빠름

 

Optimizing Skew Join

참고: https://blog.cloudera.com/how-does-apache-spark-3-0-increase-the-performance-of-your-sql-workloads/

파티션 데이터가 불균형하다면 조인 시 쿼리 성능을 저하시킬 수 있다.

https://blog.cloudera.com/how-does-apache-spark-3-0-increase-the-performance-of-your-sql-workloads/

예를들어 위 이미지의 경우 A0 파티션에 다른 파티션보다 데이터가 몰려있음을 확인할 수 있다.

AQE 를 사용하지 않으면 A1-B1, A2-B2, A3-B3의 처리가 모두 완료되어도 A0-B0의 처리를 기다려야 하고 이는 전체 어플리케이션 실행 시간에 영향을 끼친다.

https://blog.cloudera.com/how-does-apache-spark-3-0-increase-the-performance-of-your-sql-workloads/

 AQE를 활성화한다면 A0과 같은 skew data를 감지하고 더 작은 파티션으로 쪼갠다. A0이 쪼개진 개수만큼 B0은 복제된다.

이렇게 되면 A0-B0의 조인 처리 시간이 줄어들고 전체 어플리케이션 처리 속도가 향상될 수 있다.

 

 

 

활성화 하는 방법은 아래와 같다

- `spark.sql.adaptive.enabled` 값 true로 설정

- `spark.sql.adaptive.skewJoin.enabled` 값 true로 설정

 

관련 설정값은 아래와 같다

`spark.sql.adaptive.skeyJon.skewedPartitionFactor` 

스큐 파티션임을 판단하는 값: 파티션 크기의 중앙값 * skewedPartitionFactor 보다 크다면 해당 파티션은 skew partition으로 간주한다.

` Spark.sql.adaptive.skewedPartitionThresholdInBytes`

절대값으로 skew가 무시되는 임계값이다. 너무 작은 파티션이 분할되는 것을 막기 위해 설정한다.