溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
測試思路：
首先，使用網絡數據發送程序發送數據；
然后，運行spark程序；
觀察效果。
說明：
1. 這里也需要設置檢查點目錄
2. 這里有四個參數：
前兩個分別是監聽的端口和每隔多少毫秒接收一次數據；
第三個參數是接收前多少毫秒的數據；（詳細請參見window具體含義）
第四個參數是每隔多少毫秒接收一次數據。
sparkStreaming
import org.apache.log4j.{LoggerLevel}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{SecondsStreamingContext}
import org.apache.spark.{SparkContextSparkConf}
import org.apache.spark.streaming.StreamingContext._
WindowWordCount {
def main(args: Array[]){
Logger.().setLevel(Level.)
Logger.().setLevel(Level.)
conf = SparkConf().setAppName().setMaster()
sc = SparkContext(conf)
ssc = StreamingContext(sc())
ssc.checkpoint()
val lines = ssc.socketTextStream(args(0),args(1).toInt,
StorageLevel.MEMORY_ONLY_SER)
words = lines.flatMap(_.split())
//windows operator
val wordCounts = words.map(x=>(x,1)).reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),
Seconds(args(2).toInt),Seconds(args(3).toInt))
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
