PySpark Broadcast和Accumulator
对于并行处理,Apache Spark使用共享变量。当驱动程序向集群上的执行器发送任务时,共享变量的副本会出现在集群的每个节点上,这样它就可以用来执行任务。
有两种类型的共享变量被Apache Spark支持 —
- Broadcast
- Accumulator
让我们详细了解它们。
Broadcast
广播变量是用来保存所有节点上的数据副本。这个变量被缓存在所有的机器上,而不是在有任务的机器上发送。下面的代码块是PySpark的Broadcast类的细节。
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的Accumulator类的细节。
class pyspark.Accumulator(aid, value, accum_param)
下面的例子显示了如何使用一个Accumulator变量。一个Accumulator变量有一个名为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