怎么联合使用SparkStreaming、Broadcast、Accumulaor
本篇内容介绍了“怎么联合使用Spark Streaming、Broadcast、Accumulaor”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
专注于为中小企业提供网站建设、成都网站设计服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业万年免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了近1000家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。
广播可以自定义，通过Broadcast、Accumulator联合可以完成复杂的业务逻辑。
以下代码实现在本机9999端口监听，并向连接上的客户端发送单词，其中包含黑名单的单词Hadoop，Mahout和Hive。
packageorg.scala.opt
importjava.io.{PrintWriter, IOException}
importjava.net.{Socket, SocketException, ServerSocket} case classServerThread(socket : Socket) extendsThread("ServerThread") {
override defrun(): Unit = {
valptWriter = newPrintWriter(socket.getOutputStream)
try{
varcount = 0
vartotalCount = 0
varisThreadRunning : Boolean = true
valbatchCount = 1
valwords = List("Java Scala C C++ C# Python JavaScript",
"Hadoop Spark Ngix MFC Net Mahout Hive")
while(isThreadRunning) {
words.foreach(ptWriter.println)
count += 1
if(count >= batchCount) {
totalCount += count
count = 0
println("batch "+ batchCount + " totalCount => "+ totalCount)
Thread.sleep(1000)
}
//out.println此类中的方法不会抛出 I/O 异常，尽管其某些构造方法可能抛出异常。客户端可能会查询调用 checkError() 是否出现错误。
if(ptWriter.checkError()) {
isThreadRunning = false
println("ptWriter error then close socket")
}
}
}
catch{
casee : SocketException =>
println("SocketException : ", e)
casee : IOException =>
e.printStackTrace();
} finally{
if(ptWriter != null) ptWriter.close()
println("Client "+ socket.getInetAddress + " disconnected")
if(socket != null) socket.close()
}
println(Thread.currentThread().getName + " Exit")
}
}
objectSocketServer {
defmain(args : Array[String]) : Unit = {
try{
vallistener = newServerSocket(9999)
println("Server is started, waiting for client connect...")
while(true) {
valsocket = listener.accept()
println("Client : "+ socket.getLocalAddress + " connected")
newServerThread(socket).start()
}
listener.close()
}
catch{
casee: IOException =>
System.err.println("Could not listen on port: 9999.")
System.exit(-1)
}
}
} |
以下代码实现接收本机9999端口发送的单词，统计黑名单出现的次数的功能。
packagecom.dt.spark.streaming_scala
importorg.apache.spark.streaming.{Seconds, StreamingContext}
importorg.apache.spark.{SparkConf, Accumulator}
importorg.apache.spark.broadcast.Broadcast
/**
* 第103课： 动手实战联合使用Spark Streaming、Broadcast、Accumulator实现在线黑名单过滤和计数
* 本期内容：
1，Spark Streaming与Broadcast、Accumulator联合
2，在线黑名单过滤和计算实战
*/
object_103SparkStreamingBroadcastAccumulator {
@volatile private varbroadcastList : Broadcast[List[String]] = null
@volatile private varaccumulator : Accumulator[Int] = null
defmain(args : Array[String]) : Unit = {
valconf = newSparkConf().setMaster("local[5]").setAppName("_103SparkStreamingBroadcastAccumulator")
valssc = newStreamingContext(conf, Seconds(5))
ssc.sparkContext.setLogLevel("WARN")
/**
* 使用Broadcast广播黑名单到每个Executor中
*/
broadcastList = ssc.sparkContext.broadcast(Array("Hadoop", "Mahout", "Hive").toList)
/**
* 全局计数器，用于通知在线过滤了多少各黑名单
*/
accumulator = ssc.sparkContext.accumulator(0, "OnlineBlackListCounter")
ssc.socketTextStream("localhost", 9999).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).foreachRDD {rdd =>{
if(!rdd.isEmpty()) {
rdd.filter(wordPair => {
if(broadcastList.value.contains(wordPair._1)) { println("BlackList word %s appeared".formatted(wordPair._1))
accumulator.add(wordPair._2)
false
} else{
true
}
}).collect()
println("BlackList appeared : %d times".format(accumulator.value))
}
}}
ssc.start()
ssc.awaitTermination()
ssc.stop()
}
} |
Server发送端日志如下，不断打印输出的次数。
Spark Streaming端打印黑名单的单词及出现的次数。
“怎么联合使用Spark Streaming、Broadcast、Accumulaor”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！
当前名称：怎么联合使用SparkStreaming、Broadcast、Accumulaor
文章网址：
http://myzitong.com/article/gcshjo.html
