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