累加器(accumulator)

功能

• 实现在Driver端和Executor端共享变量 写的功能

实现机制

• Driver端定义的变量,在Executor端的每个Task都会得到这个变量的副本;
  在每个Task对自己内部的变量副本值更新完成后,传回给Driver端,然后将每个变量副本的值进行累计操作;

触发/生效 时机

• 受惰性求值机制的影响,只有在行动算子执行时 累加器才起作用;

使用地方

• 最好只在行动算子中使用,不要在转换算子中使用,因为转换算子可能出现失败时会重试,这时对应的累加器的值也会重试,这样累加器的值就是脏写;

使用场景

• 需要 求和 或 计数时;

注意事项

• 在对同一个rdd执行多次执行 行动算子时可能会导致 累加器 多次重复计算,导致累加器的结果错误;可以 通过在转换算子后面添加cache()解决;

一般思维下定义的无效示例

# -*- coding: utf-8 -*-
"""
(C) rgc
All rights reserved
create time '2021/5/30 20:06'

Usage:
此处累加器失效的原因是 Driver端定义了累加器,将Driver端的累加器序列化到Executor端,这时是对Executor端的累加器进行写操作;
结果没有同步到Driver端,所以Driver端累加器的值仍然是0
"""
# 构建spark
from pyspark.conf import SparkConf
from pyspark.context import SparkContext

conf = SparkConf()
# 使用本地模式;且 executor设置为1个方便debug
conf.setMaster('local[1]').setAppName('rgc')
sc = SparkContext(conf=conf)
# 偶数累加器
even_num_acc = 0
# 奇数累加器
uneven_num_acc = 0
rdd = sc.parallelize([2, 1, 3, 4, 4], 1)


def map_func(x: int) -> tuple:
    """
    将每个元素转为元祖
    :param x: rdd中每个元素
    :return:
    """
    global even_num_acc
    global uneven_num_acc
    # 偶数
    if x % 2 == 0:
        even_num_acc += 1
    else:
        uneven_num_acc += 1
    return (x, 1)


# map操作
map_rdd = rdd.map(map_func)
print(map_rdd.collect())  # [(2, 1), (1, 1), (3, 1), (4, 1), (4, 1)]
print('偶数累加器', even_num_acc)  # 0
print('奇数累加器', uneven_num_acc)  # 0

累加器用法

# -*- coding: utf-8 -*-
"""
(C) rgc
All rights reserved
create time '2021/5/30 20:06'

Usage:
累加器:实现在Driver端和Executor端共享变量 写的功能
实现机制:Driver端定义的变量,在Executor端的每个Task都会得到这个变量的副本,在每个Task对自己内部的变量副本值更新完成后,传回给Driver端,然后将每个变量副本的值进行累计操作;
触发时机:只有在行动算子执行时 累加器才起作用;
使用地方:最好只在行动算子中使用,不用在转换算子中使用,因为转换算子可能出现失败时会重试,这时对应的累加器的值也会重试,这样累加器的值就是脏写;
使用场景:
1.需要 求和 或 计数时;

注意事项:
1.在对同一个rdd执行多次行动算子时可能会导致在 转换算子中的 累加器 多次重复计算,导致累加器的结果错误;可以 通过在转换算子后面添加cache()解决;
"""
# 构建spark
from pyspark.conf import SparkConf
from pyspark.context import SparkContext

conf = SparkConf()
# 使用本地模式;且 executor设置为1个方便debug
conf.setMaster('local[1]').setAppName('rgc')
sc = SparkContext(conf=conf)
# 偶数累加器
even_num_acc = sc.accumulator(0)
# 奇数累加器
uneven_num_acc = sc.accumulator(0)
rdd = sc.parallelize([2, 1, 3, 4, 4], 1)
rdd1 = sc.parallelize([2, 1, 3, 4, 4], 1)


def map_func(x: int) -> tuple:
    """
    将每个元素转为元祖
    :param x: rdd中每个元素
    :return:
    """
    global even_num_acc
    global uneven_num_acc
    # 偶数
    if x % 2 == 0:
        even_num_acc += 1
    else:
        uneven_num_acc += 1
    return (x, 1)


# 操作算子 添加cache的map操作
map_rdd = rdd.map(map_func).cache()
print(map_rdd.collect())  # [(2, 1), (1, 1), (3, 1), (4, 1), (4, 1)]
print('操作算子 添加cache的map操作 偶数累加器', even_num_acc)  # 3
print('操作算子 添加cache的map操作 奇数累加器', uneven_num_acc)  # 2

print(map_rdd.collect())  # [(2, 1), (1, 1), (3, 1), (4, 1), (4, 1)]
print('操作算子 添加cache的map操作 偶数累加器', even_num_acc)  # 3
print('操作算子 添加cache的map操作 奇数累加器', uneven_num_acc)  # 2

print('')

# 将累加器的值 置零
even_num_acc.value = 0
uneven_num_acc.value = 0

# 操作算子 未添加cache的map操作
map_rdd = rdd1.map(map_func)
print(map_rdd.collect())  # [(2, 1), (1, 1), (3, 1), (4, 1), (4, 1)]
print('操作算子 未添加cache的map操作 偶数累加器', even_num_acc)  # 3
print('操作算子 未添加cache的map操作 奇数累加器', uneven_num_acc)  # 2

print(map_rdd.collect())  # [(2, 1), (1, 1), (3, 1), (4, 1), (4, 1)]
print('操作算子 未添加cache的map操作 偶数累加器', even_num_acc)  # 6
print('操作算子 未添加cache的map操作 奇数累加器', uneven_num_acc)  # 4

结果:

自定义累加器

实现计算list中每个值出现的次数,用dict表示出来

# -*- coding: utf-8 -*-
"""
(C) rgc
All rights reserved
create time '2021/5/30 20:06'

Usage:

"""
# 构建spark
from pyspark import AccumulatorParam
from pyspark.conf import SparkConf
from pyspark.context import SparkContext


class MyAccum(AccumulatorParam):

    def zero(self, value):
        """
        task内部累加操作时的 初始化 默认值
        :param value:
        :return:
        """
        return {}

    @classmethod
    def dict_add(cls, a: dict, b: dict) -> dict:
        """
        用户自定义方法
        2个dict的value相加
        :param a:
        :param b:
        :return:
        Usage:
        >>> a = {'c': 2, 'e': 1, 'd': 1}
        >>> b = {'e': 2, 'f': 2}
        >>> dict_add(a, b) # {'c': 2, 'e': 3, 'd': 1, 'f': 2}
        """
        b_key_list = list(b.keys())
        for k in b_key_list:
            if k in a:
                a[k] += b[k]
            else:
                a[k] = b[k]
        return a

    def addInPlace(self, value1, value2: str or dict) -> dict:
        """
        实现父类的方法
        此方法需要实现 分区间第一次的操作;分区间非第一次的操作;分区内每个task内部的操作 3个部分 才能保证不报错
        :param value1: 上一次 累加器的值
        :param value2: 这次新增的数据
        :return:
        """
        # 此处主要在 Driver端 分区之间第一次进行操作时,这时value1默认为空,所以新的值直接为value2
        if value1 == "":
            print('Driver端第一次操作', f'value1:{value1},value2:{value2}')
            return value2
        # 此处主要在 Driver端 分区之间 非第一次 进行 操作;只有在 分区个数>=2时才执行到此处
        # value1是dict类型,如 {'a':1,'b':2} 表示 之前 分区之间进行累加器操作的结果dict
        # value2也是dict类型,如 {'a':1,'b':2} 表示 最新 分区的累加器的结果dict
        if isinstance(value2, dict):
            # rdd 可能会被分割成多份并行计算,所以这里处理当 value2 为某部分 rdd 计算得到的值
            value = self.dict_add(value1, value2)
            print('Driver端非第一次操作', value1, value2, value)
            return value
        else:
            # 此处主要在 Execturo端 每个task内部 进行操作
            # value1是dict类型,如 {'a':1,'b':2}
            # value2是str类型,也就是rdd中每个元素的值;如 'a' 或 'b' 或 'c'
            # 如果 rdd中的元素在 累加器的 dict类型的值中,则加一;不在 则设置为1
            print('Executor', value1, value2)
            if value1.get(value2) is not None:
                value1[value2] += 1
            else:
                value1[value2] = 1
            # 返回最新的 累加器的值
            return value1


conf = SparkConf()
# 使用本地模式;且 executor设置为1个方便debug
conf.setMaster('local[1]').setAppName('rgc')
sc = SparkContext(conf=conf)
accum = sc.accumulator("", accum_param=MyAccum())
rdd = sc.parallelize(["a", "b", "a", "c", "e", "d", "c"], 2)

# accum.add()操作实际调用的就是 addInPlace(value,x)方法
rdd = rdd.map(lambda x: accum.add(x))
rdd.count()
print(accum.value, 'result')
assert accum.value == {'a': 2, 'b': 1, 'c': 2, 'e': 1, 'd': 1}

结果:

自定义累加器实现注意点

• 理解 累加器实现机制
• 继承自 AccumulatorParam类,实现其 zero,addInPlace 2个方法
• zero方法 用来设置 Executor端 task内部累加操作时的 初始化 默认值(不是Driver端分区间操作的默认值)
• addInPlace方法 需要实现 分区间第一次的操作;分区间非第一次的操作;分区内每个task内部的操作 3个部分 才能保证不报错

相关链接

• https://blog.csdn.net/qq_41489540/article/details/110003165
• https://blog.csdn.net/zlbingo/article/details/112635574
• https://waterandair.github.io/2018-04-03-pyspark-custom-accumulator