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

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)