PySpark 버킷팅 — 반복되는 큰 조인의 셔플을 제거하기

큰 테이블 두 개를 조인하면 양쪽이 조인 키로 셔플됩니다. 그런데 같은 두 테이블을 하루에도 수십 번 조인한다면? 매번 같은 셔플을 반복하는 셈입니다. 버킷팅(Bucketing) 은 이 셔플을 데이터를 쓸 때 미리 한 번 해두어, 이후 조인에서 셔플을 없애는 기법입니다.

이 글은 버킷팅이 셔플을 제거하는 원리, 생성·조인 방법, 버킷 수 설계, 그리고 Lakehouse 시대에 버킷팅을 언제 써야 하는지를 정리합니다.

1. 문제 — 반복되는 셔플

매 조인마다:
  big_A ──(조인키로 셔플)──┐
                          ├─ SortMergeJoin
  big_B ──(조인키로 셔플)──┘
→ 두 테이블 모두 매번 네트워크로 재분배 (비쌈)
→ 같은 조인을 반복하면 같은 셔플을 반복

조인 키가 같은데 broadcast 하기엔 둘 다 크다면, SortMergeJoin 은 매번 양쪽을 셔플합니다. 이 비용이 누적되면 막대합니다.

2. 버킷팅의 아이디어 — 셔플을 미리 해두기

버킷팅은 테이블을 쓸 때 조인 키의 해시로 N개 버킷에 미리 나눠 저장합니다. 같은 키는 항상 같은 버킷 번호로 갑니다. 두 테이블을 같은 키·같은 버킷 수로 버킷팅해 두면, 조인 시 "같은 버킷끼리만" 맞추면 되므로 셔플이 필요 없습니다.

[버킷팅된 테이블]
  big_A: bucket_0, bucket_1, ... bucket_31  (user_id 해시로 분배)
  big_B: bucket_0, bucket_1, ... bucket_31  (동일)
 
조인: bucket_0 ↔ bucket_0, bucket_1 ↔ bucket_1 ...
→ 셔플 없이 같은 버킷끼리 로컬 조인

핵심: 셔플 비용을 "쓰기 시점 한 번"으로 옮기는 것입니다. 한 번 버킷팅해두면 이후 모든 조인이 셔플 없이 빠릅니다.

3. 버킷 테이블 생성

버킷팅은 메타스토어 테이블(saveAsTable) 로 저장해야 합니다. 단순 write.parquet 경로 저장으로는 버킷 정보가 보존되지 않습니다.

# user_id 로 32개 버킷, 버킷 내 정렬까지
(big_A.write
    .bucketBy(32, "user_id")
    .sortBy("user_id")              # 정렬해두면 SMJ 의 정렬 단계도 절약
    .mode("overwrite")
    .saveAsTable("analytics.events_bucketed"))
 
(big_B.write
    .bucketBy(32, "user_id")        # 같은 키, 같은 버킷 수!
    .sortBy("user_id")
    .mode("overwrite")
    .saveAsTable("analytics.users_bucketed"))

4. 버킷 조인 — 셔플이 사라지는지 확인

a = spark.table("analytics.events_bucketed")
b = spark.table("analytics.users_bucketed")
 
joined = a.join(b, "user_id")
joined.explain()
# → Exchange(셔플) 노드가 사라지고 SortMergeJoin 만 남으면 성공

EXPLAIN 에서 조인 양쪽의 Exchange 가 없으면 버킷팅이 동작한 것입니다(별도 글 "PySpark 느린 잡 디버깅"의 EXPLAIN 읽기 참고). 버킷이 다르면(버킷 수 불일치 등) 셔플이 다시 등장합니다.

5. 버킷 수 설계

버킷 수는 신중히 정해야 합니다 — 나중에 바꾸려면 전체 재작성이 필요하기 때문입니다.

# 대략적 추정: 총 크기 / 목표 버킷 크기
# 예: 256GB / 512MB ≈ 512 버킷 (2의 거듭제곱 선호)

흔한 함정: 한쪽은 32 버킷, 다른 쪽은 64 버킷으로 만들면 셔플 제거가 안 됩니다. 함께 조인할 테이블들은 버킷 수를 통일하세요.

6. 버킷팅의 한계와 함정

버킷팅은 반복되는 조인에서만 이득입니다. 한 번 쓰고 마는 조인이라면 쓰기 시 셔플 비용만 추가될 뿐입니다.

7. Lakehouse 시대 — Iceberg/Delta 와 버킷팅

이 블로그 독자에게 중요한 부분입니다. Iceberg/Delta 는 테이블 포맷 차원의 버킷 파티셔닝을 제공해, Hive 식 버킷팅보다 유연합니다.

Iceberg bucket transform

-- Iceberg: bucket transform 으로 hidden partitioning
CREATE TABLE analytics.events (
  user_id BIGINT, event_time TIMESTAMP, ...
) USING iceberg
PARTITIONED BY (days(event_time), bucket(32, user_id));

Iceberg 의 bucket(N, col) 은 파티션 전략의 일부로, 파티션 진화로 나중에 바꿀 수 있고 메타데이터 기반 프루닝과 함께 동작합니다. (Iceberg 파티션 transform 은 별도 글 "Trino + Iceberg 는 파티션 문제를 어떻게 해결하는가"에서 다뤘습니다.)

Lakehouse 를 쓴다면, 순수 Hive 버킷팅보다 Iceberg/Delta 의 bucket 파티셔닝을 우선 검토하세요. 더 유연하고 Trino 등 다른 엔진과도 일관되게 동작합니다.

8. 버킷팅 vs 다른 셔플 절감 기법

순서: broadcast 로 되면 broadcast → 안 되고 반복 조인이면 버킷팅/Iceberg bucket.

9. 정리

버킷팅은 "셔플 비용을 미래의 모든 조인에서 한 번으로 줄이는" 투자입니다. 같은 큰 테이블을 반복 조인하는 워크로드라면 강력하지만, 일회성 조인이나 작은 테이블에는 손해입니다. 그리고 Lakehouse 환경이라면 더 유연한 Iceberg/Delta 의 bucket 파티셔닝을 먼저 고려하는 것이 현명합니다 — 셔플 제거라는 목적은 같으면서, 파티션 진화와 멀티 엔진 호환이라는 이점이 따라오기 때문입니다.

이 글은 Spark 3.5 + Iceberg/Delta 기준으로 작성되었습니다. 반복 조인 최적화나 Lakehouse 파티션 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀