大数据 - Spark系列《十一》- Spark累加器详解-编程知识

Spark系列文章：

大数据 - Spark系列《一》- 从Hadoop到Spark：大数据计算引擎的演进-CSDN博客

大数据 - Spark系列《二》- 关于Spark在Idea中的一些常用配置-CSDN博客

大数据 - Spark系列《三》- 加载各种数据源创建RDD-CSDN博客

大数据 - Spark系列《四》- Spark分布式运行原理-CSDN博客

大数据 - Spark系列《五》- Spark常用算子-CSDN博客

大数据 - Spark系列《六》- RDD详解-CSDN博客

大数据 - Spark系列《七》- 分区器详解-CSDN博客

大数据 - Spark系列《八》- 闭包引用-CSDN博客

大数据 - Spark系列《九》- 广播变量-CSDN博客

大数据 - Spark系列《十》- rdd缓存详解-CSDN博客

1. 简介

累加器用来把Executor端变量信息聚合到Driver端。在 Driver程序中定义的变量，在Executor端的每个Task都会得到这个变量的一份新的副本,每个task更新这些副本的值后，传回 Driver端进行merge。

观察一个问题: 原因是数据在executor端执行完毕以后并没有将acc结果数据返回

def main(args: Array[String]): Unit = {val sc: SparkContext = SparkUtil.getScval rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)var count:Long = 0L//rdd.map(count+=_)rdd.foreach(num=>{count+=num})//计算的结果为0println(count)sc.stop()}

解决方案:应该将每个executor执行的结果数据返回到Driver端进行聚合操作 , 返回最终结果数据

`2. LongAccumulator`

LongAccumulator 是 Spark 中的一种累加器（Accumulator）类型，用于在分布式计算中对长整型（Long）类型的数据进行累加。累加器是一种特殊的共享变量，它可以在各个节点上对其进行添加操作，并将结果汇总到驱动器程序中。

2.1 🥙主要特点：

分布式累加器： LongAccumulator 可以在不同节点上的任务中并行地对其进行添加操作，然后将结果汇总到驱动器程序中。
长整型数据类型：适用于对长整型数据进行累加的场景，如计数器、求和等。
只支持累加操作： LongAccumulator 只支持累加操作，不能进行减法或其他运算。
原子性操作：对累加器的操作是原子性的，可以保证在并发执行的情况下不会发生数据错误。

package com.doit.day0219import com.alibaba.fastjson.JSON
import com.doit.day0126.Movie
import org.apache.spark.{SparkConf, SparkContext}/*** @日期: 2024/2/20* @Author: Wang NaPao* @Blog: https://blog.csdn.net/weixin_40968325?spm=1018.2226.3001.5343* @Tips: 和我一起学习吧* @Description: 使用 Spark 累加器统计解析 JSON 数据失败的次数*/object Test02 {def main(args: Array[String]): Unit = {// 创建SparkConf对象，并设置应用程序名称和运行模式val conf = new SparkConf().setAppName("Starting...") // 设置应用程序名称.setMaster("local[*]") // 设置运行模式为本地模式// 创建SparkContext对象，并传入SparkConf对象val sc = new SparkContext(conf)// 从文件加载 JSON 数据val rdd1 = sc.textFile("Data/movie.json")// 定义计数器变量//var cnt = 0// 使用 spark 内置的全局计数器val cnt1 = sc.longAccumulator("my long Accumulator") // 合并// 遍历 RDD 中的每一行数据rdd1.foreach(line => {try {// 尝试解析 JSON 数据val bean = JSON.parseObject(line, classOf[Movie])println(bean)} catch {case e: Exception => {// JSON 解析失败，增加计数器cnt1.add(1) // 计数1}}})// 打印累加器的值，即解析失败的次数println(cnt1.value) // 合并后的结果}
}

2.2 🥙注意事项：

累加器数据属于全局变量 ,由行动算子触发执行 , 没有触发不执行累加没算
如果多次触发行动算子 , 累加器会执行多次
建议将累加器的变化操作编写在行动算子中

2.3 🥙累加器方法

add(value: T)

向累加器中添加一个值，将这个值与累加器中已有的值进行累加。累加器的值类型必须与添加的值类型相符合。

package com.doit.day0219
import com.alibaba.fastjson.JSON
import com.doit.day0126.Movie
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/*** @日期: 2024/2/20* @Author: Wang NaPao* @Blog: https://blog.csdn.net/weixin_40968325?spm=1018.2226.3001.5343* @Tips: 和我一起学习吧* @Description:*/object Test05 {def main(args: Array[String]): Unit = {// 创建 SparkConf 对象，设置应用程序名称和运行模式val conf = new SparkConf().setAppName("Starting...") // 设置应用程序名称.setMaster("local[*]") // 设置运行模式为本地模式// 创建 SparkContext 对象，传入 SparkConf 对象val sc = new SparkContext(conf)val sumAccumulate = sc.longAccumulator("sumAccumulate")val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5))rdd.foreach(x=>sumAccumulate.add(x))println("累加器的值："+sumAccumulate.value) //15sc.stop()}
}

reset()

重置累加器的值为初始值，通常是零或空。

// 重置累加器的值为初始值
accumulator.reset()

value

获取累加器的当前值。

// 获取累加器的当前值
val currentValue = accumulator.value
println("当前累加器的值：" + currentValue)

2.3 🧀实例1

累加器，主要用于正常业务作业过程中的一些附属信息统计

* （比如程序遇到的脏数据条数，

* 程序处理的数据总行数，

* 程序处理的特定数据条数，

* 程序处理所花费的时长）

业务上需要对如下数据进行统计：比如统计每个city的用户数

"1,Mr.duan,18,beijing"
"2,Mr.zhao,28,beijing"
"b,Mr.liu,24,shanghai"
"4,Mr.nai,22,shanghai"
"a,Mr.liu,24,shanghai"
"6,Mr.ma"

同时，还想在业务统计的过程中，附带统计出原始数据中的脏数据条数，并按多种不正确的格式进行分别统计，如：id字段无法数字化的条数，字段数量不够的条数，其他不正确的条数

package com.doit.day0219import com.alibaba.fastjson.JSON
import com.doit.day0126.Movie
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}/*** @日期: 2024/2/20* @作者: Wang NaPao* @博客: https://blog.csdn.net/weixin_40968325?spm=1018.2226.3001.5343* @Tips: 和我一起学习吧* @描述: 这个对象包含了一个 Spark 应用程序的入口点，用于处理从文件加载 JSON 数据的场景，统计每个城市的用户数量，并且附带统计原始数据中的脏数据条数。*/
object Test03 {def main(args: Array[String]): Unit = {// 创建 SparkConf 对象，设置应用程序名称和运行模式val conf = new SparkConf().setAppName("Starting...") // 设置应用程序名称.setMaster("local[*]") // 设置运行模式为本地模式// 创建 SparkContext 对象，传入 SparkConf 对象val sc = new SparkContext(conf)// 从文件加载 JSON 数据val rdd1 = sc.textFile("Data/city.txt")// 创建一个累加器用于统计脏数据条数val cnt1 = sc.longAccumulator("dirtyDataCount")// 对 RDD 进行处理：将每行数据拆分为数组，判断数组长度，若为4则返回 (城市, 1)，否则更新累加器并返回 Noneval rdd2 = rdd1.map(line => {try {val arr1 = line.split(",")val no = arr1(0).toIntval name = arr1(1)val age = arr1(2).toIntval city = arr1(3)(city,1)} catch {case exception: Exception => {cnt1.add(1)("error", 1)}}}).filter(e=>{e._1!="error"}) // 过滤掉脏数据.reduceByKey(_ + _) // 对相同键的值进行累加.foreach(println) // 打印结果// 打印脏数据条数println("脏数据条数为：" + cnt1.value)}
}