广播变量与累加器

/**
* share variables:BroadcastApp
* 广播大变量实现以map join取代reduce join
*/
object BroadcastApp {

def main(args: Array[String]): Unit = {
// commomJoin()

broadcastJoin()
}
def commomJoin(): Unit = {

val conf = new SparkConf().setMaster("local[2]").setAppName("BroadcastApp")
val sc = new SparkContext(conf)

val wideInfo = sc.parallelize(List(("01","阿呆"),("02","sk"),("03","shjqi")))
val baseInfo = sc.parallelize(List(("01",("北京","22")),("05",("上海","22"))))

wideInfo.join(baseInfo).map(x =>(x._1,x._2._1,x._2._2._1,x._2._2._2))
.collect().foreach(println)

Thread.sleep(2000000)
sc.stop()
}

def broadcastJoin(): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("BroadcastApp")
val sc = new SparkContext(conf)

val wideInfo = sc.parallelize(List(("01","阿呆"),("02","sk"),("03","shjqi")))
val baseInfo = sc.parallelize(List(("01",("北京","22")),("05",("上海","22"))))
.collectAsMap()
val broadcastBaseInfo = sc.broadcast(baseInfo)

wideInfo.mapPartitions(x =>{
val value = broadcastBaseInfo.value
for((k,v)<-x if(value.contains(k))) 
yield (k,v,value.get(k).getOrElse().asInstanceOf[(String,String)]._1,value.get(k).getOrElse().asInstanceOf[(String,String)]._2)
}).foreach(println)

Thread.sleep(2000000)
sc.stop()
}
}

object AccumulatorsApp {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("AccumulatorsApp")
    val sc = new SparkContext(conf)

    val wskAccum = sc.longAccumulator("wsktets")//定义一个累加器
    val nums = sc.parallelize(List(1,2,3,4),2)
    println("driver端:计数器值："+wskAccum.value)

    nums.map(x=>{
      println("计数前值："+wskAccum.value)//还原一个累加器
      wskAccum.add(x)
      println("计数后值："+wskAccum.value)
      x
    }).collect()
    println("driver端:最终计数器值："+wskAccum.value)

    Thread.sleep(2000000)
    sc.stop()
  }
}

package Mapreduce



import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

class CustomerAcc extends AccumulatorV2[String,mutable.HashMap[String,Int]] {

  private val _hash = new mutable.HashMap[String,Int]()

//  检测是否为空
  override def isZero: Boolean = {
    _hash.isEmpty
  }
//  拷贝一个新的累加器
  override def copy(): AccumulatorV2[String,mutable.HashMap[String,Int]] = {
    val copyHash = new CustomerAcc()
//    创造一个copy的累加器，然后用synchronized方法设置同步操作
    _hash.synchronized{
      copyHash._hash++=_hash
    }
    copyHash
  }
//  重置累加器
  override def reset(): Unit = {
    _hash.clear()
  }
//  每一个分区中用于添加数据的方法 小Sum
  override def add(v: String) ={

    _hash.get(v) match {
      case None=>_hash+=((v,1))
      case Some(x)=>_hash+=((v,x+1))
    }

  }
//  合并每一个分区的输出 总Sum
  override def merge(other: AccumulatorV2[String,mutable.HashMap[String,Int]]) = {
    other match{
      case o:AccumulatorV2[String,mutable.HashMap[String,Int]]=>{
        for ((k,v)<- o.value){
        _hash.get(k) match {
          case None=>_hash+=((k,v))
          case Some(x)=>_hash+=((k,v+x))
        }
    }
    }
    }
  }
//  输出值
  override def value(): mutable.HashMap[String,Int] = {
    _hash
  }
}

object CustomerAcc {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("Partition1").setMaster("local[*]")

    val sc = new SparkContext(sparkConf)

    val hash = new CustomerAcc()
    sc.register(hash)

    val rdd = sc.makeRDD(Array("a","b","c","a","b","c","d"))

    rdd.foreach(hash.add(_))

    for((k,v)<-hash.value){
      println("["+k+":"+v+"]")
    }

    sc.stop()
  }
}