DStream实战之Spark Streaming接收socket数据实现WordCount 31

蔚落 2023-06-07 00:54 266阅读 0赞

# 需求 #

现在想要通过socket发送数据, 然后Spark Streaming接收数据并且统计socket发送的每个单词的个数.

# 1. 架构图 #

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

# 2. 实现流程 #

1.  安装并启动生产者  
    首先在linux服务器上用**yum安装netcat工具**,netcat工具用了发送socket套接字,**netcat的命令是nc**, 它使用来设置路由器的,我们可以利用它向某个端口发送数据.  
    **linux安装netcat命令如下:**

yum intstall -y nc

1.  通过netcat工具向指定的端口发送数据

nc -lk 9999

1.  IDEA编写Spark Streaming代码

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流式处理接受socket数据，实现单词统计 */
    object SparkStreamingNC{ 
    	def main(args: Array[String]): Unit = { 
    		//配置sparkConf参数
        	val sparkConf: SparkConf = new 	 SparkConf().setAppName("SparkStreamingTCP").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))
         //分组聚合
         val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
         //打印数据
         result.print()
         scc.start()
         scc.awaitTermination()
    	}
    }

**注意:**  
由于使用的是本地模式**local\[2\]**, 所以可以直接在本地运行程序  
要指定**并行度**, 如在本地运行设置setMaster(“local\[2\]”)，相当于启动两个线程，一个给receiver，一个给computer。**如果是在集群中运行，必须要求集群中可用core数大于1**。

# 3. 执行查看结果 #

1.  先执行`nc -lk 9999`  
    ![在这里插入图片描述][20191012142809879.png]
2.  然后再在IDEA中启动代码  
    ![在这里插入图片描述][20191012143633598.png]
3.  不断的在1.中输入不同的单词，观察IDEA控制台输出  
    ![在这里插入图片描述][20191012143656381.png]  
    ![在这里插入图片描述][20191012143700201.png]

# 4. 结果 #

**现象：sparkStreaming每隔5s计算一次当前5s内的数据，然后将每个批次的数据输出。**

[20191012114259927.png]: /images/20230601/18980e2ea54b49bead69160f51ad1c3b.png
[20191012142809879.png]: /images/20230601/b0008665b2fe41399043e40378197b50.png
[20191012143633598.png]: /images/20230601/a41a545da4f34640abc382e7d4a6dea7.png
[20191012143656381.png]: /images/20230601/d57d65fd7bef45aea1c31904fa76aac0.png
[20191012143700201.png]: /images/20230601/aeb2086e2fba43619130d763fba84118.png