- 需求&准备
●图解
●首先在linux服务器上安装nc工具
nc是netcat的简称,原本是用来设置路由器,我们可以利用它向某个端口发送数据
yum install -y nc
●启动一个服务端并开放9999端口,等一下往这个端口发数据
nc -lk 9999
●发送数据
-
代码演示
package SparkStrimng
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Created by 一个蔡狗 on 2020/4/10.
*
* sparkStreaming每隔5s计算一次当前5s内的数据,然后将每个批次的数据输出
*
* WroldCount
*/
object SparkStreaming_01 {
/* world count
*
* */
def main(args: Array[String]): Unit = {
//1.创建StreamingContext
val conf = new SparkConf().setAppName("wc").setMaster("local[*]")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")
val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD
//2.监听Socket接收数据
//ReceiverInputDStream就是接收到的所有的数据组成的RDD,封装成了DStream,接下来对DStream进行操作就是对RDD进行操作
ssc.socketTextStream("node001",9999).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
//第二种方式 : 2.监听Socket接收数据 并 操作数据
/* ssc.socketTextStream("node001",9999).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreachRDD( RDD=> {
//复杂代码逻辑
print()
})*/
ssc.start()//开启
ssc.awaitTermination()//等待停止
}
}