Apache Spark(아파치 스파크): Dynamic Partition Pruning이란?

스파크 3.0 Release [2]에서 추가된 Dynamic Partition Pruning(DPP)은 아래와 같이 전반적인 성능 향상과 더불어, 특정 쿼리에서는 x100에 가까운 최적화를 가능하게 해주었습니다. 이번 글에서는 이 DPP에 대해서 자세히 알아보겠습니다.

TPCDS 1 TB Speedup by DPP - Image from [1]

 

---

Pruning이란?

pruning의 의미를 찾아보면 '가지치기'라는 뜻을 가지고 있습니다. 일반적인 데이터 시스템에서의 pruning은 얻고자 하는 데이터를 가지고 있지 않은 파일들을 읽지 않고 스킵하는 형태의 최적화를 말합니다. 

 

Pruning의 의미를 이해했다면, 이번 글에서 설명드릴 Dynamic한 형태를 설명드리기 전에 먼저 반대되는 형태인 Static한 형태를 알아볼 필요가 있겠습니다.

 

 

Static Partition Pruning

Static Partition Pruning(SPP)는 말그대로 정적으로 파티션을 pruning하는 것입니다. 아래와 같은 이미지에서 좌측은 Scan이 먼저 발생하여 모든 데이터를 스캔하는 반면에, 필터를 pushdown한 우측과 같은 경우 원하는 데이터를 가지고 있는 부분만 추린 이후에 스캔하기 때문에 후자는 전자에 비해 적은 데이터를 스캔합니다. 이는 성능으로 이어지게 됩니다.

Filter Pushdown - Image from Author inspired by [1]

위에서 간략히 그린 Data라는 부분을 스파크의 환경에서 구체적으로 살펴보면, 스파크는 데이터를 논리적으로 여러 파티션으로 인식합니다. 그렇기에 위에서 말씀드린 pruning은 파티션 단위로 결정되게 됩니다. 

SPP with Partition - Image from Author inspired by [1]

 

이러한 단일한 형태라면 SPP로도 충분합니다. 하지만 실제 데이터 환경에서는 아래와 같이 join이 추가된 형태의 쿼리가 자주 발생하게 됩니다:

 

SELECT *
FROM sales
JOIN date 
ON sales.day = date.day
WHERE date.day_of_week = 'Mon'

No SPP on sales - Image from Author inspired by [1]

위와 같은 경우, date 테이블에 대해서는 진행된 SPP가 join으로 이어진 sales에는 적용되지 못하는 것을 알 수 있습니다. 이러한 Fact - Dimension 테이블 간의 조인은 데이터 분석에서 흔한 패턴인데, Fact 테이블의 사이즈가 보통 매우 크기 때문이 위와 같은 fact 테이블의 SPP의 부재는 많은 양의 데이터 스캔을 발생시킵니다.

 

 

Dynamic Partition Pruning

DPP는 위와 같은 상황에서 아래 이미지와 같이 런타임 시에 동적으로 생성된 Partition Pruning이 적용되도록 하는 기능입니다:

 

Dynamic partition pruning is about pruning partitions based on information that can only be inferred at run time
[4]

 

DPP - Image from Author inspired by [1]

 

DPP의 구현에 있어서는 스파크 SQL 최적화에서 논리적 계획, 물리적 계획 페이즈와 연관되어 있습니다. 

 

단순하게 구현하면 아래와 같이 Dimension 테이블의 필터를 Fact Table 쿼리의 서브쿼리로 inject하여 구현할 수 있습니다:

A simple approach - Image from Author inspired by [1]

 

하지만 위와 같은 경우, 필터의 selectivity를 통계에 기반해 정확히 산정하지 못하면 매우 expensive할 수 있습니다. 그렇기에 물리적 계획 단계에서 이와 같은 서브쿼리 duplication을 제거해주게 됩니다. 

 

그러한 부분을 이해하기 위해서는 먼저 스파크 join에 대해서 이해할 필요가 있습니다.

 

Broadcast Hash Join

위와 같은 쿼리에서 dimension 테이블이 충분히 작다면, 스파크는 Broadcast hash join 형태로 join을 수행하게 됩니다. Dimension 테이블로부터 hash 테이블이 생성되고 각 노드로 distribute 됩니다(Broadcast Exchange). 그렇기에 join은 셔플 없이 각 노드가 받은 hash 테이블을 가지고 fact 테이블에 대한 스캔과 처리를 진행하게 됩니다. 

 

이러한 부분은 위의 단순한 구현에서의 서브쿼리 injection과 처리 순서를 가지기에 약간의 변형을 통해 Dynamic Filter를 만들 수 있습니다:

 

Reusing Broadcast Results - Image from Author inspired by [1]

 

 

 

Reference

[1] https://databricks.com/session_eu19/dynamic-partition-pruning-in-apache-spark

[2] https://databricks.com/blog/2020/06/18/introducing-apache-spark-3-0-now-available-in-databricks-runtime-7-0.html

[3] https://github.com/apache/spark/blob/v3.0.0/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala

[4] https://issues.apache.org/jira/browse/SPARK-11150