使用pyspark 建立spark 的累加器

一，累加器特征

1，PySpark累加器是一个共享变量，与RDD和DataFrame一起使用，以执行与Map reduce计数器类似的求和和和计数器操作。

2，只有分布在各个节点上的task任务才能更新累加器的数值，并且只有driver 端可以读取数值。

二，累加器的创建和使用

sparkContext.accumulator() 可以定义累加器

add() function 增加或者更新累加器的值

value 属性（累加器中的）可以读取到值。

代码：

import pyspark

from pyspark.sql import SparkSession

spark=SparkSession.builder.appName("accumulator").getOrCreate()

accum=spark.sparkContext.accumulator(0)

rdd=spark.sparkContext.parallelize([1,2,3,4,5])

rdd.foreach(lambda x:accum.add(x))   #统计rdd内的数值和

print(accum.value)   # driver program 中获取值

accuSum=spark.sparkContext.accumulator(0)

def countFun(x):    # 功能同上，使用函数方法

    global accuSum

    accuSum+=x

rdd.foreach(countFun)

print(accuSum.value)

accumCount=spark.sparkContext.accumulator(0)

rdd2=spark.sparkContext.parallelize([1,2,3,4,5])

rdd2.foreach(lambda x:accumCount.add(1))   #作为计数器功能

print(accumCount.value)

巴特西

使用pyspark 建立spark 的累加器

最新文章

热门文章