Spark Streaming处理Socket流简单实例

心已赠人 2023-06-09 14:27 16阅读 0赞

#### 在本文中我将在IDEA工具中开发一个`SparkStream`程序用于监听本机9999端口所接收的数据 ####

首先，我们将`Spark Streaming`类的名称以及从`StreamingContext`进行的一些隐式转换导入到我们的环境中，以便向我们需要的其他类（如DStream）添加有用的方法。StreamingContext是所有流功能的主要入口点。我们创建具有8个执行线程(最大线程数)和10秒批处理间隔的本地StreamingContext。

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    var sparkConf =new SparkConf().setMaster("local[*]").setAppName("SparkStream").set("spark.testing.memory", "2147480000")
    
    //分析环境对象以及采集周期
    val streamContext = new StreamingContext(sparkConf,Seconds(10))

使用此上下文，我们可以创建一个DStream，该DStream表示来自TCP源的流数据，指定为主机名（例如bigdata，这是我的虚拟机实例）和端口（例如9999）

val socketStreamLine: ReceiverInputDStream[String] = streamContext.socketTextStream("bigdata",9999)

接下来对采集到的数据进行处理，使用flatMap，map，reduceByKey进行wordcount统计

//socketStreamLine表示将从数据服务器接收的数据流。此DStream中的每个记录都是一行文本。接下来，我们要用空格将行分割成单词。
    	val dStream: DStream[String] = socketStreamLine.flatMap(line => line.split(" "))
    
        //将数据进行结构转变
        val map: DStream[(String, Int)] = dStream.map((_,1))
        //聚合处理
        val key: DStream[(String, Int)] = map.reduceByKey(_+_)

Spark Streaming仅设置启动时将执行的计算，并且尚未开始任何实际处理。在完成所有转换后，要开始处理，我们最终调用

//结果打印
        key.print()
     	streamContext.start()
        //等待采集器执行
        streamContext.awaitTermination()

启动项目后，使用以下命令将Netcat作为数据服务器运行

$ nc -lk 9999

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]

完整代码托管于[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]，欢迎fork，star  
![在这里插入图片描述][20191021171522892.jpg]

[20191021171036342.png]: /images/20230531/38f1ef58f91348dfa357b74238dddd8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]: /images/20230531/f7b9abcb76294b15adcda679406c5c42.png
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark
[20191021171522892.jpg]: /images/20230531/a044b7a8418348b7ab7d2587f4763da4.png