spark streaming測試之四設(shè)置窗口大小接收數(shù)據(jù)

測試思路：

    首先，使用網(wǎng)絡(luò)數(shù)據(jù)發(fā)送程序發(fā)送數(shù)據(jù)；

    然后，運(yùn)行spark程序；

    觀察效果。

說明：

    1. 這里也需要設(shè)置檢查點(diǎn)目錄

    2. 這里有四個(gè)參數(shù)：

       前兩個(gè)分別是監(jiān)聽的端口和每隔多少毫秒接收一次數(shù)據(jù)；

       第三個(gè)參數(shù)是接收前多少毫秒的數(shù)據(jù)；（詳細(xì)請參見window具體含義）

       第四個(gè)參數(shù)是每隔多少毫秒接收一次數(shù)據(jù)。

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()
  }
}