DStream实战之Spark Streaming开窗函数reduceByKeyAndWindow,实现某一段时间单词计数 33

「爱情、让人受尽委屈。」 2023-06-07 03:24 3阅读 0赞

# 需求 #

我想**统计一段时间内**每个单词出现的次数.

# 1. 架构图 #

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

# 2. 实现流程 #

1.  安装并启动生成者  
    首先在linux服务器上用YUM安装nc工具，nc命令是netcat命令的简称,都是用来设置路由器。我们可以利用它向某个端口发送数据。  
    **yum install -y nc**
2.  启动一个服务端并监听9999端口  
    **nc-lk 9999**  
    向指定的端口发送数据
3.  编写Spark Streaming程序  
    **在逻辑代码写完后,使用reduceByKeyAndWindow()算子, 将一段时间框起来.**

package cn.acec.sparkStreamingtest
    
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * sparkStreming开窗函数---统计一定时间内单词出现的次数 */
    object SparkStreamingTCPWindow { 
    
      def main(args: Array[String]): Unit = { 
        //配置sparkConf参数
        val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingTCPWindow").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))
        result.print()
        scc.start()
        scc.awaitTermination()
      }
    }

# 3. 执行如上代码 #

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

# 4. 查看结果 #

**现象：sparkStreaming每隔5s计算一次当前在窗口大小为10s内的数据，然后将结果数据输出。**

[20191012162224302.png]: /images/20230601/334f61fd8f954a9cad5cc1d0bfb21eb7.png
[20191012162622667.png]: /images/20230601/b451cd5ed1da419b9e383040a91e3f35.png
[20191012162632714.png]: /images/20230601/c3b9ba1ce68b4564a792c983c9bbd9bc.png
[20191012162653493.png]: https://img-blog.csdnimg.cn/20191012162653493.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191012162704673.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ==,size_16,color_FFFFFF,t_70
[20191012162738132.png]: /images/20230601/a363a45f65394bbaa05cdba62b7dfa17.png