windows下监听端口有很多工具，推荐一种：socketTool，linux下推荐使用nc命令，如nc -l 8080，可以收到8080端口的数据。
http://blog.csdn.net/haigenwong/article/details/26485745

目录
Spark DStream创建
通过监听端口创建DStream
通过队列创建DStream
通过自定义数据源创建DStream
通过Kafka数据源创建DStream
spark-streaming-kafka-0-8_2.11 模式
通过ReceiverAPI连接kafka数据源
通过DirectAPI连接kafka数据源

Spark DStream创建
通过监听端口创建DStream
使用netcat工具向9999端口不断的发送数据，通过SparkStreaming读取端口数据并统计不同单词出现的次数
//windows中命令
nc -lp 8888
//linux中命令
nc -lk 8888

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>2.1.1</version>
    <scope>compile</scope>
</dependency>

package com.xcu.bigdata.spark.streaming

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

/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期四
 * @Desc : 流式WordCount
 */
object SparkStreaming01_WordCount {
  def main(args: Array[String]): Unit = {
    // 创建配置文件对象
    // 注意：SparkStreaming程序执行至少需要2个线程，所以不能设置为local
    val conf = new SparkConf().setAppName("SparkStreaming01_WordCount").setMaster("local[*]")
    // 创建SparkStreaming程序执行入口
    // 设置采集周期为3s
    val ssc = new StreamingContext(conf, Seconds(3))
    // 从指定的端口获取数据
    val socketDS = ssc.socketTextStream("localhost", 8888)
    // 业务操作
    val faltMapDS = socketDS.flatMap(_.split(" "))
    val mapDS = faltMapDS.map((_, 1))
    val reduceDS = mapDS.reduceByKey(_ + _)
    reduceDS.print
    // 启动采集器
    ssc.start()
    // 等待采集结束之后，终止程序
    ssc.awaitTermination()
  }
}

注意：如果程序运行时，log日志太多，可以将spark conf目录下的log4j文件里面的日志拷贝到当前项目的resources目录下，并将级别改成WARN。
log4j.properties
log4j.rootCategory=ERROR, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n

# Set the default spark-shell log level to ERROR. When running the spark-shell, the
# log level for this class is used to overwrite the root logger's log level, so that
# the user can have different defaults for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.Main=ERROR

# Settings to quiet third party logs that are too verbose
log4j.logger.org.spark_project.jetty=ERROR
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=ERROR
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=ERROR
log4j.logger.org.apache.parquet=ERROR
log4j.logger.parquet=ERROR

# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR

通过队列创建DStream
package com.xcu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable

/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期四
 * @Desc :
 */
object SparkStreaming02_RDDQueue {
  def main(args: Array[String]): Unit = {
    //创建配置文件对象
    //注意Streaming程序执行至少需要2个线程，所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming02_RDDQueue").setMaster("local[*]")
    //创建SparkStreaming上下文环境对象
    //指定采集周期为3s
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //创建可变队列，里面放的是RDD
    val queue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]()
    //从队列中采集数据，获取DS
    val queueDS: InputDStream[Int] = ssc.queueStream(queue, false) //false在一个采集周期可以采集多个RDD
    //处理采集到的数据
    val resDS: DStream[(Int, Int)] = queueDS.map((_, 1)).reduceByKey(_ + _)
    resDS.print()
    //启动采集器
    ssc.start()
    //循环创建RDD，并将创建的RDD放到队列里
    for (i <- 1 to 5) {
      queue.enqueue(ssc.sparkContext.makeRDD(6 to 10))
      Thread.sleep(2000)
    }
    //等待采集结束之后，终止程序
    ssc.awaitTermination()

  }
}

通过自定义数据源创建DStream
需要继承Receiver，并实现onStart、onStop方法来自定义数据源采集。
需求：自定义数据源，实现监控某个端口号，获取该端口号内容
package com.xcu.bigdata.spark.streaming

import java.io.{BufferedReader, InputStreamReader}
import java.net.{ConnectException, Socket}
import java.nio.charset.StandardCharsets

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期四
 * @Desc : 通过自定义数据源方式创建DStream
 */
object SparkStreaming03_CustomerReceiver {
  def main(args: Array[String]): Unit = {
    //创建配置文件对象
    //注意Streaming程序执行至少需要2个线程，所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming03_CustomerReceiver").setMaster("local[*]")
    //创建SparkStreaming上下文环境对象
    //指定采集周期为3s
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //从端口中读取数据
    val myDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver("localhost", 7777))
    //扁平化
    val flatMapDS: DStream[String] = myDS.flatMap(_.split(" "))
    //结构转换
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_, 1))
    //聚合
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)
    //结果打印
    reduceDS.print
    //启动采集器
    ssc.start()
    //等待采集器采集结束之后，终止程序
    ssc.awaitTermination()
  }
}

//Receiver[T] T泛型即读取数据类型
class MyReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {
  private var socket: Socket = _

  //真正处理数据的逻辑
  def receive() {
    try {
      //创建连接
      socket = new Socket(host, port)
      //根据连接对象获取输入流
      //socket.getInputStream字节流-->InputStreamReader转换流-->BufferedReader缓存字符流
      val reader: BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))
      //定义一个变量,用于接收读取到的一行数据
      var line: String = null
      while ((line = reader.readLine()) != null) {
        store(line)
      }
    } catch {
      case e: ConnectException =>
        restart(s"Error connecting to $host:$port", e)
        return
    } finally {
      onStop()
    }
  }

  override def onStart(): Unit = {
    new Thread("Socket Receiver") {
      setDaemon(true) //设置后台守护线程

      override def run() {
        receive()
      }
    }.start()
  }

  override def onStop(): Unit = {
    synchronized {
      if (socket != null) {
        socket.close()
        socket = null
      }
    }
  }
}

通过Kafka数据源创建DStream
spark-streaming-kafka-0-8_2.11 模式
通过ReceiverAPI连接kafka数据源
1 需要一个专门的Executor去接收数据，然后发送给其他的Executor做计算。存在的问题，接收数据的Executor和计算的Executor速度会有所不同，特别在接收数据的Executor速度大于计算的Executor速度，会导致计算数据的节点内存溢出。
2 offset维护在zookeeper中，程序停止后，继续生产数据，再次启动程序，仍然可以继续消费。
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
    <version>2.1.1</version>
</dependency>

package com.xcu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期六
 * @Desc : 通过ReceiverAPI连接kafka数据源，获取数据
 *       需要导入依赖
 */
object SparkStreaming04_ReceiverAPI {
  def main(args: Array[String]): Unit = {
    //创建配置文件对象
    //注意：SparkStreaming程序执行至少需要2个线程，所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming04_ReceiverAPI").setMaster("local[*]")
    //创建SparkStreaming上下文环境对象,并指定采集周期为3s
    var ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //连接kafka，创建DStream
    val kafkaDStream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(
      ssc,
      "hadoop201:2181,hadoop202:2181,hadoop203:2181",
      "test0105",
      Map("bigdata-0105" -> 2)
    )
    //获取kafka中的消息，我们只需要v的部分
    val lineDS: DStream[String] = kafkaDStream.map(_._2)
    //扁平化
    val flatMapDS: DStream[String] = lineDS.flatMap(_.split(" "))
    //结构转换
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_, 1))
    //聚合
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)
    //打印输出
    reduceDS.print
    //开启任务
    ssc.start()
    ssc.awaitTermination()
  }
}

通过DirectAPI连接kafka数据源
这种API是由计算的Executor来主动消费Kafka的数据，速度由自身控制。

自动维护offset

offset维护在checkpoint中，获取StreamingContext的方式也改变了，但是目前这种方式会丢失消息
package com.xcu.bigdata.spark.streaming

import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期六
 * @Desc : 通过DirectAPI连接kafka数据源
 *       缺点：
 *       1 小文件过多
 *       2 在checkpoint中，只记录最后offset的时间戳，再次启动程序时，会从这个时间到当前时间，把所有的周期都执行一次
 */

object SparkStreaming05_DirectAPI_Auto {
  def main(args: Array[String]): Unit = {
    //修改StreamingContext对象的获取方式，先从检查点cp中获取，如果检查点cp没有，通过函数创建。
    val ssc: StreamingContext = StreamingContext.getActiveOrCreate("/cp", () => getStreamingContext)
    //启动采集器
    ssc.start()
    //等到采集器结束，终止程序
    ssc.awaitTermination()
  }

  def getStreamingContext(): StreamingContext = {
    //创建配置文件对象
    //注意：SparkStreaming程序执行至少需要2个线程，所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming05_DirectAPI_Auto01").setMaster("local[*]")
    //创建SparkStreaming上下文环境对象，指定采集周期为3s
    val ssc = new StreamingContext(conf, Seconds(3))
    //设置检测点路径
    ssc.checkpoint("/cp")
    //准备kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop201:9092,hadoop202:9092,hadoop203:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )
    val kafkaDstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc,
      kafkaParams,
      Set("bigdata-0105")
    )
    //获取kafka中的消息，我们只需要v的部分
    val lineDS: DStream[String] = kafkaDstream.map(_._2)
    //扁平化
    val flatMapRDD: DStream[String] = lineDS.flatMap(_.split(" "))
    //结构转换
    val mapDS: DStream[(String, Int)] = flatMapRDD.map((_, 1))
    //聚合
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)
    //打印输出
    reduceDS.print
    ssc
  }
}


手动维护offset

package com.xcu.bigdata.spark.streaming

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.{SparkConf}
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Package : com.xcu.bigdata.spark.streaming
 * @Author : 
 * @Date : 2020 10月 星期六
 * @Desc : DirectAPI连接kafka数据源（手动维护offset）
 */
object SparkStreaming07_DirectAPI_Hander {
  def main(args: Array[String]): Unit = {
    //创建配置文件对象
    //注意：SparkStreaming执行至少需要两个线程
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming07_DirectAPI_Hander").setMaster("local[*]")
    //创建SparkStreaming上下文环境对象
    //执行周期为3s
    val ssc = new StreamingContext(conf, Seconds(3))
    //准备kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop201:9092,hadoop202:9092,hadoop203:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )
    //获取上一次消费的位置(偏移量)
    //实际项目中，为了保证数据精准的一致性，我们对数据进行消费处理之后，通常将偏移量保存至有事务的存储中，如MySQL
    val fromOffsets: Map[TopicAndPartition, Long] = Map[TopicAndPartition, Long](
      TopicAndPartition("bigdata-0105", 0) -> 10L,
      TopicAndPartition("bigdata-0105", 1) -> 10L
    )
    //从指定的offset读取数据进行消费
    val kafkaDstream: InputDStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
      ssc,
      kafkaParams,
      fromOffsets,
      (m: MessageAndMetadata[String, String]) => m.message()
    )
    //消费完毕之后，对偏移量offset进行更新
    val offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]
    kafkaDstream.transform {
      rdd => {
        val offsetRnages: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    }.foreachRDD {
      rdd => {
        for (o <- offsetRanges) {
          println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
        }
      }
    }
    //启动采集器
    ssc.start()
    //等待采集器结束之后，结束程序
    ssc.awaitTermination()
  }
}

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



一、版本

通常的Linux发行版中都带有NetCat（简称nc），甚至在拯救模式光盘中也由busybox提供了简版的nc工具。但不同的版本，其参数的使用略有差异。

NetCat 官方地址：





引用[root@hatest1 ~]# cat /etc/asianux-release

Asianux release 2.0 (Trinity SP2)

[root@hatest1 ~]# cat /etc/redflag-release

Red Flag DC Server release 5.0 (Trinity SP2)

[root@hatest1 ~]# type -a nc

nc is /usr/bin/nc

[root@hatest1 ~]# rpm -q nc

nc-1.10-22



建议在使用前，先用man nc看看帮助。这里以红旗DC Server 5.0上的1.10版本进行简单说明。

假设两服务器信息：



server1: 192.168.10.10

server2: 192.168.10.11



二、常见使用





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



server2上运行： nc -l 1234 > text.txt



server1上运行： nc 192.168.10.11 1234 < text.txt



注：server2上的监听要先打开









2、克隆硬盘或分区

操作与上面的拷贝是雷同的，只需要由dd获得硬盘或分区的数据，然后传输即可。

克隆硬盘或分区的操作，不应在已经mount的的系统上进行。所以，需要使用安装光盘引导后，进入拯救模式（或使用Knoppix工具光盘）启动系统后，在server2上进行类似的监听动作：



 nc -l -p 1234 | dd of=/dev/sda



server1上执行传输，即可完成从server1克隆sda硬盘到server2的任务：



 dd if=/dev/sda | nc192.168.10.11 1234



※ 完成上述工作的前提，是需要落实光盘的拯救模式支持服务器上的网卡，并正确配置IP。








3、端口扫描

可以执行：



# nc -v -w 2 192.168.10.11 -z 21-24

nc: connect to 192.168.10.11 port 21 (tcp) failed: Connection refused

Connection to 192.168.10.11 22 port [tcp/ssh] succeeded!

nc: connect to 192.168.10.11 port 23 (tcp) failed: Connection refused

nc: connect to 192.168.10.11 port 24 (tcp) failed: Connection refused 

-z后面跟的是要扫描的端口











4、保存Web页面



# while true; do nc -l -p 80 -q 1 < somepage.html; done












5、模拟HTTP Headers



引用[root@hatest1 ~]# nc 80

GET / HTTP/1.1

Host: ispconfig.org

Referrer: mypage.com

User-Agent: my-browser



HTTP/1.1 200 OK

Date: Tue, 16 Dec 2008 07:23:24 GMT

Server: Apache/2.2.6 (Unix) DAV/2 mod_mono/1.2.1 mod_python/3.2.8 Python/2.4.3 mod_perl/2.0.2 Perl/v5.8.8

Set-Cookie: PHPSESSID=bbadorbvie1gn037iih6lrdg50; path=/

Expires: 0

Cache-Control: no-store, no-cache, must-revalidate, post-check=0, pre-check=0

Pragma: no-cache

Cache-Control: private, post-check=0, pre-check=0, max-age=0

Set-Cookie: oWn_sid=xRutAY; expires=Tue, 23-Dec-2008 07:23:24 GMT; path=/

Vary: Accept-Encoding

Transfer-Encoding: chunked

Content-Type: text/html

[......]



在nc命令后，输入红色部分的内容，然后按两次回车，即可从对方获得HTTP Headers内容。










6、聊天

nc还可以作为简单的字符下聊天工具使用，同样的，server2上需要启动监听：



server2上启动：# nc -lp 1234 

server1上传输：# nc 192.168.10.11 1234





这样，双方就可以相互交流了。使用Ctrl+D正常退出。













7、传输目录

从server1拷贝nginx-0.6.34目录内容到server2上。需要先在server2上，用nc激活监听，



server2上运行：# nc -l 1234 |tar xzvf -



server1上运行：# tar czvf - nginx-0.6.34|nc 192.168.10.11 1234



 





8、用nc命名操作memcached



1）存储数据：printf “set key 0 10 6rnresultrn” |nc 192.168.10.11 11211
2）获取数据：printf “get keyrn” |nc 192.168.10.11 11211
3）删除数据：printf “delete keyrn” |nc 192.168.10.11 11211
4）查看状态：printf “statsrn” |nc 192.168.10.11 11211
5）模拟top命令查看状态：watch “echo stats” |nc 192.168.10.11 11211
6）清空缓存：printf “flush_allrn” |nc 192.168.10.11 11211 (小心操作，清空了缓存就没了）












参数详解：
这仅是一个1.10版本的简单说明，详细的参数使用还是需要看man：
引用想要连接到某处: nc [-options] hostname port[s] [ports] ... 
绑定端口等待连接: nc -l -p port [-options] [hostname] [port] 
参数: 
-g gateway source-routing hop point[s], up to 8 
-G num source-routing pointer: 4, 8, 12, ... 
-h 帮助信息 
-i secs 延时的间隔 
-l 监听模式，用于入站连接 
-n 指定数字的IP地址，不能用hostname 
-o file 记录16进制的传输 
-p port 本地端口号 
-r 任意指定本地及远程端口 
-s addr 本地源地址 
-u UDP模式 
-v 详细输出——用两个-v可得到更详细的内容 
-w secs timeout的时间 

-z 将输入输出关掉——用于扫描时，其中端口号可以指定一个或者用lo-hi式的指定范围。 







1.84版本参数简介 
nc [-46DdhklnrStUuvzC] [-i interval] [-p source_port]
 [-s source_ip_address] [-T ToS] [-w timeout] [-X proxy_protocol] [-x  proxy_address[:port]]
 [hostname] [port[s]] 
 
 -4 强制使用ipv4 
 -6 强制使用ipv6 
 -D 允许socket通信返回debug信息 
 -d 不允许从标准输入中读取 
 -h 显示nc帮助文档 
 -i interval   指定每行之间内容延时发送和接受，也可以使多个端口之间的连接延时 
 -k 当一个连接结束时，强制nc监听另一个连接。必须和-l一起使用 
 -l 用于监听传入的数据链接，不能与-p -z -s一起使用。-w 参数的超时也会被忽略 
 -n 不执行任何地址，主机名，端口或DNS查询 
 -p 指定nc使用的源端口，受权限限制且不能余-l一起使用 
 -r 指定nc使用的源端口和目的端口，不能使用系统原来就指定的那些端口 
 -S 允许在RFC 2385的TCP MD5签名选项 
 -s source_ip_address  指定用于发包的接口的IP地址，不能和-l一起使用 
 -T ToS  指定链接的IP服务类型（TOS） 
 -C 自动换行 
 -t 使nc能够与telnet交互 
 -U 使用UNIX域socket 
 -u 使用udp代替默认的tcp选项 
  -v 输出详细报告 
 -w timeout  一个链接一段时间无操作，则自动断开，默认无超时 
 -X proxy_version   指定nc使用代理时所采用的协议,可选的有socksv4，socks5以及https。默认socks5 
 -x proxy_address[:port]    指定nc使用的代理地址和端口。默认设置:1080(SOCKS),3128(HTTPS) 
 -z 只监听不发送任何包

netstat -ano
查看所以的端口监听情况，数据比较多。


netstat -aon|findstr “6000”
从结果中过滤指定端口


tasklist|findstr “6292”
查看某个进程ID对应的进程名称