[spark] Adaptive Query Execution

실제 spark의 설계를 하다보면 처음에는 균등하게 파티션을 분배하면 되겠지 라고 실행해보면, 실제로는 하나의 파티션이 대부분을 차지하게 되어서 다른 작업들이 밀리는 현상을 보게 된다.

 

다음은 spark의  실행계획 이미지이다.  왼쪽은 통계 추정으로 만든 초기 계획, 오른쪽은 실행이 끝난 뒤의 최종 계획이다.

 

최종계획을 보면 Sort 두 개와 SortMergeJoin이 사라지고 BroadcastHashJoin으로 바뀌었다.

실행해 보니 오른쪽 테이블이 추정보다 훨씬 작았다는 뜻이다. 새로 나타난 AQEShuffleRead가 AQE 개입 지점이고, 뒤에 붙는 local · coalesced 같은 꼬리표가 무엇을 했는지 알려준다. 이 두 계획의 차이가 곧 AQE다.

 

 

AQE는 shuffle이 끝난 다음 partition을 coalesce(병합)를 해주는 기능이다. 결국

- task 스케쥴링 오버헤드 감소

- 디스크 및 네트워크 io 효율 극대화

- 동적 파티션 좆어으로 메모리 효율

- 최종 파일 개수 최적화 

를 통해 처리 속도가 빨라진다.

 

정적 통계만으로는 부족한 이유

Spark가 여러 실행 계획 중 하나를 고르는 근거는, 각 노드가 몇 행을 뱉을지에 대한 추정치다.

통계 출처	내용	갱신 방법
Data source	Parquet 파일 메타데이터의 행 수, min/max 등 데이터 소스가 직접 들고 있는 값	데이터 소스가 관리
Catalog	Hive Metastore 같은 카탈로그에 저장된 테이블·컬럼 통계	ANALYZE TABLE
Runtime	쿼리가 도는 동안 Spark가 직접 계산한 값 — AQE의 영역	자동

 

얼마나 데이터가 있는지 정확도를 높이려면 컬럼 히스토그램 같은 상세 통계를 유지해야 하는데 항상 최신으로 유지하는 비용이 만만치 않다. AQE는 그 비용과 정확도 사이의 저울질을 계획 시점에서 실행 시점으로 미루는 방식으로 우회한다.

 

 

언제 다시 계획하는가

재최적화의 핵심 질문은 [언제 실행되는가?] 다. Spark의 연산자들은 보통 파이프라인으로 묶여 병렬 실행되지만 shuffle 또는 broadcast exchange가 그 파이프라인을 끊는다. 이 지점을 materialization point라 부르고 그 사이 구간을 query stage라 한다.

query stage는 중간 결과를 물리적으로 떨어뜨린다. 다음 단계는 앞 단계가 전부 끝나야 진행할 수 있어 그 대기 시간이 곧 재최적화의 기회다. 

 

AQE는 아무 쿼리에나 붙지 않는다. 공식 문서 기준으로 두 조건을 만족해야 한다.

• 스트리밍 쿼리가 아닐 것
• exchange를 최소 하나 포함하거나 서브쿼리를 포함할 것 — 보통 join, 집계, 윈도우 연산이 있을 때

바꿔 말하면 단순 스캔 / 필터 / 저장으로 끝나는 파이프라인은 AQE의 대상이 아니다. 

 

 

3. AQE가 실제로 하는 세 가지

① 파티션 자동 병합 — Coalesce Shuffle Partitions

 

spark.sql.shuffle.partitions는 shuffle 후 파티션 개수를 정한다. 기본값은 200.

여기에 구조적인 문제가 있다. 최적값이 쿼리마다 다르다. 수십 GB를 집계할 때는 200이 적절하지만, 결과가 100행뿐인 쿼리에서 200개 파티션은 파티션당 0.5행이 된다. 태스크 200개를 띄우고 스케줄링하는 비용이 실제 연산을 압도한다.

AQE는 파티션별 실제 크기를 보고, 인접한 작은 파티션들을 spark.sql.adaptive.advisoryPartitionSizeInBytes(기본 64MB)에 가깝게 묶는다.

 

② 편중 파티션 자동 분할 — Skew Join

join할 때 같은 키는 같은 파티션으로 모이는데,  하나의 키가 전체의 절반을 차지하면 그 파티션 하나가 전체 작업의 절반을 차지하게된다. 

이때 다음 설정값으로 skew join 을 할수 있다. 

 

설정	기본값	설명
spark.sql.adaptive.skewJoin.enabled	true	skew join 분할 활성화
spark.sql.adaptive.skewJoin.skewedPartitionFactor	5	중앙값(평균 아님)의 몇 배부터 편중으로 판정할지
spark.sql.adaptive.skewedPartitionThresholdInBytes	256MB	절대 크기 하한. 두 조건을 모두(AND) 만족해야 분할
spark.sql.adaptive.advisoryPartitionSizeInBytes	64MB	쪼갠 조각의 목표 크기

주의 

• 두 조건이 AND로 동작한다. 데이터가 작으면 비율상 심하게 편중돼도 256MB 문턱에 걸려 분할되지 않는다.
• 대상은 join이다. groupBy 집계의 편중은 별개 문제이고, 키에 난수를 붙여 흩뜨리는 salting 같은 수동 기법이 필요하다.
• join 타입별 제약이 있다. 예를 들어 LEFT OUTER JOIN은 왼쪽 편중만 최적화된다.

③ join 전략 런타임 교체 — Sort-Merge → Broadcast

join 전략은 양쪽 크기에 따라 갈린다.

 

Broadcast Hash Join	한쪽이 작음 (기본 10MB 이하)	작은 쪽을 전 executor에 복제. 큰 쪽은 제자리
Sort-Merge Join	양쪽 다 큼	양쪽 shuffle + 정렬. 비쌈

문제는 계획 시점에 필터 적용 후의 크기를 모른다는 것이다. 원본이 크니까 Sort-Merge로 정했는데 실제로 필터가 1/100로 줄였다면, broadcast가 압도적으로 유리하다. AQE는 shuffle이 끝난 뒤 실제 크기가 spark.sql.adaptive.autoBroadcastJoinThreshold(미설정 시 정적 임계값 10MB) 이하면 계획을 갈아끼운다.

 

 

4. 부작용!!

AQE에는 부작용이 있다. stage마다 재계획하며 stage를 개별 job으로 제출하므로 쿼리 하나가 여러 job으로 쪼개진다.

실습에서 두 줄짜리 코드를 실행했는데 Spark UI에는 job이 5개 잡혔다. 게다가 각 job의 description이 우리 코드 위치가 아니라 AQE 내부의 비동기 실행 래퍼 함수명으로 찍힌다.

 

 

5. 확인 방법

df = spark.range(10_000_000)

# local[*] 에서는 코어 수에 따라 결정된다 (spark.default.parallelism)
print(df.rdd.getNumPartitions())

# filter 는 narrow
print(df.filter("id % 7 = 0").rdd.getNumPartitions())

# groupBy 는 wide — shuffle 이 일어나며 파티션 수가 재결정된다
grouped = df.selectExpr("id % 100 as k").groupBy("k").count()
print(grouped.rdd.getNumPartitions())     # AQE 개입 여부

grouped.explain(mode="formatted")          # AQEShuffleRead 노드 확인

input(f"UI({spark.sparkContext.uiWebUrl}) SQL 탭 확인 후 Enter >> ")
spark.stop()

explain() 함수는 초기 계획을 보여준다. 실행 전에는 isFinalPlan=false로 실행 후 다시 찍으면 true로 바뀌며 실제 AQE가 채택된 계획이 출력된다 

 

== Physical Plan ==
AdaptiveSparkPlan (11)
+- == Final Plan ==
   ResultQueryStage (8)
   +- * HashAggregate (7)
      +- AQEShuffleRead (6)
         +- ShuffleQueryStage (5), Statistics(sizeInBytes=23.4 KiB, rowCount=1.00E+3)
            +- Exchange (4)
               +- * HashAggregate (3)
                  +- * Project (2)
                     +- * Range (1)
+- == Initial Plan ==
   HashAggregate (10)
   +- Exchange (9)
      +- HashAggregate (3)
         +- Project (2)
            +- Range (1)

 

 

결국 AQE를 사용하면 어느정도는 최적화가 될테지만, 사용하는 데이터와 코드에 따라 사람이 직접 파라미터 조정을 해줘야 한다.

 

끝.