[Spark] Accumulator와 Broadcast (공유변수)

참고: https://spark.apache.org/docs/latest/rdd-programming-guide.html#shared-variables

공유변수

스파크에서는 분산하여 병렬처리하기 위해 데이터를 분할해서(파티셔닝) 여러 머신에서 동시에 처리한다. 이때 사용된 모든 변수는 배포될 때 복사된 복사본으로 작동하는데, 이러한 변수의 업데이트는 driver에 다시 전달되지 않는다. 그러나 어떤 조건(?) 을 가지고 있어야 데이터를 처리할 수 있는 경우가 발생하는데, 이럴때 제한된 유형의 공유 변수를 사용하여 병렬 처리 후 원하는 값을 받을 수 있다.

 

Broadcast Variable

브로드캐스트 변수를 사용하면 변수의 복사본을 전달하는게 아닌, Read Only 변수를 각 워커 노드에 캐시된 상태로 유지할 수 있다.

액션마다 재전송 없이 사용할 수 있는 변수이다. 브로드캐스트된 후에는 객체를 수정해서는 안된다.

>>> broadcastVar = sc.broadcast([1, 2, 3])
<pyspark.broadcast.Broadcast object at 0x102789f10>

>>> broadcastVar.value
[1, 2, 3]

이렇게 되면 broadcastVar은 모든 워커노드에 캐시된다.

`.unpersist()`: 브로드캐스트 변수가 executor에 복사된 리소스를 해제할 때 호

`.destroy()`: 브로드캐스트 변수에서 사용하는 모든 리소스를 영구적으로 해제할 때 호출

여러 단계의 작업에 동일한 데이터가 필요한 경우에 사용하면 좋다.

 

Accumulator

분산 환경의 executor node 전체에 값을 누적할 수 있는 공유 변수이다.

각 executor 별로 처리한 내역을 Accumulator라는 분산 공유형 변수에 갱신하여 처리한 결과를 추후 driver에서 머지할 수 있다.

주로 데이터를 집계하거나 계산 중 전역 통계를 추적할 때 사용한다.

아래와 같이 초기값을 설정할 수 있다.

accum = sc.accumulator(0)
sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
accum.value
# result: 10

 

일반 변수와 달리 accumulator 변수는 executor가 이 변수에 `add` 등의 메소드를 통해 값을 업데이트할 수는 있지만 값 자체를 읽어올 수는 없고, driver 에서만 `value` 메서드를 사용하여 읽어올 수 있다.

Int 값만 지원하지만 사용자가 직접 Accumulator을 만들수도 있다.

accumulator에 대한 업데이트가 action 내에서 수행되기 때문에 작업 당 한 번만 이루어지도록 보장한다. 즉 retry된 tasks 해당 값을 업데이트하지 않는다. 그러나 transformation 작업 중 accumualtor 값에 접근하려고 하면 에상치 못한 값을 얻을 수 있다.

예를들어 위의 예시같은 경우 `map` 안에서 accumulator의 업데이트를 수행하고 있으나 action이 이루어지지 않았기 때문에 `va` 값은 항상 0이다.

 

AccumulatorParam

스칼라의 경우 `AccumulatorV2`를, 파이썬에서는 `AccumulatorParam`을 상속받아 사용자 정의 Accumulator를 만든다.

AccumulatorParam에서는 `zero`와 `addInPlace`만 구현해주면 되는데, pyspark 예제는 아래와 같다: 출처

from pyspark.accumulators import AccumulatorParam
class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, value):
        return [0.0] * len(value)
    def addInPlace(self, val1, val2):
        for i in range(len(val1)):
             val1[i] += val2[i]
        return val1
va = sc.accumulator([1.0, 2.0, 3.0], VectorAccumulatorParam())
va.value
# result: [1.0, 2.0, 3.0]

def g(x):
    global va
    va += [x] * 3
rdd = sc.parallelize([1,2,3])
rdd.foreach(g)
va.value
# result: [7.0, 8.0, 9.0]

 

`zero(value)`: 제로값을 어떤 값으로 할 것인지 설정한다.

`addInPlace(value1, value2)`: value1은 현재 accumulator에서 가지고 있는 값, value2는 갱신을 요청하는 값이다.

 

executor 별로 partition을 처리하도록 구성할 때 accumulator를 활용할 수 있다.

 

활용 예시

https://medium.com/@ARishi/mastering-accumulators-in-apache-spark-and-not-screwing-yourself-in-the-process-8708cdb4de27

ETL 프로세스 중 데이터 품질 모니터링과 같은 작업이 필요할 때 accumulator를 사용할 수 있다.

예를 들어 데이터 처리 중 유효하지 않은 이메일 주소의 수를 모니터링 하는 경우에, invalidEmails 라는 accumulator를 만들고, 데이터를 읽어올 때 각 row마다 유효하지 않은 이메일을 발견할 때마다 accumulator에 값을 추가한다.