package com.shujia.spark.streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} import org.apache.spark.streaming.{Durations, StreamingContext} object Demo1WordCount { def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf() .setAppName("streaming") .setMaster("local[2]") /** * 创建streaming上下文对象对象,指定batch时间,多久计算一次 * */ val ssc = new StreamingContext(conf, Durations.seconds(5)) //设置checkpoint路径 ssc.checkpoint("data/checkpoint") /** * ReceiverInputDStream:被动接收数据,将接收过来的数据放在内存或者磁盘上 * 接收数据会一直占用资源,所以资源给多一点 local[2] * * nc -lk 8888 * yum install nc (如果没有上述命令 ,安装 ) */ //读取数据 val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888) //统计单词数量 val wordsDS: DStream[String] = linesDS.flatMap(_.split(",")) val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1)) /** * reduceByKey:只统计当前batch的数据,不会进行累加计算 */ // val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _) /** * * @param seq :当前batch 每一个key所有的value * @param option : 之前batch累加计算的结果 * @return : 返回最新的单词的数量 */ def updateFun(seq: Seq[Int], option: Option[Int]): Option[Int] = { //计算当前batch单词的数量 val currCount: Int = seq.sum //获取之前单词的数量 val lastCount: Int = option.getOrElse(0) //返回最新单词的数量 Some(currCount + lastCount) } /** * 有状态算子 * updateStateByKey:每一次计算更新每一个key的状态(单词的数量) * * 需要设置checkpoint的路径,用于保存计算中的状态 */ val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateFun) //打印数据 countDS.print() //启动streaming ssc.start() ssc.awaitTermination() //等待关闭 这三行代码必须要写 ssc.stop() } }