sparkstreaming输入测试（网络端口数据）

川长思鸟来 2022-07-14 13:44 169阅读 0赞

scala程序实现  
接收一个端口发来的信息：端口信息使用模拟程序来实现一个端口发来的信息：代码如下：  
程序生成A-G的随机字母

package sparkStreaming_
    
    import java.io.PrintWriter
    import java.net.ServerSocket
    
    /** * Created by xiaopengpeng on 2016/12/11. */
    class LoggerSimulation { 
    
    }
    object LoggerSimulation{ 
      /** * 生成一个字母 * @param 字母的下标 * @return 生成的字母 */
      def gennerateContent(index:Int):String = {
        import scala.collection.mutable.ListBuffer
        val charList = ListBuffer[Char]();
        for (i<- 65 to 90){
          charList += i.toChar
        }
        val charArray = charList.toArray
        charArray(index).toString();
      }
    
      /** * 生成随机下标 * @return 返回一个下标 */
      def index = {
        import java.util.Random
        val rdm = new Random()
        rdm.nextInt(7)
      }
    
      /** * 启动一个main方法来创建一个serversockt发送消息 * @param args 端口，发送的时间间隔 */
      def main(args: Array[String]): Unit = {
        if (args.length !=2){
          System.err.println("Usage:<port><millisecond>")
          System.exit(1);
        }
    
        val listener = new ServerSocket(args(0).toInt)
        println("已经做好连接的准备-------")
        while(true){
          val socket = listener.accept()
          new Thread(){
            override def run(): Unit = {
              println("Got client connected from:"+socket.getInetAddress)
              val out = new PrintWriter(socket.getOutputStream,true)
              while(true){
                Thread.sleep(args(1).toLong)
                val content = gennerateContent(index)
                println(content)
                out.write(content+"\n")
                out.flush()
              }
              socket.close()
            }
          }.start()
        }
      }
    }

用来接收数据的程序：

package sparkStreaming_
    
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    /** * Created by xiaopengpeng on 2016/12/11. * 这是一个接收来自网络端口的信息 * 参数 spark集群的主节点地址，网络通信的节点地址，网络通信的端口，每个多长时间作为一个单位进行执行任务 * local[*] localhost 8888 5 */
    class NetWorkWordCount { 
    
    }
    object NetWorkWordCount{ 
      def main(args: Array[String]): Unit = {
        if(args.length!=4){
          System.err.println("Usage:NetWorkWordCount<master> <hostname> <port> <seconds>\n+" +
            "In local modle,<master> should be 'local[n]' with n >1 ");
          System.exit(1);
        }
    
        val config = new SparkConf().setAppName("NetWorkWordCount").setMaster(args(0));
        val ssc = new StreamingContext(config,Seconds(args(3).toInt));
        val lines = ssc.socketTextStream(args(1),args(2).toInt,StorageLevel.MEMORY_ONLY_SER)
        val words = lines.flatMap(line => line.split(" "));
        val wordCount = words.map(x=>(x,1)).reduceByKey(_+_);
        wordCount.print()
        ssc.start();
        ssc.awaitTermination();
      }
    }