Spark的广播变量和累加器使用方法代码示例

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)

scala> val accum = sc.accumulator(0, "My Accumulator")
accum: spark.Accumulator[Int] = 0
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s
scala> accum.value
res2: Int = 10

object VectorAccumulatorParam extends AccumulatorParam[Vector] {
 def zero(initialValue: Vector): Vector = {
  Vector.zeros(initialValue.size)
 }
 def addInPlace(v1: Vector, v2: Vector): Vector = {
  v1 += v2
 }
}
// Then, create an Accumulator of this type:
val vecAccum = sc.accumulator(new Vector(...))(VectorAccumulatorParam)

val accum = sc.accumulator(0)
data.map { x => accum += x; f(x) }
//在这里,accum的值仍然是0，因为没有动作操作引起map被实际的计算.

/**
 * 实例：利用广播进行黑名单过滤！
 * 检查新的数据 根据是否在广播变量-黑名单内，从而实现过滤数据。
 */
public class BroadcastAccumulator {
 /**
  * 创建一个List的广播变量
  *
  */
 private static volatile Broadcast> broadcastList = null;
 /**
  * 计数器！
  */
 private static volatile Accumulator accumulator = null;
 public static void main(String[] args) {
  SparkConf cOnf= new SparkConf().setMaster("local[2]").
    setAppName("WordCountOnlineBroadcast");
  JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

  /**
   * 注意：分发广播需要一个action操作触发。
   * 注意：广播的是Arrays的asList 而非对象的引用。广播Array数组的对象引用会出错。
   * 使用broadcast广播黑名单到每个Executor中！
   */
  broadcastList = jsc.sc().broadcast(Arrays.asList("Hadoop","Mahout","Hive"));
  /**
   * 累加器作为全局计数器！用于统计在线过滤了多少个黑名单！
   * 在这里实例化。
   */
  accumulator = jsc.sparkContext().accumulator(0,"OnlineBlackListCounter");

  JavaReceiverInputDStream lines = jsc.socketTextStream("Master", 9999);

  /**
   * 这里省去flatmap因为名单是一个个的！
   */
  JavaPairDStream pairs = lines.mapToPair(new PairFunction() {
   @Override
   public Tuple2 call(String word) {
    return new Tuple2(word, 1);
   }
  });
  JavaPairDStream wordsCount = pairs.reduceByKey(new Function2() {
   @Override
   public Integer call(Integer v1, Integer v2) {
    return v1 + v2;
   }
  });
  /**
   * Funtion里面 前几个参数是 入参。
   * 后面的出参。
   * 体现在call方法里面！
   *
   */
  wordsCount.foreach(new Function2, Time, Void>() {
   @Override
   public Void call(JavaPairRDD rdd, Time time) throws Exception {
    rdd.filter(new Function, Boolean>() {
     @Override
     public Boolean call(Tuple2 wordPair) throws Exception {
      if (broadcastList.value().contains(wordPair._1)) {
       /**
        * accumulator不仅仅用来计数。
        * 可以同时写进数据库或者缓存中。
        */
       accumulator.add(wordPair._2);
       return false;
      }else {
       return true;
      }
     };
     /**
      * 广播和计数器的执行，需要进行一个action操作！
      */
    }).collect();
    System.out.println("广播器里面的值"+broadcastList.value());
    System.out.println("计时器里面的值"+accumulator.value());
    return null;
   }
  });

  jsc.start();
  jsc.awaitTermination();
  jsc.close();
 }
 }

package com.Streaming
import java.util
import org.apache.spark.streaming.{Duration, StreamingContext}
import org.apache.spark.{Accumulable, Accumulator, SparkContext, SparkConf}
import org.apache.spark.broadcast.Broadcast
/**
 * Created by lxh on 2016/6/30.
 */
object BroadcastAccumulatorStreaming {
 /**
 * 声明一个广播和累加器！
 */
 private var broadcastList:Broadcast[List[String]] = _
 private var accumulator:Accumulator[Int] = _
 def main(args: Array[String]) {
 val sparkCOnf= new SparkConf().setMaster("local[4]").setAppName("broadcasttest")
 val sc = new SparkContext(sparkConf)
 /**
  * duration是ms
  */
 val ssc = new StreamingContext(sc,Duration(2000))
 // broadcastList = ssc.sparkContext.broadcast(util.Arrays.asList("Hadoop","Spark"))
 broadcastList = ssc.sparkContext.broadcast(List("Hadoop","Spark"))
 accumulator= ssc.sparkContext.accumulator(0,"broadcasttest")
 /**
  * 获取数据！
  */
 val lines = ssc.socketTextStream("localhost",9999)
 /**
  * 1.flatmap把行分割成词。
  * 2.map把词变成tuple(word,1)
  * 3.reducebykey累加value
  * (4.sortBykey排名)
  * 4.进行过滤。 value是否在累加器中。
  * 5.打印显示。
  */
 val words = lines.flatMap(line => line.split(" "))
 val wordpair = words.map(word => (word,1))
 wordpair.filter(record => {broadcastList.value.contains(record._1)})
 val pair = wordpair.reduceByKey(_+_)
 /**
  * 这个pair 是PairDStream
  * 查看这个id是否在黑名单中，如果是的话，累加器就+1
  */
/* pair.foreachRDD(rdd => {
  rdd.filter(record => {
  if (broadcastList.value.contains(record._1)) {
   accumulator.add(1)
   return true
  } else {
   return false
  }
  })
 })*/
 val filtedpair = pair.filter(record => {
  if (broadcastList.value.contains(record._1)) {
   accumulator.add(record._2)
   true
  } else {
   false
  }
  }).print
 println("累加器的值"+accumulator.value)
 // pair.filter(record => {broadcastList.value.contains(record._1)})
 /* val keypair = pair.map(pair => (pair._2,pair._1))*/
 /**
  * 如果DStream自己没有某个算子操作。就通过转化transform！
  */
 /* keypair.transform(rdd => {
  rdd.sortByKey(false)//TODO
 })*/
 pair.print()
 ssc.start()
 ssc.awaitTermination()
 }
}