Spark SQL과 Catalyst 옵티마이저

SQL은 문자열이 아니라 실행 계획이 된다

Spark에서 spark.sql("...")을 호출하거나 DataFrame API로 select, filter, join, groupBy를 이어 붙이면, Spark는 그 코드를 곧바로 클러스터에 던지지 않는다. 먼저 “무엇을 계산하려는가”를 트리 형태의 계획으로 바꾸고, 그 계획을 여러 단계로 고친 뒤, 실제 Executor에서 실행할 물리 계획으로 낮춘다. 이 중심에 있는 컴포넌트가 Catalyst optimizer다.

실무에서 Catalyst를 알아야 하는 이유는 단순히 내부 구조가 흥미롭기 때문이 아니다. 느린 쿼리를 만났을 때 “코드를 조금 바꿔보자”보다 먼저 해야 할 일이 있다. Spark가 어떤 필터를 먼저 적용했는지, 어떤 컬럼을 읽지 않게 만들었는지, 어떤 join 전략을 골랐는지, 어디서 Exchange가 생겼는지 확인해야 한다. 즉 Spark SQL 튜닝은 코드를 보는 일과 실행 계획을 읽는 일이 같이 가야 한다.

Catalyst가 다루는 네 가지 계획

EXPLAIN EXTENDED를 실행하면 보통 다음 네 종류의 계획을 볼 수 있다.

EXPLAIN EXTENDED
SELECT customer_id, sum(amount) AS revenue
FROM orders
WHERE order_date >= DATE '2026-01-01'
GROUP BY customer_id;

중요한 점은 DataFrame API도 Spark SQL과 같은 최적화 경로를 탄다는 것이다. RDD처럼 사용자가 모든 실행 흐름을 직접 조립하는 방식이 아니라, Catalyst가 논리 계획을 보고 더 효율적인 형태로 고칠 수 있다. 그래서 Spark에서 구조화된 데이터 처리는 대체로 SQL/DataFrame을 우선으로 생각하는 편이 좋다.

논리 최적화: 덜 읽고, 덜 옮기고, 덜 계산하게 만든다

Catalyst의 첫 번째 강점은 규칙 기반 논리 최적화다. 대표적인 규칙은 다음과 같다.

이 최적화는 “결과가 같아야 한다”는 조건 안에서만 일어난다. Catalyst가 비즈니스 의미를 추측해 임의로 조건을 바꾸지는 않는다. 예를 들어 UDF 안에 복잡한 로직을 숨기면 Catalyst가 그 내부 의미를 잘 이해하지 못해 pushdown이나 재정렬 기회를 잃을 수 있다. 가능하면 내장 함수와 명시적인 컬럼 연산으로 표현하는 것이 계획 최적화에 유리하다.

물리 계획: 같은 논리도 여러 방식으로 실행될 수 있다

논리 계획이 “무엇을 계산할지”라면, 물리 계획은 “클러스터에서 어떻게 계산할지”다. 같은 join이라도 Spark는 여러 실행 전략 중 하나를 선택할 수 있다.

물리 계획에서 특히 눈여겨볼 단어는 Exchange다. Exchange는 데이터가 repartition되거나 broadcast되는 경계다. 대개 stage가 나뉘고 네트워크 I/O가 발생한다. 지난 장에서 본 셔플 문제는 실제 계획에서는 Exchange hashpartitioning(...) 같은 형태로 드러난다.

== Physical Plan ==
*(2) HashAggregate(keys=[customer_id], functions=[sum(amount)])
+- Exchange hashpartitioning(customer_id, 200)
   +- *(1) HashAggregate(keys=[customer_id], functions=[partial_sum(amount)])
      +- *(1) Project [customer_id, amount]
         +- *(1) Filter (order_date >= 2026-01-01)
            +- *(1) FileScan parquet ...

이 계획에서 읽을 수 있는 내용은 꽤 많다. 필터가 scan 가까이에 있고, 필요한 컬럼만 project되며, 부분 집계 후 customer_id 기준 셔플이 발생한다. 즉 코드는 한 줄의 groupBy처럼 보이지만 실제 실행은 scan, filter, partial aggregate, exchange, final aggregate로 나뉜다.

통계가 없으면 좋은 계획을 고르기 어렵다

Catalyst와 Cost-Based Optimization은 통계가 있을수록 더 나은 결정을 내린다. Spark 문서는 통계의 출처를 크게 데이터 소스 메타데이터, Catalog 통계, AQE runtime 통계로 나눈다. Parquet 파일의 row group 통계, Hive Metastore나 Catalog에 저장된 table statistics, 실행 중 shuffle stage에서 얻은 실제 partition 크기 등이 모두 계획 선택에 영향을 줄 수 있다.

운영에서 자주 생기는 문제는 “데이터는 변했는데 통계는 낡은” 상태다. Optimizer가 작은 테이블이라고 믿고 broadcast를 골랐는데 실제로는 커졌거나, join cardinality 추정이 틀려 비싼 join 순서를 고를 수 있다. 그래서 관리형 테이블을 많이 쓰는 환경에서는 통계 갱신도 성능 운영의 일부로 봐야 한다.

-- 테이블 통계 갱신 예시
ANALYZE TABLE orders COMPUTE STATISTICS;
ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS customer_id, order_date;

-- 추정 통계 확인
EXPLAIN COST
SELECT customer_id, sum(amount)
FROM orders
WHERE order_date >= DATE '2026-01-01'
GROUP BY customer_id;

통계를 무조건 많이 모으면 된다는 뜻은 아니다. 컬럼 통계 수집도 비용이 있고, 모든 임시 데이터에 완벽한 통계를 유지하기는 어렵다. 핵심은 join 순서, broadcast 여부, partition 크기처럼 성능에 큰 영향을 주는 테이블과 컬럼부터 관리하는 것이다.

AQE: 실행 중에 계획을 다시 고친다

Spark 3.x 이후의 중요한 변화는 Adaptive Query Execution(AQE)이다. 전통적인 최적화는 실행 전에 가진 통계로 계획을 고른다. 하지만 실제 데이터 분포는 예측과 다를 수 있다. AQE는 shuffle이나 broadcast 같은 materialization point에서 runtime statistics를 얻고, 남은 계획을 다시 최적화한다.

AQE가 대표적으로 하는 일은 세 가지다.

1. Shuffle partition coalescing: 너무 작은 post-shuffle partition들을 합쳐 task 수와 작은 I/O를 줄인다.
2. Join strategy switching: 실행 중 확인해 보니 한쪽 결과가 작다면 sort merge join 대신 broadcast hash join으로 바꿀 수 있다.
3. Skew join optimization: 일부 partition만 지나치게 큰 경우 해당 skew partition을 나누어 긴 tail task를 줄인다.

실행 전 계획
SortMergeJoin
+- Exchange hashpartitioning(key, 200)
+- Exchange hashpartitioning(key, 200)

AQE runtime statistics 이후
BroadcastHashJoin
+- 큰 쪽 scan
+- BroadcastExchange(실제로 작아진 쪽)

AQE는 튜닝을 대신해 주는 마법이 아니라, “초기 추정이 틀릴 수 있다”는 현실을 보완하는 장치다. spark.sql.adaptive.enabled가 켜져 있어도 입력 파일이 지나치게 잘게 쪼개져 있거나, UDF 때문에 필터 pushdown이 막히거나, join key 자체가 잘못 설계된 문제까지 해결하지는 못한다.

EXPLAIN을 읽는 습관

Spark SQL 튜닝을 처음 할 때는 EXPLAIN 출력이 길고 낯설다. 모든 줄을 완벽하게 이해하려고 하기보다, 다음 순서로 읽으면 실무 판단이 쉬워진다.

1. scan 대상 확인: 의도한 table/path만 읽는가? partition pruning이 되었는가?
2. filter 위치 확인: Filter가 scan 가까이에 있는가? pushdown이 되는 파일 포맷인가?
3. column pruning 확인: 필요 없는 큰 컬럼을 읽고 있지 않은가?
4. Exchange 개수 확인: join, aggregate, sort 때문에 셔플이 몇 번 생기는가?
5. join strategy 확인: broadcast가 적절한가, sort merge가 불가피한가?
6. AQE 적용 확인: Spark UI나 formatted plan에서 adaptive plan과 runtime statistics를 확인했는가?

실제 장애나 지연 분석에서는 Spark UI와 함께 봐야 한다. EXPLAIN은 계획을 보여주고, Spark UI는 그 계획이 실행되며 어떤 stage와 task에서 시간이 쓰였는지 보여준다. 계획상 Exchange가 하나뿐이어도 특정 key skew 때문에 한 task만 오래 걸릴 수 있고, 이때는 SQL 문장보다 stage detail의 partition size와 task duration이 더 직접적인 증거가 된다.

Catalyst를 돕는 코드 작성 습관

Catalyst는 강력하지만, 사용자가 정보를 숨기면 할 수 있는 일이 줄어든다. 다음 습관이 계획 품질에 직접 영향을 준다.

반대로 Python UDF를 남발하거나, 문자열 SQL을 동적으로 조립해 조건을 숨기거나, 모든 컬럼을 select("*")로 계속 들고 다니면 Catalyst가 읽고 줄이고 재배치할 여지가 줄어든다. Spark 성능 튜닝은 설정값 몇 개를 외우는 일이 아니라 Optimizer가 좋은 선택을 하도록 충분한 구조와 통계를 제공하는 일에 가깝다.

마무리

Catalyst는 Spark SQL과 DataFrame 코드를 실행 가능한 분산 계획으로 바꾸는 최적화 계층이다. 분석 단계에서는 컬럼과 타입을 해석하고, 논리 최적화 단계에서는 필터·컬럼·상수식을 정리하며, 물리 계획 단계에서는 join과 shuffle 전략을 선택한다. AQE는 여기에 runtime statistics를 더해 실행 중 계획을 다시 고칠 수 있게 한다.

운영 관점에서 가장 중요한 습관은 EXPLAIN과 Spark UI를 함께 보는 것이다. 느린 Spark SQL을 만났다면 먼저 plan에서 Filter, Project, Exchange, join strategy를 확인하고, UI에서 실제 stage와 task 분포가 그 계획과 어떻게 맞물렸는지 확인하자. 그래야 “Spark가 느리다”가 아니라 “어떤 계획 선택과 데이터 분포 때문에 느린지”까지 좁힐 수 있다.

References

• Apache Spark Documentation — Performance Tuning
• Apache Spark Documentation — EXPLAIN
• Databricks — What is a Catalyst Optimizer?
• Databricks Blog — Adaptive Query Execution: Speeding Up Spark SQL at Runtime