来一个需求：对搜狗日志的数据进行处理，统计10点搜索的数据一共有多少条

错误写法：使用普通全局变量计数

普通全局变量 = 每人一本私人小本子，自己写完自己留着，不上交，老板看不到总数。

错误代码展示

"""
------------------------------------------
  SourceFile : 04_累加器
  Author  : xiaoyang
-------------------------------------------
"""
import os
import re

from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel

if __name__ == '__main__':
    os.environ['JAVA_HOME'] = r'D:\app\jdk\jdk1.8.0_181'
    os.environ['HADOOP_HOME'] = r'D:\app\hadoop\hadoop3.1.4\hadoop-3.1.4'
    os.environ['PYSPARK_PYTHON'] = r'C:\Users\86184\miniconda3\python.exe'
    os.environ['PYSPARK_DRIVER_PYTHON'] = r'C:\Users\86184\miniconda3\python.exe'
    conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")
    sc = SparkContext(conf=conf)
    # 读取搜狗搜索日志，统计早上10点一共有多少条搜索记录。
    rdd1=sc.textFile("../../datas/sparkdata/sogou.tsv",minPartitions=8)
    rdd2=rdd1.filter(lambda line:len(re.split("\s+",line))==6)\
    .map(
        lambda line:(re.split("\s+",line)[0],re.split("\s+",line)[1],re.split("\s+",line)[2][1:-1]
        )
    ).persist(StorageLevel.MEMORY_AND_DISK_2)
    # 统计一天每小时点击量并按照点击量降序排序
    _sum = 0
    def sumTotalLine(tuple1):
        global _sum  # 把_sum 设置为全局变量
        timeStr = tuple1[0]  # 10:19:18
        if timeStr[0:2] == '10':
            _sum += 1
    rdd2.foreach(lambda tuple1: sumTotalLine(tuple1))
    print(_sum)  # 结果是0

    sc.stop()

在 Driver 主线程定义一个初始值为 0 的变量，在分布式任务里做累加。

报错原因

1. total=0 写在主程序 Driver里
2. Spark 分布式运行时，会拆分出很多子任务 Task去干活
3. 每个子任务都会单独复制一份 total 变量，变成自己私有变量
4. 子任务只修改自己手里复制的数字，不会把累加结果传回主程序
5. 最后打印的还是主程序最原始的 0，统计彻底失效

正确解决方案：Spark 累加器 Accumulator

1. 累加器核心作用

专门用来做分布式全局统计，求和、计数、累加数值，实现所有子任务计算结果统一汇总到主程序。

2. 工作原理

1. 主程序 Driver 创建公共累加计数器
2. 所有分布式子任务共用这一个累加器
3. 子任务内部完成数据判断，执行累加操作
4. 所有任务执行完毕，自动把所有累加结果汇总合并，传回主程序
5. 主程序直接读取最终统计数值

3. 通俗总结

累加器 = 全班共用一本总账本，所有人往上面记账，最后班长统一查看总账总数。

累加器固定使用三步法

• 创建累加器（主程序内）累加器名 = sc.accumulator(初始数值)
• 执行累加操作（算子内部）累加器名.add(累加数值)
• 获取最终结果（任务结束后）累加器名.value

正确代码展示

"""
------------------------------------------
  SourceFile : 04_累加器
  Author  : xiaoyang
-------------------------------------------
"""
import os
import re

from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel

if __name__ == '__main__':
    os.environ['JAVA_HOME'] = r'D:\app\jdk\jdk1.8.0_181'
    os.environ['HADOOP_HOME'] = r'D:\app\hadoop\hadoop3.1.4\hadoop-3.1.4'
    os.environ['PYSPARK_PYTHON'] = r'C:\Users\86184\miniconda3\python.exe'
    os.environ['PYSPARK_DRIVER_PYTHON'] = r'C:\Users\86184\miniconda3\python.exe'
    conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")
    sc = SparkContext(conf=conf)
    # 统计一天每小时点击量并按照点击量降序排序
    # 创建累加器，初始值为0
    ten_count=sc.accumulator(0)
    # 读取搜狗搜索日志，统计早上10点一共有多少条搜索记录。
    rdd1 = sc.textFile("../../datas/sparkdata/sogou.tsv", minPartitions=8)
    rdd2 = rdd1.filter(lambda line: len(re.split("\s+", line)) == 6) \
        .map(
        lambda line: (re.split("\s+", line)[0], re.split("\s+", line)[1], re.split("\s+", line)[2][1:-1]
                      )
    ).persist(StorageLevel.MEMORY_AND_DISK_2)
    # 遍历判断，符合条件就累加
    def judge_hour(hour):
        time_str=hour[0]
        hour1=time_str[0:2]
        if hour1 =="10":
            ten_count.add(1)
    rdd2.foreach(judge_hour)
    print("总条数："+str(ten_count.value))#总条数104694
    sc.stop()

累加器使用注意事项

1. 累加器只能做累加操作，不能随意修改赋值
2. 累加器取值.value只能放在算子外面，不能放在遍历逻辑里
3. 适合场景：数据条数统计、数值求和、日志条数统计等全局汇总业务
4. 属于 Spark 两大共享变量之一：广播变量 + 累加器
   • 广播变量：只读共享大数据
   • 累加器：分布式全局计数累加