第1关：套接字流实现黑名单过滤
 

简介

套接字流是通过监听Socket端口接收的数据，相当于Socket之间的通信，任何用户在用Socket（套接字）通信之前，首先要先申请一个Socket号，Socket号相当于该用户的电话号码。同时要知道对方的Socket，相当于对方也有一个电话号码。然后向对方拨号呼叫，相当于发出连接请求。对方假如在场并空闲，相当于通信的另一主机开机且可以接受连接请求，拿起电话话筒，双方就可以正式通话，相当于连接成功。Spark Streaming通过监听套接字端口获取流数据信息并处理。
黑名单过滤是获取套接字流发送的名单信息，通过指定的黑名单信息对其进行过滤，输出结果为没有黑名单的名单信息。

任务描述

本关任务：本关是利用套接字流监听方法，监听名单信息并过滤黑名单信息。首先需要模拟名单的生成，首先需要建一个文档，每行为一个姓名。然后编写代码，当有指定套接字连接产生时，从文件中依次选取所有名单，发送给套接字端口。另外在编写代码，通过连接套接字端口，监听端口的数据，获取发送的名单，并过滤黑名单。需用提供的文档文件内的名单来完成此实验。名单文档内容如下：
Jim
Mary
Tom
Jack
Abby
Bee
Belle
Babs
Carla
Dale
Dan
Gary
Ken
Jane
Paige

相关知识

1.名单生成器的创建

代码依次获取指定文件中的一个名单。内部是首先获取指定名单文件，然后指定监听端口，然后等待连接建立，连接建立后依次获取一个名单总个数内的整数，并索引到名单，将其发送至套接字端口。

import java.io.{PrintWriter}
import java.net.ServerSocket
import scala.io.Source
object NameProducer {def main(args: Array[String]): Unit = {//获取指定文件val filename = "/root/data/Namelist"//指定文件按行切分为listval lines = Source.fromFile(filename).getLines.toList//获取list长度val filerow = lines.length//指定监听窗口，当外部程序请求时建立连接val listener = new ServerSocket(5566)//等待socket连接成功println("bengin11111111111!!!")val socket = listener.accept()new Thread() {override def run = {println("Got client connected from: " + socket.getInetAddress)//创建写入socket对象val out = new PrintWriter(socket.getOutputStream(), true)println("bengin222222222222222222!!!")for (i <- 0 to filerow-1) {//逐行取一个名字val content = lines(i)println(content)//写入socketout.write(content + '\n')//清空缓冲区数据out.flush()}socket.close()}}.start()}
}

2.监听套接字端口过滤黑名单

首先要设置需要过滤的黑名单：

val BlackList = Array("name1", "name2")

接着把名单转换成RDD，然后连接套接字端口，连接端口代码如下：

ssc.socketTextStream("localhost", 5566)

编程要求

根据提示，补充监听套接字并过滤黑名单代码文TransformBlackList。

测试说明

平台会对你编写的代码进行测试：
测试输入：自动获取指定的Namelist文件。
预期输出：

Jim
Tom
Jack
Bee
Belle
Babs
Dale
Dan
Gary
Jane

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Secondsobject TransformBlackList {def main(args: Array[String]): Unit = {/********** Begin **********/// 初始化val sparkConf = new SparkConf().setAppName("TransformBlackList").setMaster("local[2]")// 创建StreamingContext，设置每五秒刷新一次val ssc = new StreamingContext(sparkConf, Seconds(5))ssc.sparkContext.setLogLevel("ERROR")// 设置需要过滤的黑名单(Abby、 Paige、 Carla、 Mary、 Ken)val BlackList = Array("Abby", "Paige", "Carla", "Mary", "Ken")// 把黑名单数组转换成rddval BlackListRdd = ssc.sparkContext.parallelize(BlackList)// 设置主机名localhost，端口号5566val NameList = ssc.socketTextStream("localhost", 5566)// 过滤黑名单算法val ValidName = NameList.transform(rdd => {rdd.filter(name => !BlackList.contains(name))})ValidName.print()ssc.start()// 等待足够长的时间以触发至少一次计算Thread.sleep(5000)ssc.stop(false, false)/********** End **********/}
}