Spark学习笔记（三）：SparkStreaming实现对文件夹和socket的监听

Bertha 。 2021-10-29 11:58 644阅读 0赞

SparkStreaming是Spark的一个流式计算框架，它支持对许多数据源进行实时监听，例如Kafka, Flume, Kinesis, TCP sockets，甚至文件夹，并实现实时计算的能力。

# 对文件夹的监听 #

def fileStreaming(): Unit ={ 
            /** * 监听文件夹的新增文件内容 */
            // 至少要启动2个线程以上，1个用于监听，1个用于处理数据
            val conf = new SparkConf().setMaster("local[8]").setAppName("SparkSql")
            val sc = new SparkContext(conf)
            sc.setLogLevel("ERROR")
            // 这里是每隔3秒进行一次数据处理
            val ssc = new StreamingContext(sc, Seconds(3))
            // 只会读取在监听期间传入监听文件夹的文件
            // 并且该文件还必须在开始监听之后进行修改过
            val lines = ssc.textFileStream("resources/data/SparkStreaming.SparkStreaming/wordCount.txt")
            val words = lines.flatMap(_.split(","))
            val wordCount = words.map(x => (x, 1)).reduceByKey(_ + _)
            wordCount.print()
            ssc.start()
            ssc.awaitTermination()
        }

**这里有些地方是需要注意：  
1.SparkContext至少要启动2个线程以上，1个用于监听，1个用于处理数据  
2.对文件监听我个人认为不是很人性化：只会读取在监听期间传入监听文件夹的文件；并且该文件还必须在开始监听之后进行修改过**

# 对socket端口的监听 #

**首先，我们需要先创建一个scoket服务端口，定时往里面写入内容，我们的SparkStreaming才能进行监听并实时接受数据**

import java.io.PrintWriter
    import java.net.ServerSocket
    
    import scala.io.Source
    
    /** * 建立一个socket服务，间隔一定时间从文件中随机读取一行内容 */
    object SocketServer { 
        private val rd = new java.util.Random()
    
        def rdInt(max: Int): Int ={ 
            rd.nextInt(max)
        }
    
        def main(args: Array[String]): Unit = { 
            val fileName = args(0) // 读取的文件路径
            val port = args(1).toInt // socket端口号
            val interval = args(2).toLong // 读取文件内容的时间间隔：毫秒
    
            val reader = Source.fromFile(fileName)
            val lines = reader.getLines().toList
            reader.close()
            val length = lines.length
            val listener = new ServerSocket(port)
            while (true){  // 一直监听该socket端口
                val socket = listener.accept()
                new Thread(){ 
                    override def run = { 
                        val out = new PrintWriter(socket.getOutputStream, true)
                        while (true){ 
                            Thread.sleep(interval)
                            val content = lines(rdInt(length))
                            println(content)
                            out.write(content + "\n")
                            out.flush()
                        }
                        socket.close()
                    }
                }.start()
            }
        }
    }

接下来，就是我们SparkStreaming的监听代码

def socketStreaming(): Unit ={ 
            /** * 监听socket端口的写入内容 */
            val conf = new SparkConf().setMaster("local[4]").setAppName("SparkSql")
            val sc = new SparkContext(conf)
            sc.setLogLevel("ERROR")
            // 这里是每隔3秒进行一次数据处理
            val ssc = new StreamingContext(sc, Seconds(3))
            val lines = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK_SER)
            val words = lines.flatMap(_.split(","))
            val wordCount = words.map(x => (x, 1)).reduceByKey(_ + _)
            wordCount.print()
            ssc.start()
            ssc.awaitTermination()
        }

完整的代码我已经上传至[GitHub][]  
欢迎关注同名公众号：“我就算饿死也不做程序员”。  
交个朋友，一起交流，一起学习，一起进步。![在这里插入图片描述][20200430114013512.jpg]

[GitHub]: https://github.com/SGyuanshi/Spark/tree/master/SparkScala/src/main/scala/SparkStreaming
[20200430114013512.jpg]: /images/20211029/77a7368801474999bbc6dd151978b1b9.png