Databricks

Shuffle과 Broadcast Join

Shuffle은 `groupBy`, `join`, `distinct` 같은 '넓은 변환(wide transformation)' 도중 발생하며, 같은 키를 가진 행이 같은 executor에 모이도록 클러스터 전체에 걸쳐 데이터를 재배치하는 과정이다. Spark에서 가장 비용이 큰 연산인 이유는, 데이터를 직렬화하고, 로컬 디스크에 쓰고, 네트워크로 다른 노드에 전송한 뒤, 다시 역직렬화하는 과정을 모두 거치기 때문이다 — 이 모든 단계가 상당한 I/O와 지연을 만든다.

Broadcast Join은 조인에서 더 작은 테이블을 클러스터의 모든 executor 메모리로 보내는(broadcast) 최적화 기법이다. 이렇게 하면 각 executor가 네트워크로 큰 테이블을 옮길 필요 없이, 자신이 가진 큰 테이블 파티션에 대해 로컬에서 조인을 수행할 수 있다. Spark는 기본적으로 10MB 미만 테이블에 대해 이를 자동으로 시도하며(`spark.sql.autoBroadcastJoinThreshold`, 기본값 10MB), `df.join(broadcast(small_df), "id")`처럼 명시적으로 강제할 수도 있다.

Q45의 시나리오처럼 2TB 테이블과 10MB 테이블을 조인하는데 느리다면, 먼저 Spark UI의 'SQL' 탭에서 실제로 `BroadcastHashJoin`이 쓰였는지 확인해야 한다. `SortMergeJoin`으로 나온다면 자동 브로드캐스트 임계값을 초과했을 가능성이 있으므로, `broadcast()` 함수로 작은 테이블을 감싸 수동으로 브로드캐스트를 강제하면 2TB 테이블의 shuffle 자체를 없앨 수 있다.

Adaptive Query Execution(AQE)과 Data Skew 대응

AQE는 Spark 3.0에서 도입된 기능으로, 실제 데이터 통계를 바탕으로 실행 도중 쿼리 계획을 바꾼다. 크게 세 가지를 수행한다. 첫째, 셔플 파티션 병합(Coalesce Shuffle Partitions) — 셔플 결과 파티션이 너무 잘게 쪼개졌다면 AQE가 이를 더 큰 파티션으로 병합해 오버헤드를 줄인다. 둘째, 조인 전략 전환(Switch Join Strategies) — 원래 크다고 판단됐던 테이블이 필터링 후 실제로는 작아졌다면, AQE가 실행 중에 Sort-Merge 조인을 더 빠른 Broadcast 조인으로 전환한다. 셋째, 스큐 조인 최적화(Optimize Skew Joins) — 한 파티션이 다른 파티션보다 훨씬 크다는 것을 감지하면 이를 분할해 클러스터 전체에 작업을 균형 있게 배분한다.

Data Skew(데이터 편향)는 하나 또는 소수의 키에 나머지보다 압도적으로 많은 데이터가 몰려, 일부 executor만 훨씬 오래 작업하게 되는('롱테일' 문제) 상황이다. 대응 방법으로는 조인 키에 무작위 접두사('솔트')를 붙여 편향된 데이터를 더 많은 파티션에 분산시킨 뒤 조인 후 솔트를 제거하는 Salting 기법과, `/*+ SKEW('table_name', 'column_name') */`처럼 힌트를 줘 AQE 옵티마이저가 스큐를 자동으로 처리하도록 하는 Skew Hints가 있다.