怎么聯合使用Spark Streaming、Broadcast、Accumulaor

194次閱讀

共計 3169 個字符，預計需要花費 8 分鐘才能閱讀完成。

本篇內容介紹了“怎么聯合使用 Spark Streaming、Broadcast、Accumulaor”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓丸趣 TV 小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

廣播可以自定義，通過 Broadcast、Accumulator 聯合可以完成復雜的業務邏輯。

以下代碼實現在本機 9999 端口監聽，并向連接上的客戶端發送單詞，其中包含黑名單的單詞 Hadoop，Mahout 和 Hive。

package org.scala.opt

import java.io.{PrintWriter, IOException}
import java.net.{Socket, SocketException, ServerSocket}

case class ServerThread(socket : Socket) extends Thread(ServerThread) {
override def run(): Unit = {
val ptWriter = new PrintWriter(socket.getOutputStream)
try {
var count = 0
var totalCount = 0
var isThreadRunning : Boolean = true
val batchCount = 1
val words = 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 {
case e : SocketException =
println(SocketException : , e)
case e : IOException =
e.printStackTrace();
} finally {
if (ptWriter != null) ptWriter.close()
println(Client + socket.getInetAddress + disconnected)
if (socket != null) socket.close()
}
println(Thread.currentThread().getName + Exit )
}
}
object SocketServer {
def main(args : Array[String]) : Unit = {
try {
val listener = new ServerSocket(9999)
println(Server is started, waiting for client connect…)
while (true) {
val socket = listener.accept()
println(Client : + socket.getLocalAddress + connected)
new ServerThread(socket).start()
}
listener.close()
}
catch {
case e: IOException =
System.err.println(Could not listen on port: 9999.)
System.exit(-1)
}
}
}

以下代碼實現接收本機 9999 端口發送的單詞，統計黑名單出現的次數的功能。

package com.dt.spark.streaming_scala

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, Accumulator}
import org.apache.spark.broadcast.Broadcast

/**
* 第 103 課：動手實戰聯合使用 Spark Streaming、Broadcast、Accumulator 實現在線黑名單過濾和計數
* 本期內容：
1，Spark Streaming 與 Broadcast、Accumulator 聯合
2，在線黑名單過濾和計算實戰
*/
object _103SparkStreamingBroadcastAccumulator {

@volatile private var broadcastList : Broadcast[List[String]] = null
@volatile private var accumulator : Accumulator[Int] = null

def main(args : Array[String]) : Unit = {
val conf = new SparkConf().setMaster( local[5] ).setAppName(_103SparkStreamingBroadcastAccumulator)
val ssc = new StreamingContext(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”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注丸趣 TV 網站，丸趣 TV 小編將為大家輸出更多高質量的實用文章！

正文完