溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
測試思路：
首先，使用上篇文章的程序一發送網絡數據；
其次，運行spark程序，觀察效果。
說明：
1. 這里使用到了更新函數；
2. 使用檢查點來保證狀態。
sparkStreaming
import org.apache.log4j.{LoggerLevel}
import org.apache.spark.streaming.{SecondsStreamingContext}
import org.apache.spark.{SparkContextSparkConf}
import org.apache.spark.streaming.StreamingContext._
object StatefulWordCount {
def main(args:Array[]){
Logger.().setLevel(Level.)
Logger.().setLevel(Level.)
updateFunc = (values: []state:Option[]) => {
currentCount = values.foldLeft()(_+_)
previousCount = state.getOrElse()
(currentCount + previousCount)
}
conf = SparkConf().setAppName().setMaster()
sc = SparkContext(conf)
ssc = StreamingContext(sc())
ssc.checkpoint()
lines = ssc.socketTextStream(args()args().toInt)
words = lines.flatMap(_.split())
wordCounts = words.map(x=>(x))
stateDstream = wordCounts.updateStateByKey[](updateFunc)
stateDstream.print()
ssc.start()
ssc.awaitTermination()
}
}免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
