DStream实战之Spark Streaming开窗函数reduceByKeyAndWindow统计一段时间内出现的词汇并排序 34

àì夳堔傛蜴生んèń 2023-06-09 10:23 59阅读 0赞

# 需求 #

想统计**一段时间内**每个单词出现的次数, 并且对出现次数的高低进行**排序**, 选出这一段时间内的**热词**

# 1. 架构图 #

![在这里插入图片描述][2019102017492813.png]

# 2. 实现流程 #

1.  安装并启动生成者  
    首先在linux服务器上用YUM安装nc工具，nc命令是netcat命令的简称,都是用来设置路由器。我们可以利用它向某个端口发送数据。  
    `yum install -y nc`
2.  启动一个服务端并监听9999端口  
    `nc-lk 9999`  
    向指定的端口发送数据
3.  编写Spark Streaming程序  
    **先使用reduceByKeyAndWindow框住一段时间内的词汇**  
    **再使用sortBy对其排序**  
    **最后使用take(n)取出前几个**

package cn.acec.sparkStreamingtest
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * sparkStreming开窗函数应用----统计一定时间内的热门词汇 */
    object SparkStreamingTCPWindowHotWords { 
    
      def main(args: Array[String]): Unit = { 
        //配置sparkConf参数
        val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingTCPWindowHotWords").setMaster("local[2]")
        //构建sparkContext对象
        val sc: SparkContext = new SparkContext(sparkConf)
        sc.setLogLevel("WARN")
        //构建StreamingContext对象，每个批处理的时间间隔
        val scc: StreamingContext = new StreamingContext(sc,Seconds(5))
        //注册一个监听的IP地址和端口 用来收集数据
        val lines: ReceiverInputDStream[String] = scc.socketTextStream("192.168.200.160",9999)
        //切分每一行记录
        val words: DStream[String] = lines.flatMap(_.split(" "))
        //每个单词记为1
        val wordAndOne: DStream[(String, Int)] = words.map((_,1))
        //reduceByKeyAndWindow函数参数意义：
        // windowDuration:表示window框住的时间长度，如本例5秒切分一次RDD，框10秒，就会保留最近2次切分的RDD
        //slideDuration: 表示window滑动的时间长度，即每隔多久执行本计算
        val result: DStream[(String, Int)] = wordAndOne.reduceByKeyAndWindow((a:Int,b:Int)=>a+b,Seconds(10),Seconds(5))
        val data=result.transform(rdd=>{ 
          //降序处理后，取前3位
          val dataRDD: RDD[(String, Int)] = rdd.sortBy(t=>t._2,false)
          val sortResult: Array[(String, Int)] = dataRDD.take(3)
          println("--------------print top 3 begin--------------")
          sortResult.foreach(println)
          println("--------------print top 3 end--------------")
          dataRDD
        })
        data.print()
        scc.start()
        scc.awaitTermination()
      }
    }

# 3. 执行查看效果 #

1.  先执行`nc -lk 9999`  
    ![在这里插入图片描述][20191020175734850.png]
2.  然后在执行以上代码  
    ![在这里插入图片描述][201910201757486.png]
3.  不断的在（1）中输入不同的单词，观察IDEA控制台输出  
    ![在这里插入图片描述][20191020175810802.png]

# 执行结果 #

sparkStreaming每隔5s计算一次当前在窗口大小为10s内的数据，然后将**单词出现次数最多的前3位**进行输出打印。  
![在这里插入图片描述][20191020175836382.png]

[2019102017492813.png]: /images/20230531/383ce2489dde4b43be735f30fdc7adba.png
[20191020175734850.png]: /images/20230531/da2fd9efef8a4c2dbbff32f65b788c78.png
[201910201757486.png]: /images/20230531/047036f9c47641d68d560cbcb0cbdb41.png
[20191020175810802.png]: /images/20230531/37abad2b6da148dca67ad9d24c01ca7e.png
[20191020175836382.png]: /images/20230531/32bf06c415074ed4a3803dda616cbf39.png