[메모리 효율] [PySpark] 동시 Spark 작업에서 Broadcast 변수 중복 전송 방지 및 최적화

PySpark 애플리케이션에서 동시성 처리는 성능 최적화의 핵심 요소입니다. 특히 대용량 데이터를 처리할 때 작은 차원(dimension) 데이터프레임을 브로드캐스트하여 조인(join)하는 것은 일반적인 최적화 기법입니다. 하지만 여러 개의 스파크 작업이 동시에 실행될 때, 이 브로드캐스트 변수가 각 작업마다 중복 생성되어 메모리 비효율을 초래하지 않을까 하는 의문이 생길 수 있습니다. 이 글에서는 PySpark의 F.broadcast() 함수와 동시성 환경에서의 브로드캐스트 변수 재사용 메커니즘을 심층적으로 분석하고 최적화 방안을 제시합니다.

1. 에러 발생 상황

한 개발자가 PySpark 애플리케이션에서 ThreadPoolExecutor를 사용하여 여러 독립적인 워크로드를 동시에 처리하는 상황을 맞닥뜨렸습니다. 각 스레드는 동일한 차원 데이터프레임(dim)을 사용하여 스파크 조인 작업을 수행합니다.

from pyspark.sql import functions as F
from concurrent.futures import ThreadPoolExecutor

# 차원 데이터프레임을 스레드 시작 전에 캐시
dim = spark.read.table("dimension")
dim.cache()
dim.count() # 캐싱 작업 트리거

def process_one(source_data_path):
    fact = spark.read.parquet(source_data_path)
    join_condition = (fact.dim_id == F.col("dim.id"))
    
    # 각 스레드에서 F.broadcast(dim) 사용
    result = fact.join(
        F.broadcast(dim), # 여기서 dim은 전역 dim 객체
        join_condition,
        "left"
    )
    
    result.write.mode("overwrite").parquet(f"output/{source_data_path.split('/')[-1]}")
    print(f"Processed: {source_data_path}")

# 예를 들어, 15개의 스레드를 동시에 실행
workload_paths = [f"s3a://data/source_{i}.parquet" for i in range(15)]

with ThreadPoolExecutor(max_workers=15) as executor:
    executor.map(process_one, workload_paths)

이 상황에서 개발자는 다음과 같은 의문을 가졌습니다. 여러 동시 스파크 작업이 F.broadcast(dim)을 사용할 때, 스파크는 각 익스큐터(executor)당 차원 데이터프레임의 브로드캐스트 복사본을 단 하나만 유지하고 모든 동시 작업/스레드가 이 동일한 복사본을 재사용할 수 있을까요? 아니면 각 독립적인 스파크 작업이 동일한 데이터프레임에 대해 자체 브로드캐스트를 생성하여, 각 익스큐터에 여러 개의 차원 복사본이 생길 수 있을까요?

궁극적인 목표는 아래와 같이 각 익스큐터에 브로드캐스트 복사본이 단 하나만 존재하고 여러 작업이 이를 공유하는 것입니다.

Executor
└── 1 broadcast copy of dimension
       ↑
       ├── Job 1
       ├── Job 2
       ├── Job 3
       └── ...

2. 명확한 발생 원인

결론부터 말씀드리자면, 스파크는 위 질문에 대한 긍정적인 답변을 제공합니다. 즉, 동일한 dim 데이터프레임 객체를 참조하는 여러 동시 스파크 작업은 각 익스큐터에서 하나의 브로드캐스트 복사본을 재사용할 수 있습니다. 이러한 동작은 스파크의 최적화 기능과 브로드캐스트 관리 메커니즘 덕분입니다.

2.1 Spark의 Broadcast 메커니즘 이해

• F.broadcast(df)는 실제로 데이터프레임을 즉시 브로드캐스트하는 것이 아니라, 스파크 옵티마이저(optimizer)에게 해당 데이터프레임을 브로드캐스트 힌트(hint)로 제공하는 역할을 합니다.
• 스파크 옵티마이저는 쿼리 계획(query plan)을 생성할 때 이 힌트를 기반으로 조인 전략을 결정합니다. 만약 데이터프레임이 브로드캐스트하기에 충분히 작다고 판단되면, BroadcastHashJoin과 같은 물리적 조인 계획을 선택합니다.
• 실제 브로드캐스트 작업은 해당 조인 작업이 실행되기 전에 트리거됩니다. 이때 스파크는 해당 데이터프레임을 직렬화하여 클러스터의 모든 익스큐터로 한 번만 전송합니다.
• 브로드캐스트된 데이터는 각 익스큐터의 BlockManager에 저장되며, 내부적으로 고유한 브로드캐스트 ID를 갖습니다.

2.2 동시성 환경에서의 재사용

제시된 코드 예시에서 dim 데이터프레임은 스레드 풀이 시작되기 전에 한 번 정의되고 캐시됩니다. process_one 함수 내에서 F.broadcast(dim)을 호출할 때, 모든 스레드는 전역적으로 정의된 동일한 dim 데이터프레임 객체를 참조합니다.

스파크의 옵티마이저는 이러한 상황을 인지하고, 동일한 데이터프레임 객체에 대한 브로드캐스트 요청이 여러 번 발생하더라도 실제 데이터 전송은 한 번만 이루어지도록 최적화합니다. 즉, 첫 번째 작업이 F.broadcast(dim)을 트리거하면 dim 데이터는 익스큐터로 전송되어 캐시되고, 이후의 모든 동시 작업들은 이미 익스큐터에 존재하는 브로드캐스트된 dim 복사본을 재사용합니다.

dim.cache() 호출은 브로드캐스트 데이터 자체의 재사용을 직접적으로 관리하지는 않습니다. 하지만 dim 데이터프레임의 데이터를 스파크 메모리에 미리 로드하여, 브로드캐스트가 필요할 때 데이터를 빠르고 효율적으로 가져올 수 있도록 돕습니다. 이는 브로드캐스트 작업 자체의 성능을 향상시키는 효과가 있습니다.

3. 해결 방법 및 코드 예시

제시된 코드의 접근 방식은 이미 스파크의 브로드캐스트 재사용 메커니즘과 잘 부합하므로, 별도의 “해결 방법”이 필요하기보다는 “최적화된 사용법”에 가깝습니다. 핵심은 dim 데이터프레임이 하나의 인스턴스로 존재하고 모든 동시 작업이 이를 참조하도록 하는 것입니다.

3.1 현재 코드의 효율성 확인

제시된 코드는 이미 최적화된 방식으로 동작합니다. dim 데이터프레임이 스레드 시작 전에 한 번만 로드되고 캐시되었으며, 각 스레드의 process_one 함수는 이 동일한 dim 객체를 F.broadcast()에 전달합니다. 따라서 스파크는 각 익스큐터에 dim 데이터의 단일 브로드캐스트 복사본만 유지하고 이를 모든 동시 작업이 재사용할 것입니다.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from concurrent.futures import ThreadPoolExecutor
import os

# SparkSession 초기화 (예시)
# spark = SparkSession.builder.appName("BroadcastReuse").getOrCreate()

# 예시용 임시 디멘션 테이블 생성
# 실제 환경에서는 spark.read.table("dimension") 사용
dim_data = [(1, "A"), (2, "B"), (3, "C")]
dim_schema = ["id", "name"]
dim = spark.createDataFrame(dim_data, schema=dim_schema)
dim.write.mode("overwrite").saveAsTable("dimension") # 테이블로 저장

# --- 기존 코드와 동일한 최적화된 접근 방식 ---

# 1. 차원 데이터프레임을 스레드 시작 전에 한 번만 로드하고 캐시
dim_df_global = spark.read.table("dimension")
dim_df_global.cache()
dim_df_global.count() # 캐싱 작업 트리거 (액션)

def process_one_optimized(source_id):
    # 각 작업은 고유한 팩트 데이터를 가짐
    fact_data = [(source_id, 1, 100), (source_id, 2, 200)]
    fact_schema = ["source_id", "dim_id", "value"]
    fact = spark.createDataFrame(fact_data, schema=fact_schema)
    
    join_condition = (fact.dim_id == F.col("dim.id"))
    
    # 2. 전역적으로 정의된 동일한 dim_df_global 객체를 F.broadcast()에 전달
    result = fact.join(
        F.broadcast(dim_df_global), # <-- 핵심: 동일 객체 참조
        join_condition,
        "left"
    )
    
    # 실제 output 경로 지정 (예시)
    output_path = f"output_optimized/job_{source_id}"
    result.write.mode("overwrite").parquet(output_path)
    print(f"Processed job: {source_id} to {output_path}")

# 여러 독립적인 워크로드 (예: 15개의 작업)
job_ids = list(range(1, 16))

print("Starting concurrent Spark jobs with shared broadcast...")
with ThreadPoolExecutor(max_workers=15) as executor:
    executor.map(process_one_optimized, job_ids)

print("All concurrent jobs finished.")
# spark.stop() # SparkSession 종료 (필요시)

위 코드에서 dim_df_global은 단 한 번 로드되고 캐시되며, process_one_optimized 함수 내에서 F.broadcast(dim_df_global)이 호출될 때마다 이 전역 객체를 참조합니다. 스파크 옵티마이저는 이를 인지하고 효율적으로 단일 브로드캐스트를 관리합니다.

4. 향후 예방을 위한 팁

4.1 Spark UI를 통해 브로드캐스트 확인

실제로 브로드캐스트가 예상대로 이루어지고 있는지 확인하는 것이 중요합니다. Spark UI의 SQL 탭(SQL/DataFrame)에서 해당 작업의 실행 계획(execution plan)을 살펴보세요. BroadcastHashJoin이 사용되었는지, 그리고 브로드캐스트된 데이터의 크기가 적절한지 확인할 수 있습니다. 데이터가 한 번만 전송되었음을 확인하려면 익스큐터의 메모리 사용량도 모니터링하는 것이 좋습니다.

4.2 `sc.broadcast`와 `F.broadcast`의 차이 이해

• F.broadcast(df) (pyspark.sql.functions.broadcast): 데이터프레임 API에서 조인 힌트(join hint)로 사용됩니다. 스파크 옵티마이저가 최적의 조인 전략을 선택하도록 돕습니다. 데이터프레임 전체를 브로드캐스트하며, 내부적으로 스파크가 데이터 수집 및 직렬화를 관리합니다. 일반적인 데이터프레임 조인에 권장됩니다.
• spark.sparkContext.broadcast(data): 낮은 수준의 RDD API에서 사용되며, 파이썬 객체를 명시적으로 브로드캐스트합니다. 이 경우 data는 드라이버 메모리에 로드되어야 하며, 일반적으로 df.collect()나 df.toPandas()를 통해 수집된 파이썬 리스트나 판다스 데이터프레임 같은 형태여야 합니다. 이 방법은 데이터가 드라이버 메모리에 충분히 들어갈 만큼 매우 작을 때만 사용해야 합니다. sc.broadcast로 생성된 Broadcast 객체는 명시적으로 여러 RDD 작업에 전달하여 재사용을 보장할 수 있습니다.

대부분의 경우 데이터프레임 조인에는 F.broadcast(df)가 더 적절하고 유연합니다. 옵티마이저가 브로드캐스트 여부 및 시기를 결정하게 하여 최적의 성능을 기대할 수 있습니다.

4.3 데이터프레임의 불변성(Immutability) 유지

dim과 같이 브로드캐스트되는 데이터프레임은 여러 동시 작업에서 공유되므로, 해당 데이터프레임이 불변성을 유지하는 것이 중요합니다. 즉, 어떤 스레드나 작업도 dim 데이터프레임의 내용을 변경해서는 안 됩니다. 스파크 데이터프레임은 기본적으로 불변(immutable)이지만, 만약 dim을 기반으로 새로운 데이터프레임을 생성하여 변경한다면, 해당 변경된 데이터프레임은 다른 객체가 되므로 재브로드캐스트될 수 있습니다.

4.4 브로드캐스트에 적합한 데이터 크기 관리

F.broadcast()는 작은 데이터프레임에 대해서만 효과적입니다. 일반적으로 스파크 구성 속성인 spark.sql.autoBroadcastJoinThreshold 값(기본값 10MB)보다 작은 데이터프레임에 적용되도록 옵티마이저가 힌트를 사용합니다. 브로드캐스트하려는 데이터프레임이 너무 크면 오히려 성능 저하를 초래하고 익스큐터 메모리 부족(OOM) 오류를 발생시킬 수 있습니다. 브로드캐스트하려는 데이터프레임의 크기를 항상 염두에 두세요.

이러한 최적화 및 예방 팁을 통해 PySpark 애플리케이션에서 브로드캐스트 조인을 더욱 효율적으로 활용하고, 동시성 환경에서의 메모리 관리를 최적화할 수 있습니다.