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

택배 상자 5억 개에 상품 이름 스티커를 붙여야 한다고 상상해 보세요. 상품 목록이 1만 개밖에 안 된다면, 목록을 모든 작업자에게 한 장씩 나눠주고 각자 붙이게 하면 됩니다. 상자를 한 곳에 모아 줄 세울 필요가 없죠. 이것이 바로 Spark의 broadcast 아이디어입니다.

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

이 글에서 배우는 것

• broadcast join과 broadcast 변수가 어떻게 다른지
• 자동 브로드캐스트 임계값이 동작하는 방식과 수동 힌트 사용법
• 너무 큰 데이터를 broadcast했을 때 생기는 OOM 위험과 예방법
• UDF 내부 lookup, 무거운 모델, 스트리밍 상황에서의 enrichment 패턴

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

차원이 작은데도 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 는 정말 작은(수십 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 는 가장 빠르고 단순한 연산이 됩니다. "작다고 생각한 것의 실제 크기"를 항상 확인하세요.

마치며 — 핵심 요약

• broadcast join: 작은 차원(수십 MB 이하)을 모든 워커에 복제해 셔플 없이 조인하는 가장 흔하고 효율적인 패턴입니다.
• 자동 vs 수동: Spark는 통계 기반으로 자동 broadcast하지만, 통계가 없으면 broadcast() 힌트로 직접 지정해야 합니다.
• OOM 함정: 수백 MB 이상을 broadcast하면 드라이버·익스큐터 모두 위험합니다. 크기를 먼저 확인하는 습관을 들이세요.
• broadcast 변수: UDF 안에서 참조 데이터가 필요할 때 쓰되, 조인으로 해결 가능하면 broadcast join을 우선하세요.
• mapInPandas: ML 모델처럼 초기화 비용이 큰 객체는 파티션 단위로 한 번만 로드하는 이 패턴이 제격입니다.
• 스트리밍: 정적 차원은 broadcast join, 자주 바뀌는 차원은 foreachBatch 안에서 최신본을 재로딩하세요.

enrichment는 복잡한 연산이 아닙니다. 올바른 도구를 골랐다면 broadcast() 한 줄이 전부입니다 — 오늘 당장 느린 enrichment 잡에 적용해 보세요.

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

— Data Dynamics 엔지니어링 팀