教材P164操作题。编写Spark Steaming程序，使用leftOuterJoin操作及filter方法过滤掉黑名单的数据

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.StorageLevels;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
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;
public final class 过滤黑名单{
        public static void main(String[] args) throws Exception 
  {
    if (args.length< 2) {
      System.err.println("需要传入参数：主机名端口号");
      System.exit(1);
              }
  // 设置拉取数据的频率,即批处理的时间间隔为1秒
  //控制台上显示的是每隔1000毫秒
    SparkConf  sparkConf = new SparkConf().setAppName
        ("JavaNetworkWordCount").setMaster("local[2]");
    JavaStreamingContext  ssc = new 
      JavaStreamingContext(sparkConf, Durations.seconds(10));
     JavaReceiverInputDStream<String> filedata = ssc.socketTextStream(
  args[0], Integer.parseInt(args[1]), StorageLevels.MEMORY_AND_DISK_SER);
     JavaRDD<String>  blackname= ssc.sparkContext().textFile("file:///homeq/eclipse-workspace/t33");
//     JavaDStream<String> blackname=ssc.textFileStream("file:///homeq/eclipse-workspace/t33");
     JavaDStream<String> words = filedata.map(f->f);
     JavaPairRDD<String,String> prd=blackname.mapToPair(f->new Tuple2<String,String>(f.split(" ")[0],f.split(" ")[1]));
     //此处的reduceByKey方法，每次只输出当次的操作记录，
     //不保留上次的记录信息。对应的就是只针对本次的key，values。
     //要保留前次的操作记录。相对应的方法就是updateStateByKey。
     JavaPairDStream<String, Tuple2<Tuple2<String, String>, Optional<String>>> pard=words.mapToPair
        (f->new Tuple2<String,Tuple2<String,String>>(f.split(" ")[1],
            new Tuple2<>(f.split(" ")[0],f.split(" ")[1]))).transformToPair(fs->{
      JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<String>>> prd2= fs.leftOuterJoin(prd);
      JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<String>>> prd3=
             prd2.filter(f->!f._2._2().toString().contains("true"));
      return prd3;
            }
                );
     JavaPairDStream<String,String> prd4=pard.mapToPair(f->f._2._1);
prd4.print();
     ssc.start();     // 启动Spark Streaming，开始计算
  ssc.awaitTermination();    // 等待计算结束
    }
  }