package com.lqs.sparkstreaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object Qs01_WordCount {
def main(args: Array[String]): Unit = {
//TODO 1、初始化SparkStreaming配置信息
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamin")
//TODO 2、初始化SparkStreamingContext
//Seconds(3):设置微批次流的间隔时间为3sec
val streamingContext = new StreamingContext(conf, Seconds(3))
//通过端口创建DStream，读取进来的数据为一行行
val lineDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("bdc112", 8888)
//将数据按行进行切分成一个个的单词
val wordDStream: DStream[String] = lineDStream.flatMap(_.split(" "))
//将单词映射成元组(word,1)
val wordToOneDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
//将相同的单词数做统计
val wordToSumDStream: DStream[(String, Int)] = wordToOneDStream.reduceByKey(_ + _)
//打印到控制台
wordToSumDStream.print()
//TODO 3、启动SparkStreamingContext
streamingContext.start()
//TODO 4、将主线程阻塞，主线程不退出
streamingContext.awaitTermination()
}
}
启动程序并通过netcat发送数据
[lqs@bdc112 datas]$ nc -lk 8888 lqs lqs Hello Word h^HH^H^H^H lqs Hello Woed lqs
运行结果：
------------------------------------------- Time: 1645009179000 ms ------------------------------------------- ------------------------------------------- Time: 1645009182000 ms ------------------------------------------- ------------------------------------------- Time: 1645009185000 ms ------------------------------------------- ,1) (Hello,1) (Word,1) (lqs,2) ------------------------------------------- Time: 1645009188000 ms ------------------------------------------- ------------------------------------------- Time: 1645009203000 ms ------------------------------------------- (Hello,1) (Woed,1) (lqs,1) ------------------------------------------- Time: 1645009206000 ms ------------------------------------------- (lqs,1)
