PySpark累加器测试报错：已声明全局变量仍提示'acc'未定义

你遇到的这个问题其实有两个核心原因：

1. 错误使用global关键字：你的acc是定义在test函数内部的局部变量，并不是模块级别的全局变量。在imap里用global acc会让Python去寻找模块全局的acc，而非test函数里的那个，自然会报找不到变量的错误。
2. Spark分布式执行的特性：即使变量引用正确，Spark会把imap函数序列化后发送到各个worker节点执行，不过Spark的累加器本身支持序列化，只要变量引用逻辑正确就能正常工作。

正确代码示例 #

from pyspark import SparkConf, SparkContext
from pyspark.sql import Row

def test():
    conf = SparkConf().setAppName("test").setMaster("local[*]")
    # 用with语句自动管理SparkContext生命周期，避免资源泄漏
    with SparkContext(conf=conf) as sc:
        rdds = sc.parallelize([
            Row(user="spark", item="book"),
            Row(user="spark", item="goods"),
            Row(user="hadoop", item="book"),
            Row(user="python", item="duck")
        ])
        
        acc = sc.accumulator(0)
        print("Initial accumulator value: {}".format(acc.value))

        # 用nonlocal引用外层test函数的acc变量，替代错误的global声明
        def imap(row):
            nonlocal acc
            acc += 1
            return row

        # 触发任务执行（map是懒加载，foreach是动作操作才会真正运行）
        processed_rdd = rdds.map(imap)
        processed_rdd.foreach(print)
        
        print("Final accumulator value: {}".format(acc.value))

if __name__ == "__main__":
    test()

关键说明 #

• nonlocal的作用：因为acc是外层test函数的局部变量，嵌套的imap函数需要用nonlocal声明要修改外层函数的变量，这样Python才能正确定位到test里的累加器对象。
• 累加器的工作逻辑：Spark会把每个worker节点上的累加结果汇总到driver端，所以最终打印的acc.value会是所有节点累加后的总和（这里对应4条数据，结果应为4）。
• 资源管理规范：用with语句包裹SparkContext，能确保任务结束后自动关闭上下文，避免不必要的资源占用。

内容的提问来源于stack exchange，提问作者littlely