[스파크 완벽 가이드] 14장 - 분산형 공유 변수

브로드캐스트 변수

브로드캐스트 변수는 읽기 전용 데이터를 클러스터의 모든 워커 노드에 효율적으로 공유하는 방법을 제공한다.

브로드캐스트 변수를 사용하지 않고, 클로저 함수에 담아 태스크에 데이터를 전달하는 방식은 매번 직렬화가 발생하여 큰 부하를 유발할 수 있다.

하지만 브로드캐스트 변수는 클러스터의 각 워커 노드에 한 번만 전송되며, 각 워커 노드에서 해당 변수를 공유 메모리로 저장하고, 해당 워커 내의 모든 태스크가 그 변수를 참조할 수 있게 됩니다.

 

사용 예시

my_collection = "Spark The Definitive Guide : Big Data Processing Made Simple".split(" ")
words = spark.sparkContext.parallelize(my_collection, 2)
# BroadCast 변수에 사용할 데이터
supplementalData = {"Spark": 1000, "Definitive": 200, "Big": -300, "Simple": 100}
# 데이터를 스파크에 BroadCast
suppBroadcast = spark.sparkContext.broadcast(supplementalData)
# BroadCast된 데이터 값 참조
suppBroadcast.value
# 사용
words.map(lambda word: (word, suppBroadcast.value.get(word, 0))).sortBy(lambda wordPair: wordPair[1]).collect()

>>>
Out[8]: [('Big', -300),
 ('The', 0),
 ('Guide', 0),
 (':', 0),
 ('Data', 0),
 ('Processing', 0),
 ('Made', 0),
 ('Simple', 100),
 ('Definitive', 200),
 ('Spark', 1000)]

 

어큐뮬레이터

브로드캐스트 변수가 읽기 전용 공유 변수라면, 어큐뮬레이터는 쓰기 전용 공유 변수이다.

 

파티션별로 특정 변수의 값을 추적하는 용도로 사용할 수 있고, 병렬 처리 과정에서 효율적으로 사용할 수 있다.

하지만 어큐뮬레이터의 값은 드라이버에서만 읽을 수 있으며, 워커 노드에서는 읽을 수 없다.

보통 카운터(맵리듀스의 카운터), 합계, 오류 발생 횟수와 같은 로그를 기록할 때 사용한다.

 

사용 예시

# 출발지나 도착지가 중국인 항공편의 수를 구하는 어큐뮬레이터 생성

accChina = spark.sparkContext.accumulator(0)

def accChinaFunc(flight_row):
    destination = flight_row["DEST_COUNTRY_NAME"]
    origin = flight_row["ORIGIN_COUNTRY_NAME"]

    if destination == "China":
        accChina.add(flight_row["count"])
    if origin == "China":
        accChina.add(flight_row["count"])


# flights 데이터 셋을 한 줄씩 순회하면서 공유 변수에 값 누적
flights.foreach(lambda flight_row: accChinaFunc(flight_row))

accChina.value
>>>
953

 

 

<어큐뮬레이터를 사용하지 않는다면?>

만약 어큐뮬레이터를 사용하지 않고, 일반적인 파이썬 프로그래밍을 적용하면 어떤 결과가 나올까?

accChinaCnt = 0

def accChinaFunc(flight_row):
    global accChinaCnt
    destination = flight_row["DEST_COUNTRY_NAME"]
    origin = flight_row["ORIGIN_COUNTRY_NAME"]

    if destination == "China":
        accChinaCnt += flight_row["count"]
    if origin == "China":
        accChinaCnt += flight_row["count"]

flights.foreach(lambda flight_row: accChinaFunc(flight_row))
print(accChinaCnt)

>>>
0

 

0이 출력된다.

그 이유는 Spark의 분산 환경에서 워커 노드의 결과값이 드라이버의 전역 변수에 반영되지 않기 때문이다.

즉, 각 워커 노드에서 나의 의도대로 작업을 했어도 그 결과를 알 수 없다.