PySpark Broadcast 변수와 대형 Lookup — 셔플 없이 데이터 보강하기

데이터 보강(enrichment)은 일상입니다 — 거래에 상품 정보를, IP 에 지역 정보를, 코드에 이름을 붙입니다. 큰 팩트 테이블(수억 행)에 작은 참조 데이터(수만~수십만 행)를 조인하는 이 작업을, 순진하게 하면 양쪽이 셔플되어 느려집니다. 작은 데이터를 모든 워커에 복제하면 셔플 없이 끝낼 수 있습니다.

이 글은 broadcast join 과 broadcast 변수의 차이, 자동 브로드캐스트의 동작, 너무 큰 broadcast 의 위험, 그리고 외부 lookup 패턴을 정리합니다.

1. 문제 — 작은 데이터 붙이는데 셔플이 발생

큰 팩트(5억 행) ──(조인키로 셔플)──┐
                                  ├─ SortMergeJoin
작은 차원(10만 행) ──(셔플)────────┘
→ 5억 행을 네트워크로 재분배 (작은 데이터 붙이려고!)

차원이 작은데도 SortMergeJoin 이면 팩트 5억 행이 통째로 셔플됩니다. 낭비입니다. 작은 차원을 복제하면 이 셔플이 사라집니다.

2. Broadcast Join — 가장 흔한 해법

작은 쪽을 모든 익스큐터에 복제해서 셔플 없이 조인합니다.

from pyspark.sql.functions import broadcast
 
enriched = fact.join(broadcast(dim_product), "product_id")
# → dim_product 가 모든 워커에 복제, fact 는 그 자리에서 조인 (셔플 0)

# 자동 브로드캐스트 임계값 (기본 10MB) — 통계상 이보다 작으면 자동 broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50MB")

Spark 는 통계상 작은 테이블을 자동으로 broadcast 합니다. 자동이 안 먹으면(통계 부재 등) broadcast() 힌트로 강제하세요. EXPLAIN 에서 BroadcastHashJoin 이 보이면 성공입니다(별도 글 "PySpark 느린 잡 디버깅").

3. 가장 큰 함정 — 너무 큰 것을 broadcast

broadcast 대상은 드라이버가 모았다가 모든 익스큐터로 복제합니다. 너무 크면:

큰 테이블을 broadcast → 드라이버가 통째로 collect → 드라이버 OOM
또는 → 각 익스큐터가 거대 복사본을 메모리에 → 익스큐터 OOM

원칙: broadcast 는 정말 작은(수십 MB 수준) 데이터에만. 차원이 커지면 broadcast 가 아니라 버킷팅(별도 글 "PySpark 버킷팅")이나 일반 조인을 고려하세요. "작다고 생각한 차원"이 실제로 큰지 크기를 확인하는 습관이 중요합니다.

4. Broadcast 변수 — 조인이 아닌 lookup

DataFrame 조인이 아니라, UDF/맵 연산 안에서 참조 데이터를 쓰고 싶을 때 broadcast 변수를 씁니다. 작은 dict/set 을 모든 익스큐터에 한 번만 복제해, 행마다 다시 보내지 않게 합니다.

# 작은 참조 데이터를 드라이버에서 dict 로 만든 뒤 broadcast
code_map = {row["code"]: row["name"] for row in dim_small.collect()}
bc = spark.sparkContext.broadcast(code_map)
 
from pyspark.sql.functions import udf
@udf("string")
def lookup_name(code):
    return bc.value.get(code)        # 복제된 dict 에서 조회
 
df = df.withColumn("name", lookup_name("code"))

대부분의 enrichment 는 broadcast join 이 낫습니다(옵티마이저가 다루고 UDF 비용 없음). broadcast 변수는 "조인으로 표현 안 되는 참조"(복잡한 lookup, 작은 ML 모델, 룰 테이블)에만 쓰세요.

5. 큰 모델·커넥션은 mapInPandas 로

ML 모델이나 DB 커넥션처럼 무거운 객체로 enrichment 할 때는, broadcast 변수보다 mapInPandas 가 적합합니다 — 파티션당 한 번만 초기화하기 때문입니다.

def enrich(batches):
    model = load_model()                 # 파티션당 1회 (행마다 아님)
    for pdf in batches:
        pdf["score"] = model.predict(pdf[features])
        yield pdf
 
df.mapInPandas(enrich, schema="... score double")

(이 패턴은 별도 글 "PySpark UDF가 느린 이유와 Pandas UDF"에서 자세히 다룹니다.)

6. 스트리밍 enrichment

스트림에 참조 데이터를 붙일 때도 broadcast 가 유효합니다. 단, 참조 데이터가 변하면 처리가 필요합니다.

# 정적 차원을 스트림에 broadcast join (차원이 거의 안 변할 때)
stream.join(broadcast(dim_static), "key")
 
# 차원이 주기적으로 바뀌면: foreachBatch 안에서 최신 차원을 다시 읽어 조인
def process(batch_df, batch_id):
    dim = spark.read.table("analytics.dim")   # 매 배치 최신 차원
    batch_df.join(broadcast(dim), "key").write...

자주 바뀌는 차원은 foreachBatch 안에서 매 배치 최신본을 읽어 broadcast 하는 패턴이 흔합니다.

7. enrichment 패턴 선택

8. 정리

데이터 보강의 핵심은 "작은 참조 데이터는 셔플하지 말고 복제하라"입니다. 대부분의 enrichment 는 broadcast() join 한 줄로 셔플 없이 끝나고, 조인으로 표현 안 되는 참조만 broadcast 변수로, 무거운 모델은 mapInPandas 로 다룹니다. 단 하나의 함정 — 너무 큰 것을 broadcast 하면 OOM — 만 피하면, enrichment 는 가장 빠르고 단순한 연산이 됩니다. "작다고 생각한 것의 실제 크기"를 항상 확인하세요.

이 글은 Spark 3.5 기준으로 작성되었습니다. 대규모 데이터 보강·조인 최적화 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀