IDEA Spark Streaming 操作(套接字流)

Posted 2020-10-10 soyosuyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DStream_socket {
  def main(args: Array[String]): Unit = {
       val Conf=new SparkConf().setAppName("套接字流").setMaster("local[2]")
       val ss=new StreamingContext(Conf,Seconds(20))
       val lines=ss.socketTextStream("localhost",6666)      //端口随便设置 只要别占用正在使用的就行（nc -v -w 1 localhost -z 0-100 查看已占用端口）

       val words=lines.flatMap(_.split(" "))
       val wordCount=words.map(x=>(x,1)).reduceByKey((x,y)=>x+y)
        wordCount.print(100)
        ss.start()
       ss.awaitTermination()
  }
}

 运行程序之前使6666处于监听模式：nc -l 6666

之后输入的数据会被程序捕捉到

nc命令详解：

NetCat，在网络工具中有“瑞士军刀”美誉，其有Windows和Linux的版本。因为它短小精悍（1.84版本也不过25k，旧版本或缩减版甚至更小）、功能实用，被设计为一个简单、可靠的网络工具，可通过TCP或UDP协议传输读写数据。同时，它还是一个网络应用Debug分析器，因为它可以根据需要创建各种不同类型的网络连接。

 

版本参数简介
语　　法：nc [-hlnruz][-g<网关...>][-G<指向器数目>][-i<延迟秒数>][-o<输出文件>][-p<通信端口>][-s<来源地址>][-v...][-w<超时秒数>][主机名称][通信端口...]

补充说明：执行本指令可设置路由器的相关参数。

参　　数：

-g<网关> 设置路由器跃程通信网关，最多可设置8个。

-G<指向器数目> 设置来源路由指向器，其数值为4的倍数。

-h 在线帮助。

-i<延迟秒数> 设置时间间隔，以便传送信息及扫描通信端口。

-l 使用监听模式，管控传入的资料。

-n 直接使用IP地址，而不通过域名服务器。

-o<输出文件> 指定文件名称，把往来传输的数据以16进制字码倾倒成该文件保存。

-p<通信端口> 设置本地主机使用的通信端口。

-r 乱数指定本地与远端主机的通信端口。

-s<来源地址> 设置本地主机送出数据包的IP地址。

-u 使用UDP传输协议。

-v 显示指令执行过程。

-w<超时秒数> 设置等待连线的时间。

-z 使用0输入/输出模式，只在扫描通信端口时使用。

使用场景：

   1.远程拷贝文件

从server1拷贝文件到server2上。需要先在server2上，用nc激活监听。

server2: nc -l 6666 >soyo.txt (soyo.txt 是你希望存数据的文件名）

server1:nc -w 1 localhost 6666 < abc.txt  (abc.txt 的内容复制到了 soyo.txt)

   2.端口扫描

nc -v -w 1 localhost -z 0-100 查看已占用端口
3.简单的聊天 (这个挺好玩）
开两个shell 1 和2
1：nc -l 6666 
2.nc localhost 6666