Apache Spark 내부 완전 가이드 2025: RDD, Catalyst Optimizer, Tungsten, Whole-Stage Codegen, Shuffle 심층 분석

A.

Lineage의 의미: RDD는 데이터 자체가 아닌 **"어떻게 만들어졌는지의 기록"**을 유지한다.

val rdd1 = sc.textFile("hdfs://data.txt")      // source
val rdd2 = rdd1.map(_.toUpperCase)              // parent: rdd1
val rdd3 = rdd2.filter(_.startsWith("A"))       // parent: rdd2
val rdd4 = rdd3.flatMap(_.split(","))           // parent: rdd3

각 RDD는 부모 RDD와 변환 함수를 기억한다. 이 관계를 이어가면 원천 데이터 소스부터 현재까지의 전체 계보가 된다.

장애 복구 원리:

시나리오: rdd4.collect() 실행 중 executor 하나가 죽음. 그 executor가 처리하던 rdd4의 partition 3이 사라짐.

복구 과정:

1. 손실 감지: Driver가 task 실패 감지.
2. Lineage 추적: rdd4 partition 3 → rdd3 partition 3 → rdd2 partition 3 → rdd1 partition 3.
3. 재계산: 다른 executor에서 원천부터 다시 계산.
   • HDFS에서 원본 텍스트 partition 3 읽기.
   • map 적용.
   • filter 적용.
   • flatMap 적용.
4. 완료: 손실된 partition 복구.

Key: Partition 단위 재계산. 전체 RDD가 아닌 손실된 partition만. 다른 partition들은 영향 없음.

Replication과의 비교:

Traditional fault tolerance (replication):

• 모든 데이터를 여러 노드에 복제.
• 네트워크 I/O 많음 (복제 시).
• 디스크 사용 2~3배.
• 장애 시 즉시 다른 복제본 사용.

RDD lineage:

• Replication 없음. 데이터 한 번만.
• 네트워크/디스크 절약.
• 장애 시 재계산 (시간 걸림).
• 메모리 기반 연산과 궁합 좋음.

왜 Spark의 선택은 lineage인가:

Spark는 반복 연산(ML, 그래프)을 타겟으로 했다. 이런 워크로드는:

• 데이터가 메모리에 있어 빠름.
• Lineage 재계산도 (메모리니까) 빠름.
• Replication 비용이 반복적 쓰기로 누적.

Lineage의 한계:

1. Lineage가 길면 재계산이 오래:

rdd1 → rdd2 → rdd3 → ... → rdd100 → action

rdd99 partition 손실 시, rdd1부터 재계산. 수 분~시간 가능.

해결: Checkpoint:

rdd50.checkpoint()

• rdd50을 디스크에 저장.
• Lineage를 이 지점에서 "잘라냄".
• 이후 복구는 rdd50부터.

2. Wide dependency의 문제:

rdd1 (4 partitions) --shuffle--> rdd2 (4 partitions)

rdd2 partition 3 손실 → rdd1의 모든 partition 필요 (shuffle이라 재분배). 비용이 큼.

해결: Shuffle 결과를 영구 저장. Spark가 기본으로 함 (shuffle files가 로컬 디스크에 남음).

3. 외부 시스템 부작용:

rdd.foreach(x => saveToDatabase(x))

재계산 시 중복 저장 위험. Lineage는 순수 함수 가정.

해결: Idempotent 연산, 또는 정확히 한 번 semantic을 별도 구현.

교훈:

RDD lineage는 **"replication 없이 내결함성"**을 가능하게 했다. 이것이 Spark가 MapReduce 대비 100배 빠를 수 있었던 기반이다. 메모리 연산 + 경량 복구 = 빅데이터의 새 패러다임.

현대 적용:

• RDD lineage → DataFrame, Dataset 아래에서도 유지.
• Checkpoint → 장기 실행 job에서 중요.
• Structured Streaming에서도 비슷한 아이디어 (log-based recovery).

분산 시스템 철학적 관점:

Lineage는 "상태를 저장하지 말고, 만드는 방법을 저장하라" 는 접근. 이는 함수형 프로그래밍의 철학과 일치한다:

• Immutable: RDD는 변하지 않음.
• Referential transparency: 같은 입력 → 같은 출력 (재계산 가능).
• Composition: RDD들의 체인.

Spark는 분산 컴퓨팅에 함수형 원칙을 적용해서 새 영역을 열었다. 그 단순하고 우아한 아이디어가 10년 넘게 이 분야를 이끌고 있다.

A.

Catalyst의 핵심 아이디어: SQL/DataFrame 쿼리를 tree 로 표현하고, 수십~수백 개의 rewrite rule을 반복 적용해서 더 효율적인 tree로 변환한다.

4단계 처리 과정:

1. Parsing → Unresolved Logical Plan

SELECT name FROM users WHERE age > 20

↓

Project [name]
  Filter (age > 20)
    UnresolvedRelation("users")

2. Analysis → Resolved Logical Plan

Catalog 참조로:

• users → 실제 테이블 스키마.
• name, age → 컬럼 참조.
• 타입 검증.

Project [name#10]
  Filter (age#11 > 20)
    Relation[name#10, age#11, email#12] parquet

3. Logical Optimization → Optimized Logical Plan

이 단계가 진짜 마법. 많은 rule이 반복 적용된다:

Rule 1: Projection Pushdown

Project [name#10]
  Filter (age#11 > 20)
    Relation[name#10, age#11, email#12]  ← email 불필요

↓

Project [name#10]
  Filter (age#11 > 20)
    Project [name#10, age#11]  ← 필요한 컬럼만
      Relation[name, age, email]

이점: Parquet reader가 name, age 컬럼만 읽음. email 스킵. I/O 절약.

Rule 2: Filter Pushdown to Data Source

Parquet/ORC는 predicate pushdown을 지원한다:

Filter (age > 20)
  Relation(parquet)

↓

Relation(parquet) WITH FILTER (age > 20)

Parquet reader가 min/max 통계를 보고 row group 전체를 스킵. 수십 배 빠른 읽기.

Rule 3: Constant Folding

WHERE age > 18 + 2

↓

WHERE age > 20

매 행마다 18+2를 계산하지 않음.

Rule 4: Simplification

WHERE true AND x > 10

↓

WHERE x > 10

WHERE x > 10 OR false

↓

WHERE x > 10

Rule 5: Join Reorder

SELECT * FROM a JOIN b ON a.id = b.id JOIN c ON b.cid = c.id
WHERE a.country = 'KR'

CBO (Cost-Based Optimizer)가 통계를 보고 최적 순서 결정:

• a를 먼저 필터 (country='KR').
• 작아진 a와 b 조인.
• 결과와 c 조인.

Catalyst는 지수 개의 조인 순서를 탐색하고 가장 싼 것 선택.

Rule 6: Subquery Unnesting

SELECT * FROM orders
WHERE user_id IN (SELECT id FROM users WHERE country = 'KR')

↓

SELECT o.* FROM orders o
INNER JOIN (SELECT id FROM users WHERE country = 'KR') u
ON o.user_id = u.id

Semi-join으로 변환. 훨씬 빠름.

Rule 7: Column Pruning Through Join

SELECT a.name FROM a JOIN b ON a.id = b.aid

b에서는 aid만 필요 (조인 키). 다른 컬럼 스킵.

Rule 8: Decimal 연산 최적화

SELECT SUM(price * quantity) FROM sales

Decimal 곱셈은 비쌈. Catalyst는 overflow를 고려해서 필요한 정밀도만 사용.

4. Physical Planning

논리 계획 → 실행 연산자:

Project → org.apache.spark.sql.execution.ProjectExec
Filter → FilterExec
Join → SortMergeJoinExec (또는 BroadcastHashJoinExec 등)
Aggregate → HashAggregateExec

여러 물리 계획을 생성하고 cost를 비교해서 선택.

예: Join 전략 선택

SELECT * FROM big JOIN small ON big.id = small.id

Catalyst는 다음을 고려:

• small 크기 < spark.sql.autoBroadcastJoinThreshold (기본 10MB)?
  • Yes: BroadcastHashJoin 선택 (small을 모든 executor에 복사, shuffle 없음).
  • No: SortMergeJoin 선택 (양쪽 shuffle).

Broadcast join은 수 배~수십 배 빠르다.

실전 효과:

간단한 쿼리:

SELECT u.name, SUM(o.amount)
FROM users u JOIN orders o ON u.id = o.user_id
WHERE u.country = 'KR'
GROUP BY u.name

최적화 전 (naive 실행):

1. users 전체 스캔.
2. orders 전체 스캔.
3. 전체 join.
4. filter (country='KR').
5. group by.

1 TB orders 스캔. 수십 분.

최적화 후 (Catalyst):

1. users에 filter pushdown (country='KR' → 10% 데이터만).
2. Projection pushdown (id, name만 읽음).
3. orders도 필요 컬럼만 읽음 (user_id, amount).
4. Broadcast join (users가 충분히 작아서).
5. group by.

I/O 90% 절약. 10배 빠른 쿼리.

Catalyst의 확장성:

Catalyst의 rule은 사용자 정의 가능:

object MyRule extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
    case Filter(p, child) => /* custom logic */
  }
}

spark.experimental.extraOptimizations = Seq(MyRule)

이 확장성이 Databricks의 Delta Lake, Iceberg 같은 프로젝트의 기반.

교훈:

Catalyst는 **"규칙 기반 트리 변환"**이라는 단순한 아이디어로 거대한 성과를 냈다. 각 rule은 단순하지만 조합되면 놀라운 최적화.

전통 DB (PostgreSQL, MySQL)도 쿼리 optimizer가 있지만 Catalyst의 강점:

1. Scala tree API로 쉬운 rule 작성.
2. Extensible: 외부 rule 추가.
3. DataFrame API와 통합: SQL이 아닌 코드도 최적화.

결과: 기본 SQL을 작성해도 손으로 최적화한 것과 비슷한 성능. 이것이 Spark가 데이터 엔지니어에게 사랑받는 이유다.

A.

JVM 객체의 숨은 비용:

평범한 Java 코드:

Integer age = 30;
String name = "Alice";

메모리 사용을 보면:

• Integer: 16 bytes header + 4 bytes value + 4 bytes padding = 24 bytes (4 bytes 데이터에)
• String "Alice":
  • String 객체: ~40 bytes
  • 내부 char[]: 16 bytes header + 10 bytes (5 chars) + padding = ~32 bytes
  • 총: ~72 bytes (5 chars 데이터에)

실제 데이터 (int 4 + 5 chars = 9 bytes)가 96 bytes를 차지. 10배 오버헤드.

빅데이터에선 이것이 재앙:

1억 row에 5개 필드가 있다면:

• 실제 데이터: ~5 GB.
• JVM 객체로: ~50 GB.

메모리 부족. GC pressure. 캐시 miss.

Tungsten의 해결:

Row를 연속된 바이트 배열에 저장:

Row(age=30, name="Alice", email="a@b.c")

Tungsten format:
[null_bitmap:8B][age:8B][name_ptr:8B][email_ptr:8B][name:"Alice"][email:"a@b.c"]

• 고정 크기 필드 (int, long, double): 인라인.
• 가변 크기 필드 (string, array): pointer + 뒷부분에 실제 데이터.
• Null bitmap: 각 필드의 null 여부.
• Header 없음: 순수 데이터만.

결과:

• 5개 필드 row: ~40 bytes (JVM 객체 ~200 bytes에서).
• 5배 메모리 절약.

UnsafeRow:

Spark의 구현 클래스:

public class UnsafeRow extends BaseMutableRow {
    private Object baseObject;   // byte[] 또는 null (off-heap)
    private long baseOffset;      // 시작 위치
    private int numFields;
    private int sizeInBytes;
}

직접 메모리 접근 (sun.misc.Unsafe):

public long getLong(int ordinal) {
    assertIndexIsValid(ordinal);
    return Platform.getLong(baseObject, baseOffset + ordinal * 8L);
}

JVM을 거치지 않고 포인터 산술로 바로 메모리 접근. Unsafe는 이름대로 안전장치를 건너뛴다. JVM이 검증 안 함.

효과 1: 메모리 절약

1억 row, 10 필드 DataFrame:

• JVM object: ~30 GB.
• Tungsten: ~5 GB.

6배 적은 메모리. 같은 클러스터에서 6배 큰 데이터 처리 가능.

효과 2: GC pressure 제거

JVM GC는 많은 작은 객체에서 느려진다. Young generation을 계속 순회해야.

Tungsten은 적은 수의 큰 배열만 사용:

• Off-heap: GC 대상 아예 아님.
• On-heap: GC 하더라도 객체 수가 적음.

결과: Full GC 빈도 대폭 감소. Pause time 대폭 감소.

효과 3: Cache friendliness

CPU cache는 연속된 메모리를 선호 (spatial locality). 각 cache line은 64 bytes.

JVM 객체:

Integer 1 @ address 0x1000
Integer 2 @ address 0x8000  ← 멀리 떨어짐 (heap 할당 랜덤)
Integer 3 @ address 0x4000

하나 읽을 때마다 cache miss 가능성. 수십 ~ 수백 cycles 낭비.

Tungsten:

[field1][field2][field3][field4]... 연속

한 번의 메모리 접근이 여러 필드를 cache에 로드. 다음 접근은 hit.

성능: 2~5배 향상 (메모리 집약적 워크로드).

효과 4: Direct serialization

UnsafeRow는 이미 byte array. 네트워크로 전송 시:

• Java 객체: Kryo/Java serialization 필요 (느림).
• UnsafeRow: 그대로 전송 (zero serialization).

Shuffle이 훨씬 빠름.

효과 5: Vectorized 처리의 기반

연속 메모리는 SIMD (벡터화 명령어) 친화적:

// 일반
for (int i = 0; i < n; i++) {
    result[i] = a[i] + b[i];
}

// SIMD (4개 동시 처리)
for (int i = 0; i < n; i += 4) {
    _mm_store_ps(result + i, _mm_add_ps(_mm_load_ps(a + i), _mm_load_ps(b + i)));
}

Tungsten의 연속 메모리가 SIMD의 기반. JVM은 자체적으로 SIMD를 잘 활용하지 못하지만, 데이터 레이아웃이 친화적이면 JIT이 인식.

비용과 제약:

복잡도: Tungsten은 매우 낮은 레벨이다. 버그가 있으면 JVM crash. Spark 팀이 조심스럽게 개발.

디버깅 어려움: JVM heap dump, profiler가 덜 유용.

특정 연산의 오버헤드: 가변 크기 필드(string) 수정이 복잡.

Python 호환: PySpark에서 UnsafeRow를 Python 객체로 변환할 때 비용.

Arrow와의 협력:

최근 Spark는 Apache Arrow와 통합:

• Arrow: 메모리 columnar 포맷.
• Python ↔ JVM 교환 시 Arrow 사용.
• Pandas UDF가 Arrow 덕분에 10배 빠름.

Tungsten과 Arrow는 서로 보완. Tungsten은 row-wise 처리, Arrow는 column-wise 교환.

Spark 3.x의 Vectorized Execution:

Spark 3.0+는 일부 연산자에 columnar 처리 도입:

• Parquet/ORC vectorized reader.
• Columnar aggregation.
• 4096 rows batch 단위.

Tungsten row format에서 columnar batch로 진화. DuckDB, ClickHouse의 방식을 따라감.

교훈:

Tungsten은 Spark의 **"underlying 재발명"**이다. 표면 API는 그대로지만 내부가 완전히 바뀌었다. 사용자는 변경 없이 2~10배 성능 향상.

이는 **"하드웨어를 존중하는 설계"**의 중요성을 보여준다:

• CPU cache가 싫어하는 것 (random access) 피하기.
• GC의 약점 (작은 객체) 회피.
• Memory bandwidth 아끼기.

JVM이 "추상화"를 제공하지만, 고성능이 필요하면 그 추상화를 영리하게 우회해야 한다. Tungsten은 "Java/Scala 속에서 C 수준 성능을 낸다"는 도전의 답이다.

모든 빅데이터 엔진이 결국 같은 방향으로 진화: 바이너리 포맷, 연속 메모리, 벡터화. Tungsten은 Spark의 해답이었고, 같은 아이디어가 Flink, Presto, Snowflake에서도 발견된다. 수렴 진화의 예다.

A.

전통 Volcano 모델:

1970년대부터 쓰인 query execution model. 각 연산자는 iterator:

interface QueryOperator {
    Row next();
    boolean hasNext();
}

예시: SELECT name FROM users WHERE age > 20

Project ← Filter ← Scan

실행:

Project.next() {
    while (true) {
        Row row = Filter.next();
        if (row == null) return null;
        return new Row(row.getString("name"));
    }
}

Filter.next() {
    while (true) {
        Row row = Scan.next();
        if (row == null) return null;
        if (row.getInt("age") > 20) return row;
    }
}

Scan.next() {
    return disk.readNextRow();
}

각 연산자가 독립 객체. next()를 계속 호출.

Volcano의 비용:

1. Virtual function calls: next()는 인터페이스 메소드. JVM은 virtual dispatch 사용 → inline 어려움.

2. Object allocation: 매 row마다 새 Row 객체 생성 가능성. GC 부담.

3. Branch prediction 실패: 연산자마다 다른 코드 경로 실행. CPU pipeline 예측 실패.

4. Cache miss: 매 연산자의 코드가 다른 메모리 위치.

5. Boxing: Java의 경우 Integer 같은 wrapper 사용 → heap 할당.

결과: 단순 쿼리도 Volcano 오버헤드만으로 2-5배 느림.

Whole-Stage Code Generation의 답:

아이디어: "여러 연산자를 하나의 함수로 병합". 런타임에 Java 코드를 생성하고 컴파일.

예시: 같은 쿼리

Volcano (의사 코드):

for each row from Scan:
    call Filter.next(row)
    call Project.next(filter_output)
    output project_output

3개 iterator, 수많은 함수 호출.

WSCG가 생성하는 코드:

public class GeneratedIterator {
    public Object[] next() {
        while (scanner.hasNext()) {
            InternalRow row = scanner.next();
            int age = row.getInt(2);   // age 컬럼
            
            if (age > 20) {  // filter inlined
                UTF8String name = row.getUTF8String(1);  // name 컬럼
                return new Object[]{ name };  // project inlined
            }
            // 조건 불만족, 다음 row로
        }
        return null;  // 끝
    }
}

핵심 차이:

1. 단일 함수: Scan, Filter, Project가 한 함수의 내부. 함수 호출 없음.

2. Inline: 컴파일러(JVM JIT)가 모든 것을 인라인. Virtual call 제거.

3. 직접 메모리 접근: row.getInt(2)는 Tungsten의 Unsafe.getInt(address).

4. 단순한 loop: 하나의 while 루프. CPU pipeline 친화적.

성능 향상:

Databricks 벤치마크 (TPC-DS):

• Volcano: 100%.
• WSCG: 400% ~ 1000%.

4-10배 향상. 마법이 아니라 오버헤드 제거.

어떻게 작동하는가:

1. Spark의 최적화 단계:

   • Logical plan → Physical plan.
   • 여러 연산자를 stage로 그룹화 (narrow dependency 내).
   • 각 stage가 WSCG 대상.

2. Code generation:

   • Physical plan 트리를 순회.
   • 각 연산자가 자기 코드 조각 제공.
   • 합쳐서 Java source code 생성.

3. Compilation:

   • Janino (가벼운 Java compiler)로 컴파일.
   • 바이트코드 생성.
   • ClassLoader에 로드.

4. Execution:

   • 컴파일된 클래스의 인스턴스 생성.
   • 일반 Java 메소드처럼 호출.
   • JVM JIT가 최적화 (hot path → native code).

실전 예시:

spark.range(1, 1000000)
     .filter($"id" > 500000)
     .select($"id" * 2)
     .count()

EXPLAIN 결과:

*(1) Filter (id#0L > 500000)
+- *(1) Range (1, 1000000, step=1, splits=...)

*(1)이 WSCG stage 번호. Range, Filter, Project가 하나의 코드로 통합.

explain("codegen")으로 실제 생성된 코드 확인 가능:

public Object generate(Object[] references) {
    return new GeneratedIteratorForCodegenStage1(references);
}

final class GeneratedIteratorForCodegenStage1 extends BufferedRowIterator {
    // ... 수백 줄의 생성된 코드
}

제약:

1. 64KB 메소드 한계: JVM의 단일 메소드가 64KB 초과 불가. 너무 많은 연산자 병합 시 분할.

2. 복잡한 연산자: Window function, UDF 등은 codegen 어려움. Fall back to Volcano.

3. 디버깅 어려움: 생성된 코드를 디버거로 추적 힘듦.

4. 컴파일 오버헤드: 매 쿼리마다 컴파일. 작은 쿼리엔 역효과 가능.

WSCG vs Vectorized:

Spark의 WSCG는 row-at-a-time + inline. 또 다른 접근:

Vectorized Execution (ClickHouse, DuckDB, Presto 일부):

• Batch 단위 (1024 rows).
• SIMD 명령 활용.
• Columnar 처리.

장점: SIMD의 힘. 단점: 구현 복잡.

Spark는 두 접근을 조합:

• Parquet reader: vectorized (columnar).
• 연산자: WSCG (row-wise).
• 향후: 더 많은 vectorization 도입 중.

Hyper / Umbra 비교:

독일 TU Munich의 HyPer/Umbra 연구 시스템은 더 급진적:

• LLVM IR로 컴파일.
• 네이티브 코드 (JIT 없음).
• 성능이 C++ 손 작성 코드 수준.

Spark WSCG는 Java 바이트코드까지. JVM JIT에 의존. Umbra만큼은 아니지만 훨씬 간단하고 이식성 있다.

Rough numbers:

• Volcano: baseline.
• WSCG: 4-10x faster.
• Vectorized: 4-10x faster.
• LLVM-based: 10-20x faster.

Spark는 "충분히 빠르면서 구현 가능"의 균형을 찾았다.

교훈:

WSCG는 전통 지혜의 재평가다. Volcano 모델이 40년간 지배한 이유는 우아함과 유연성이지만, 현대 CPU 아키텍처 (deep pipeline, cache, SIMD)에선 적합하지 않다.

해결: "연산자 추상화를 유지하면서도 런타임에 구체화". 이것이 codegen의 본질.

이는 "지연된 optimization" 철학이다:

• 사용자는 high-level 코드 작성.
• 시스템이 물리적 최적화.
• 컴파일러의 전통적 역할을 런타임으로.

JIT 컴파일러가 수십 년간 해온 일이지만, 빅데이터에 적용한 것은 Spark가 처음 크게 성공시켰다. 이후 거의 모든 현대 데이터 엔진이 이 패턴을 따르고 있다.

현대 SQL 엔진의 성능은 하드웨어 성능의 몇 % 를 사용하는가의 경쟁이다. Volcano는 10%, WSCG는 40%, vectorized는 60%, LLVM-based는 80%. 각 진화가 이 수치를 밀어올린다.

A.

Static optimization의 한계:

전통 쿼리 optimizer (Catalyst 포함)는 실행 전에 최적화를 결정한다. 근거는:

• 통계 (ANALYZE TABLE).
• 데이터 소스 메타데이터.
• Schema.

문제: 통계가 부정확하거나 최신이 아닐 수 있다:

1. 데이터가 변경됨 (ANALYZE 이후).
2. 통계 계산이 비쌈 (대용량).
3. 복잡한 조건의 selectivity 추정 어려움.
4. Join 결과 크기 예측 어려움.

결과: 잘못된 계획 선택 → 느린 쿼리.

AQE의 기본 아이디어:

"Stage 경계에서 실제 통계를 수집하고, 남은 계획을 재최적화"

Spark는 이미 shuffle 단계에서 중간 결과가 디스크에 저장된다. 이 데이터의 실제 크기를 측정해서 남은 단계를 최적화할 수 있다.

AQE의 세 가지 최적화:

1. Dynamically Coalescing Shuffle Partitions

문제: spark.sql.shuffle.partitions 기본값 200. 실제 데이터 크기와 무관.

시나리오:

• 작은 table filter 결과: 100 MB.
• 200 partition = 각 500 KB.
• 200개의 매우 작은 task. 오버헤드 비율 높음.

AQE 해결:

• Shuffle 완료 후 실제 partition 크기 측정.
• 작은 partition들을 병합.
• 예: 200 → 20 (각 ~5 MB).
• Task 수 감소, 오버헤드 감소, 성능 향상.

설정:

spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.coalescePartitions.minPartitionSize=1MB
spark.sql.adaptive.advisoryPartitionSizeInBytes=64MB

2. Dynamically Switching Join Strategies

문제: Join 전략 선택 시 table 크기 추정이 부정확.

시나리오:

SELECT * FROM big JOIN medium ON big.id = medium.id
WHERE medium.status = 'active'

계획 시 추정: medium 크기 = 100 MB → SortMergeJoin 선택.

실제 실행: 필터 후 medium.active = 5 MB. Broadcast가 훨씬 나았을 것.

AQE 해결:

• Medium 필터 결과를 shuffle.
• 실제 크기가 5 MB임을 확인.
• BroadcastHashJoin으로 동적 전환.
• 수 배 빠름.

예시:

Before AQE: 120 seconds (SortMergeJoin)
After AQE:  25 seconds (auto-switched to BroadcastHashJoin)

설정:

spark.sql.adaptive.autoBroadcastJoinThreshold=10MB

3. Dynamically Optimizing Skew Joins

문제: Data skew. 일부 key가 과도하게 많음.

시나리오:

SELECT * FROM orders JOIN users ON orders.user_id = users.id

• 대부분 user는 10~100 orders.
• 한 user (봇, 관리자)가 1,000,000 orders.
• 해당 partition이 100배 크다.
• 그 task가 100배 오래 걸림.
• 나머지 tasks는 빨리 끝나지만 skewed task 기다림.

전통적 해결: Salting (수동, 복잡).

AQE 해결:

1. Shuffle 후 각 partition 크기 확인.
2. 평균의 5배 이상 크면 skewed로 판단.
3. Skewed partition을 여러 sub-partition으로 분할.
4. 각 sub-partition이 다른 task에서 처리.
5. 반대쪽 partition도 복제 (broadcast 필요).

예시:

Before:
Task 1: 1 GB (skewed user)  → 100 seconds
Task 2: 10 MB               → 1 second
Task 3: 10 MB               → 1 second
...
Total time: 100 seconds (Task 1 bound)

AQE divides Task 1 into 10 sub-tasks:
Task 1a: 100 MB → 10 seconds
Task 1b: 100 MB → 10 seconds
...
Task 2: 10 MB → 1 second
Total time: 10 seconds (10x faster)

설정:

spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

4. Dynamic Partition Pruning (DPP) (AQE와 함께 발전)

Star schema의 fact + dimension join:

SELECT f.amount
FROM fact_sales f JOIN dim_date d ON f.date_id = d.id
WHERE d.year = 2024

전통: fact_sales 전체 스캔 + join + filter.

DPP:

1. dim_date에서 year=2024인 id 먼저 추출.
2. 이 id들을 fact_sales의 partition filter로 pushdown.
3. fact_sales 스캔 시 해당 partition만 로드.

효과:

• 파티션된 fact_sales (year별)이면 1 TB → 50 GB.
• 20배 빠른 쿼리.

실측 효과:

Databricks TPC-DS 벤치마크:

• AQE 비활성: 100% (baseline).
• AQE 활성: 130~200%.
• 일부 쿼리: 3배+ 향상.
• 평균: 1.8배.

거의 공짜. 기존 코드 변경 없이.

Spark 버전별:

• Spark 3.0: AQE 도입 (기본 비활성).
• Spark 3.2: AQE 기본 활성화.
• Spark 3.3+: 개선 지속.

내부 작동:

AQE는 Spark의 query execution을 stage 단위로 실행하되, stage 사이에 pause한다:

1. Stage 1 실행.
2. Stage 1 완료 → shuffle 파일 생성.
3. Shuffle 파일의 실제 통계 수집.
4. Stage 2의 physical plan을 re-optimize.
5. Stage 2 실행.
6. 반복.

이 stage 단위 통계 → 재최적화 루프가 AQE의 핵심.

한계와 함정:

1. Streaming에 제한적: Structured Streaming은 AQE 덜 적용됨.

2. 작은 쿼리엔 불필요: Re-optimization 오버헤드가 작은 쿼리에선 역효과.

3. 복잡한 쿼리 예측 어려움: 매우 복잡한 쿼리는 여전히 수동 튜닝 필요.

4. DPP와의 상호작용: 복잡한 경우 예측 안 되는 계획 변경.

실전 권장:

# 기본 설정 (Spark 3.2+는 이미 기본)
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true

# 파라미터 조정
spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB  # 목표 partition 크기
spark.sql.adaptive.coalescePartitions.minPartitionSize=32MB  # 최소
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5  # skew 기준

교훈:

AQE는 "optimization은 한 번에 끝나지 않는다" 는 철학의 구현이다. 전통 optimizer는 ahead-of-time (실행 전 모든 결정), AQE는 just-in-time (필요 시 재최적화).

이는 JIT 컴파일러의 철학과 같다:

• 처음엔 inexpert 결정.
• 런타임 관찰.
• 최적화.

분산 데이터 처리에서 같은 원칙 적용. Spark가 선구자였고, 이후 Trino, Presto도 유사 기능 추가.

**"정적 vs 동적 최적화"**의 트레이드오프:

• 정적: 단순, 예측 가능. 완전 최적은 아닐 수 있음.
• 동적: 복잡, 오버헤드. 실제 데이터에 적응.

AQE는 이 둘 사이의 실용적 타협이다. 완벽한 동적 재계획은 아니지만, stage 경계에서 핵심 결정을 조정함으로써 대부분의 이득을 얻는다.

미래:

AQE는 계속 발전 중:

• Multi-stage re-optimization.
• Cost model 개선.
• ML 기반 최적화 (실험적).
• Streaming adaptation.

Spark가 10년 넘게 데이터 처리 영역을 선도하는 이유는 이런 지속적 혁신 때문이다. "충분히 좋다"에 만족하지 않고, 매 버전마다 의미 있는 개선을 가져온다. AQE는 그 한 예이며, 앞으로도 더 나아질 것이다.