使用idea编写Spark Streaming_WordCount

  1. 需求&准备
    图解:
    在这里插入图片描述
    首先在linux服务器上安装nc工具
    nc是netcat的简称,原本是用来设置路由器,我们可以利用它向某个端口发送数据
    yum install -y nc
    启动一个服务端并开放9999端口,等一下往这个端口发数据
    ==nc -lk 9999 ==
    发送数据
    代码演示:

package cn.itcast.streaming
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object WordCount {
def main(args: Array[String]): Unit = {
//1.创建StreamingContext
//spark.master should be set as local[n], n > 1
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进行操作
val dataDStream: ReceiverInputDStream[String] = ssc.socketTextStream(“node01”,9999)
//3.操作数据
val wordDStream: DStream[String] = dataDStream.flatMap(.split(" "))
val wordAndOneDStream: DStream[(String, Int)] = wordDStream.map((
,1))
val wordAndCount: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(+)
wordAndCount.print()
ssc.start()//开启
ssc.awaitTermination()//等待停止
}
}

2、执行
1.先执行nc -lk 9999
2.然后执行代码
3.不断的在1中输入不同的单词
hadoop spark sqoop hadoop spark hive hadoop
4.观察IDEA控制台输出
sparkStreaming每隔5s计算一次当前5s内的数据,然后将每个批次的数据输出

发布了238 篇原创文章 · 获赞 429 · 访问量 25万+

猜你喜欢

转载自blog.csdn.net/qq_45765882/article/details/105563034