apache-spark
공유 변수

브로드 캐스트 변수

브로드 캐스트 변수는 SparkContext.broadcast 메서드로 만들 수있는 읽기 전용 공유 객체입니다.

val broadcastVariable = sc.broadcast(Array(1, 2, 3))

value 방법을 사용하여 읽음 :

val someRDD = sc.parallelize(Array(1, 2, 3, 4))

someRDD.map(
    i => broadcastVariable.value.apply(i % broadcastVariable.value.size)
)

축전지

Accumulators는 쓰기 전용 변수이며 SparkContext.accumulator 로 만들 수 있습니다.

val accumulator = sc.accumulator(0, name = "My accumulator") // name is optional

+= 수정 됨 :

val someRDD = sc.parallelize(Array(1, 2, 3, 4))
someRDD.foreach(element => accumulator += element)

value 메소드를 사용하여 액세스 :

accumulator.value // 'value' is now equal to 10

어큐뮬레이터를 사용하는 것은 스파크가 적어도 한번은 변형을 보증했기 때문에 복잡합니다. 어떤 이유로 든 변환을 다시 계산해야하는 경우 해당 변환 중에 누적 기가 업데이트됩니다. 이는 누적 기 값이 작업이 한 번만 실행 된 경우와 다를 수 있음을 의미합니다.

노트 :

1. 실행자는 누적 기의 값을 읽을 수 없습니다 . 드라이버 프로그램 만이 값 메서드를 사용하여 누적 기의 값을 읽을 수 있습니다.
2. Java / MapReduce의 카운터와 거의 유사합니다. 따라서 누적기를 카운터에 연결하여 쉽게 이해할 수 있습니다.

스칼라의 사용자 정의 누산기

AccumulatorParam 정의

import org.apache.spark.AccumulatorParam

object StringAccumulator extends AccumulatorParam[String] {
  def zero(s: String): String = s
  def addInPlace(s1: String, s2: String)=  s1 + s2
}

용도:

val accumulator = sc.accumulator("")(StringAccumulator)
sc.parallelize(Array("a", "b", "c")).foreach(accumulator += _)

파이썬의 사용자 정의 누산기

AccumulatorParam 정의 :

from pyspark import AccumulatorParam

class StringAccumulator(AccumulatorParam):
    def zero(self, s):
        return s
    def addInPlace(self, s1, s2):
        return s1 + s2

accumulator = sc.accumulator("", StringAccumulator())

def add(x): 
    global accumulator
    accumulator += x

sc.parallelize(["a", "b", "c"]).foreach(add)