本博文主要包括：
1、Spark Streaming與Broadcast、Accumulator聯合
2、線上黑名單過濾和計數實戰

一、Spark Streaming與Broadcast、Accumulator聯合：

在企業實戰中，廣播本身廣播到叢集的時候，聯合上計數器的話就有很大殺傷力，這時候你可以自定義，例如自定義廣播中的內容，可以實現非常複雜的內容。

之所以廣播和計數器特別重要，一方面鑑於廣播和計數器本身的特性，另一方面廣播和計數器可以說實現非常複雜的操作。線上黑名單過濾實戰中，將黑名單放在廣播中，有Accumulator你可以計數黑名單。

二、線上黑名單過濾和計數實戰：

1、程式碼如下：

import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache
 
.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.Time;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext
 
;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * Created by zpf on 2016/8/31.
 */
public class SparkStreamingBroadcastAccumulator {

    private static volatile Broadcast<List<String>> broadcastList = null;
    private static volatile Accumulator<Integer> accumulator = null;

    public void main(String[] args) throws InterruptedException {

        SparkConf conf = new SparkConf().setMaster("local[2]").
                setAppName("SparkStreamingBroadcastAccumulator");


        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(15));

        //例項化我們的broadcast,使用Broadcast廣播黑名單到每個Executor中
        broadcastList = jsc.sparkContext().broadcast(Arrays.asList("Hadoop","Mahout","Hive"));
      /*全域性計數器，用於統計線上過濾多少黑名單
      * */
        accumulator = jsc.sparkContext().accumulator(0,"OnlineBlacklistCount");

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

        JavaPairDStream<String,Integer> pairs = lines.mapToPair(new PairFunction<String,String,Integer>() {

            public Tuple2<String,Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word,1);
            }
        });

        JavaPairDStream<String,Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });

        /*過濾黑明單我們一般把內容寫在foreach中*/

        wordsCount.foreachRDD(new Function2<JavaPairRDD<String,Integer>, Time, Void>(){

            public Void call(JavaPairRDD<String, Integer> rdd, Time time) throws Exception {
                rdd.filter(new Function<Tuple2<String, Integer>, Boolean>() {
                    public Boolean call(Tuple2<String, Integer> wordPair) throws Exception {
                        if(broadcastList.value().contains(wordPair._1)){
                            accumulator.add(wordPair._2);
                            return false;

                        }else{
                            return true;
                        }
                    }



                }).collect();

//                System.out.println(broadcastList.value().toString() + ":" + accumulator.value());
                System.out.println("BlackList append : " + ":" + accumulator.value() + "times");
                return null;


            }
        });



        jsc.start();

        jsc.awaitTermination();

        jsc.close();
    }

}

2、在命令端輸入資料

nc -lk 9999

3、觀察結果