Spark基础：（五）Spark编程进阶

 

共享变量

（1）累加器：是用来对信息进行聚合的，同时也是Spark中提供的一种分布式的变量机制，其原理类似于mapreduce，即分布式的改变，然后聚合这些改变。累加器的一个常见用途是在调试时对作业执行过程中的事件进行计数。 Spark内置的提供了Long和Double类型的累加器

object AccumulatorDemo {
  def main(args: Array[String]): Unit = {
    //通过conf创建sc 创建Spark配置对象
    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("AccumulatorDemo"))
    val file = sc.textFile("file:///F:/spark/c.txt")
    val blankLines = sc.longAccumulator("blankLines");
    val info = file.flatMap(line => {
      if (line == "") {
        blankLines.add(1L)
      }
      line.split(" ")
    })
    info.saveAsTextFile("file:///F:/spark/out1")
    println("BlankLines= " + blankLines)
  }
}

此代码用于实现累加文档中的空行数目。输出

BlankLines= LongAccumulator(id: 0, name: Some(blankLines), value: 0)

**需要注意的是：**XXXXAccumulator也是lazy的，在行动操作前的转化操作是不会进行求值的。
这里用一个更好的例子来解释这个lazy的效果。

    val accum = sc.longAccumulator("AccumulatorDemo")
    val numberRDD = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),2).map(n=>{
      accum.add(1L)
      n+1
    })
    numberRDD.count
    println("accum1:"+accum.value)
    numberRDD.reduce(_+_)
    println("accum2: "+accum.value)

结果如下

accum1:9
accum2: 18

可见，虽然在map里面进行了累加器加1的操作，两个累加值却不一样，这是因为count和reduce都是action，而且第一次count的action并没有持久化，此时的accum的value已经是9了，在下一次的reduce的action中依然要重新计算一次，相当于提交了两次作业，那reduce执行之后accume的value将会从9变为18。如果我们在count的action之前调用persist()(cache())进行持久化，这样的话reduce的action就不需要从头计算了，两次的accum的就会一样了。

自定义累加器：

isZero: 判断是否为初始值
copy:拷贝累加器
add: 操作数据累加方法实现
merge: 合并数据
value: AccumulatorV2对外访问的数据结果 “`

import org.apache.spark.util.AccumulatorV2
 
/**
  * 实现字符串拼接
  * */
class MyAccumulator  extends AccumulatorV2[String, String] {
 
  private var res=""
  override def isZero: Boolean = {
    res==""
  }
 
  override def copy(): AccumulatorV2[String, String] ={
    val newAcc=new MyAccumulator()
    newAcc.res=this.res
    newAcc
  }
 
  override def reset(): Unit = {
    res=""
  }
 
  override def add(v: String): Unit = {
    res+=v+'-'
  }
 
  override def merge(other: AccumulatorV2[String, String]): Unit = other match{
    case o:MyAccumulator => res+=o.res
    case _=>throw new UnsupportedOperationException(
      s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
  }
 
  override def value: String = res
}

调用实现

def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf()
    .setAppName("Accumulator1")
    .setMaster("local"))
 
    val myAcc = new MyAccumulator
    sc.register(myAcc,"myAcc")
    //val acc = sc.longAccumulator("avg")
    val nums = Array("a","b","c","d","e","f","h","i")
    val numsRdd = sc.parallelize(nums)
    numsRdd.foreach(num => myAcc.add(num))
    println(myAcc)
    sc.stop()
  }

执行结果

MyAccumulator(id: 0, name: Some(myAcc), value: a-b-c-d-e-f-h-i-)

(2)广播变量
首先要简单来了解一下闭包的概念：函数可以访问函数外面的变量，但是函数内对变量的修改，在函数外是不可见的。（闭包是一个函数，它返回值取决于在此函数之外声明的一个或多个变量的值。）

Spark的第二种共享变量类型是广播变量，它可以让程序高效的向所有的工作节点发送一个较大的只读值，以供一个或者多个spark操作来使用，广播变量可以解决闭包函数引用外部大变量引起的性能问题，广播变量将只读变量缓存在每个worker节点中，Spark使用了高效广播算法分发变量从而提高通信性能。

广播变量的优势：是因为不是每个task一份变量副本，而是变成每个节点的executor才一份副本。这样的话，就可以让变量产生的副本大大减少。

广播变量，初始的时候，就在Drvier上有一份副本。
task在运行的时候，想要使用广播变量中的数据，此时首先会在自己本地的Executor对应的BlockManager中，尝试获取变量副本；如果本地没有BlockManager，也许会从远程的Driver上面去获取变量副本；也有可能从距离比较近的其他节点的Executor的BlockManager上去获取，并保存在本地的BlockManager中；BlockManager负责管理某Executor对应的内存和磁盘上的数据，此后这个executor上的task，都会直接使用本地的BlockManager中的副本。

例如，50个executor，1000个task。一个map，10M：
默认情况下，1000个task，1000份副本。10G的数据，网络传输，在集群中，耗费10G的内存资源。
如果使用了广播变量。50个execurtor，50个副本。500M的数据，网络传输，而且不一定都是从Driver传输到每个节点，还可能是就近从最近的
节点的executor的bockmanager上拉取变量副本，网络传输速度大大增加；500M，大大降低了内存消耗。

import java.util
import org.apache.spark.{SparkConf, SparkContext}
object testBroadcast {
  def main(args: Array[String]): Unit = {
    //声明一个SparkContext对象
    val sc=new SparkContext(new SparkConf().setMaster("local[*]").setAppName("BroadcastDemo"))
    val rdd = sc.parallelize(List("1,张三","0,李四","3,王五"))
    val map = new util.HashMap[String,String]()
    map.put("1","男人")
    map.put("0","女人")
    //声明一个广播变量
    val bd = sc.broadcast(map)
    val rdd1=rdd.map(e=>{
      val splits = e.split(",")
      val sid = splits(0)
      //获取广播变量中的值
      val name = bd.value.getOrDefault(sid,"未知")
      splits(1)+" is "+name+""
    }).cache()
    rdd1.foreach(println(_))
  }
}

注意： 不能将RDD使用一个广播变量广播出去，因为RDD是不存储数据的。可以将RDD的结果广播出去。通过广播变量的value获取广播变量的值。

参考文章：
https://blog.csdn.net/u013468917/article/details/70617085
https://blog.csdn.net/leen0304/article/details/78866353
http://www.ccblog.cn/103.htm
https://www.cnblogs.com/newdingwei/p/6802972.html