package Demo
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
**
@author 不温卜火
**
* @create 2020-08-01 12:18
**
* MyCSDN : https://buwenbuhuo.blog.csdn.net/
*
*/
object AccDemo1 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
val sc = new SparkContext(conf)
val p1 = Person(10)
// 将来会把对象序列化之后传递到每个节点上
val rdd1 = sc.parallelize(Array(p1))
val rdd2: RDD[Person] = rdd1.map(p => {p.age = 100; p})
rdd2.count()
// 仍然是 10
println(p1.age)
}
}
case class Person(var age:Int)
正常情况下, 传递给 Spark 算子(比如: map, reduce 等)的函数都是在远程的集群节点上执行, 函数中用到的所有变量都是独立的拷贝.
这些变量被拷贝到集群上的每个节点上, 都这些变量的更改不会传递回驱动程序.
支持跨 task 之间共享变量通常是低效的, 但是 Spark 对共享变量也提供了两种支持:
累加器用来对信息进行聚合,通常在向 Spark 传递函数时,比如使用 map() 函数或者用 filter() 传条件时,可以使用驱动器程序中定义的变量,但是集群中运行的每个任务都会得到这些变量的一份新的副本,所以更新这些副本的值不会影响驱动器中的对应变量。
如果我们想实现所有分片处理时更新共享变量的功能,那么累加器可以实现我们想要的效果。
累加器是一种变量, 仅仅支持“add”, 支持并发. 累加器用于去实现计数器或者求和. Spark 内部已经支持数字类型的累加器, 开发者可以添加其他类型的支持.
需求:计算文件中空行的数量
package Demo
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}
/**
**
@author 不温卜火
**
* @create 2020-08-01 12:22
**
* MyCSDN : https://buwenbuhuo.blog.csdn.net/
*
*/
object AccDemo2 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.textFile("D:/words.txt")
// 得到一个 Long 类型的累加器. 将从 0 开始累加
val emptyLineCount: LongAccumulator = sc.longAccumulator
rdd.foreach(s => if (s.trim.length == 0) emptyLineCount.add(1))
println(emptyLineCount.value)
}
}
通过继承类AccumulatorV2来自定义累加器.
下面这个累加器可以用于在程序运行过程中收集一些文本类信息,最终以List[String]的形式返回。
package Demo
import java.util
import java.util.{ArrayList, Collections}
import org.apache.spark.util.AccumulatorV2
/**
**
@author 不温卜火
**
* @create 2020-08-01 12:56
**
* MyCSDN : https://buwenbuhuo.blog.csdn.net/
*
*/
object MyAccDemo {
def main(args: Array[String]): Unit = {
}
}
class MyAcc extends AccumulatorV2[String, java.util.List[String]] {
private val _list: java.util.List[String] = Collections.synchronizedList(new ArrayList[String]())
override def isZero: Boolean = _list.isEmpty
override def copy(): AccumulatorV2[String, util.List[String]] = {
val newAcc = new MyAcc
_list.synchronized {
newAcc._list.addAll(_list)
}
newAcc
}
override def reset(): Unit = _list.clear()
override def add(v: String): Unit = _list.add(v)
override def merge(other: AccumulatorV2[String, util.List[String]]): Unit =other match {
case o: MyAcc => _list.addAll(o.value)
case _ => throw new UnsupportedOperationException(
s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
}
override def value: util.List[String] = java.util.Collections.unmodifiableList(new util.ArrayList[String](_list))
}
package Demo
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
**
*
*@author 不温卜火
**
* @create 2020-08-01 12:57
**
* MyCSDN : https://buwenbuhuo.blog.csdn.net/
*
*/
object MyAccDemo1 {
def main(args: Array[String]): Unit = {
val pattern = """^\d+$"""
val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
val sc = new SparkContext(conf)
// 统计出来非纯数字, 并计算纯数字元素的和
val rdd1 = sc.parallelize(Array("abc", "a30b", "aaabb2", "60", "20"))
val acc = new MyAcc
sc.register(acc)
val rdd2: RDD[Int] = rdd1.filter(x => {
val flag: Boolean = x.matches(pattern)
if (!flag) acc.add(x)
flag
}).map(_.toInt)
println(rdd2.reduce(_ + _))
println(acc.value)
}
}
注意: 在使用自定义累加器的不要忘记注册sc.register(acc)
广播变量在每个节点上保存一个只读的变量的缓存, 而不用给每个 task 来传送一个 copy.
例如, 给每个节点一个比较大的输入数据集是一个比较高效的方法. Spark 也会用该对象的广播逻辑去分发广播变量来降低通讯的成本.
广播变量通过调用SparkContext.broadcast(v)来创建. 广播变量是对v的包装, 通过调用广播变量的 value方法可以访问.
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)
scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)
本次的分享就到这里了