1/23 学习进度笔记

今日学习了如何使用Spark中的累加器功能，并按要求完成了广播变量加上累加器的统计单词的小案例

累加器的功能？

在分布式执行的代码中，累加器accumulator可以进行全局累加

小案例：在下图文本中实现1.统计特殊符号出现次数2.对单词出现进行统计

 原码：

# coding:utf8

from pyspark import SparkContext, SparkConf
import re

if __name__ == '__main__':
    conf = SparkConf().setMaster("local[*]").setAppName("test")
    sc = SparkContext(conf=conf)

    file_rdd = sc.textFile("../data/accumulator_broadcast_data.txt")

    abnormal_char = [",", ".", "!", "#", "$", "%"]

    broadcast = sc.broadcast(abnormal_char)

    acmlt = sc.accumulator(0)

    lines_rdd = file_rdd.filter(lambda line: line.strip())

    data_rdd = lines_rdd.map(lambda line: line.strip())

    words_rdd = data_rdd.flatMap(lambda line: re.split("\s+",line))

    def filter_words(word):
        global acmlt
        abnormal_chars = broadcast.value
        if word in abnormal_chars:
            acmlt+=1
            return False
        else:
            return True

    normal_words_rdd = words_rdd.filter(filter_words)

    words_with_one_rdd = normal_words_rdd.map(lambda x: (x, 1))

    result_rdd = words_with_one_rdd.reduceByKey(lambda x,y: x+y)

    print(result_rdd.collect())
    print("特殊字符数量：",acmlt)