PySpark - 廣播變數 & 累加器

為了並行處理，Apache Spark 使用共享變數。當驅動程式向叢集中的執行器傳送任務時，共享變數的副本會複製到叢集的每個節點上，以便用於執行任務。

Apache Spark 支援兩種型別的共享變數：

• 廣播變數 (Broadcast)
• 累加器 (Accumulator)

讓我們詳細瞭解它們。

廣播變數 (Broadcast)

廣播變數用於在所有節點上儲存資料的副本。此變數快取在所有機器上，不會發送到具有任務的機器上。以下程式碼塊包含 PySpark 廣播類的詳細資訊。

class pyspark.Broadcast (
   sc = None, 
   value = None, 
   pickle_registry = None, 
   path = None
)

以下示例顯示如何使用廣播變數。廣播變數具有一個名為 value 的屬性，該屬性儲存資料並用於返回廣播值。

----------------------------------------broadcast.py--------------------------------------
from pyspark import SparkContext 
sc = SparkContext("local", "Broadcast app") 
words_new = sc.broadcast(["scala", "java", "hadoop", "spark", "akka"]) 
data = words_new.value 
print "Stored data -> %s" % (data) 
elem = words_new.value[2] 
print "Printing a particular element in RDD -> %s" % (elem)
----------------------------------------broadcast.py--------------------------------------

命令 - 廣播變數的命令如下：

$SPARK_HOME/bin/spark-submit broadcast.py

輸出 - 以下命令的輸出如下所示。

Stored data -> [
   'scala',  
   'java', 
   'hadoop', 
   'spark', 
   'akka'
]
Printing a particular element in RDD -> hadoop

累加器 (Accumulator)

累加器變數用於透過關聯和交換運算聚合資訊。例如，您可以將累加器用於求和運算或計數器（在 MapReduce 中）。以下程式碼塊包含 PySpark 累加器類的詳細資訊。

class pyspark.Accumulator(aid, value, accum_param)

以下示例顯示如何使用累加器變數。累加器變數具有一個名為 value 的屬性，類似於廣播變數。它儲存資料並用於返回累加器的值，但僅可在驅動程式程式中使用。

在此示例中，累加器變數由多個工作器使用並返回累積值。

----------------------------------------accumulator.py------------------------------------
from pyspark import SparkContext 
sc = SparkContext("local", "Accumulator app") 
num = sc.accumulator(10) 
def f(x): 
   global num 
   num+=x 
rdd = sc.parallelize([20,30,40,50]) 
rdd.foreach(f) 
final = num.value 
print "Accumulated value is -> %i" % (final)
----------------------------------------accumulator.py------------------------------------

命令 - 累加器變數的命令如下：

$SPARK_HOME/bin/spark-submit accumulator.py

輸出 - 以上命令的輸出如下所示。

Accumulated value is -> 150