57、SparkStreaming:window滑动窗口以及热点搜索词滑动统计案例

Spark Streaming提供了滑动窗口操作的支持&＃xff0c;从而让我们可以对一个滑动窗口内的数据执行计算操作。每次掉落在窗口内的RDD的数据&＃xff0c;
会被聚合起来执行计算操作&＃xff0c;然后生成的RDD&＃xff0c;会作为window DStream的一个RDD。比如下图中&＃xff0c;就是对每三秒钟的数据执行一次滑动窗口计算&＃xff0c;
这3秒内的3个RDD会被聚合起来进行处理&＃xff0c;然后过了两秒钟&＃xff0c;又会对最近三秒内的数据执行滑动窗口计算。所以每个滑动窗口操作&＃xff0c;都必须指定
两个参数&＃xff0c;窗口长度以及滑动间隔&＃xff0c;而且这两个参数值都必须是batch间隔的整数倍。&＃xff08;Spark Streaming对滑动窗口的支持&＃xff0c;是比Storm更加完善和强大的&＃xff09;

package cn.spark.study.streaming;import java.util.List;import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;import scala.Tuple2;/*** 基于滑动窗口的热点搜索词实时统计* &＃64;author Administrator**/
public class WindowHotWord {public static void main(String[] args) {SparkConf conf &＃61; new SparkConf().setMaster("local[2]").setAppName("WindowHotWord"); JavaStreamingContext jssc &＃61; new JavaStreamingContext(conf, Durations.seconds(1));// 说明一下&＃xff0c;这里的搜索日志的格式// leo hello// tom worldJavaReceiverInputDStream searchLogsDStream &＃61; jssc.socketTextStream("spark1", 9999);// 将搜索日志给转换成&＃xff0c;只有一个搜索词&＃xff0c;即可JavaDStream searchWordsDStream &＃61; searchLogsDStream.map(new Function() {private static final long serialVersionUID &＃61; 1L;&＃64;Overridepublic String call(String searchLog) throws Exception {return searchLog.split(" ")[1];}});// 将搜索词映射为(searchWord, 1)的tuple格式JavaPairDStream searchWordPairDStream &＃61; searchWordsDStream.mapToPair(new PairFunction() {private static final long serialVersionUID &＃61; 1L;&＃64;Overridepublic Tuple2 call(String searchWord)throws Exception {return new Tuple2(searchWord, 1);}});// 针对(searchWord, 1)的tuple格式的DStream&＃xff0c;执行reduceByKeyAndWindow&＃xff0c;滑动窗口操作// 第二个参数&＃xff0c;是窗口长度&＃xff0c;这里是60秒// 第三个参数&＃xff0c;是滑动间隔&＃xff0c;这里是10秒// 也就是说&＃xff0c;每隔10秒钟&＃xff0c;将最近60秒的数据&＃xff0c;作为一个窗口&＃xff0c;进行内部的RDD的聚合&＃xff0c;然后统一对一个RDD进行后续// 计算// 所以说&＃xff0c;这里的意思&＃xff0c;就是&＃xff0c;之前的searchWordPairDStream为止&＃xff0c;其实&＃xff0c;都是不会立即进行计算的// 而是只是放在那里// 然后&＃xff0c;等待我们的滑动间隔到了以后&＃xff0c;10秒钟到了&＃xff0c;会将之前60秒的RDD&＃xff0c;因为一个batch间隔是&＃xff0c;5秒&＃xff0c;所以之前// 60秒&＃xff0c;就有12个RDD&＃xff0c;给聚合起来&＃xff0c;然后&＃xff0c;统一执行redcueByKey操作// 所以这里的reduceByKeyAndWindow&＃xff0c;是针对每个窗口执行计算的&＃xff0c;而不是针对某个DStream中的RDDJavaPairDStream searchWordCountsDStream &＃61; //Function2&＃xff1a;一个双参数函数&＃xff0c;它接受类型为T1和T2的参数并返回一个RsearchWordPairDStream.reduceByKeyAndWindow(new Function2() {private static final long serialVersionUID &＃61; 1L; &＃64;Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1 &＃43; v2;}}, Durations.seconds(60), Durations.seconds(10));// 到这里为止&＃xff0c;就已经可以做到&＃xff0c;每隔10秒钟&＃xff0c;出来&＃xff0c;之前60秒的收集到的单词的统计次数// 执行transform操作&＃xff0c;因为&＃xff0c;一个窗口&＃xff0c;就是一个60秒钟的数据&＃xff0c;会变成一个RDD&＃xff0c;然后&＃xff0c;对这一个RDD// 根据每个搜索词出现的频率进行排序&＃xff0c;然后获取排名前3的热点搜索词JavaPairDStream finalDStream &＃61; searchWordCountsDStream.transformToPair(new Function, JavaPairRDD>() {private static final long serialVersionUID &＃61; 1L;&＃64;Overridepublic JavaPairRDD call(JavaPairRDD searchWordCountsRDD) throws Exception {// 执行搜索词和出现频率的反转JavaPairRDD countSearchWordsRDD &＃61; searchWordCountsRDD.mapToPair(new PairFunction, Integer, String>() {private static final long serialVersionUID &＃61; 1L;&＃64;Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {return new Tuple2(tuple._2, tuple._1);}});// 然后执行降序排序JavaPairRDD sortedCountSearchWordsRDD &＃61; countSearchWordsRDD.sortByKey(false);// 然后再次执行反转&＃xff0c;变成(searchWord, count)的这种格式JavaPairRDD sortedSearchWordCountsRDD &＃61; sortedCountSearchWordsRDD.mapToPair(new PairFunction, String, Integer>() {private static final long serialVersionUID &＃61; 1L;&＃64;Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {return new Tuple2(tuple._2, tuple._1);}});// 然后用take()&＃xff0c;获取排名前3的热点搜索词List> hogSearchWordCounts &＃61; sortedSearchWordCountsRDD.take(3);for(Tuple2 wordCount : hogSearchWordCounts) {System.out.println(wordCount._1 &＃43; ": " &＃43; wordCount._2); }return searchWordCountsRDD;}});// 这个无关紧要&＃xff0c;只是为了触发job的执行&＃xff0c;所以必须有output操作
finalDStream.print();jssc.start();jssc.awaitTermination();jssc.close();}}##在eclipse中启动程序##服务器上启动nc&＃xff0c;并输入内容
[root&＃64;spark1 ~]# nc -lk 9999
leo hello
tom word
leo hello
jack you
leo you##统计结果
(hello,2)
(word,1)
(you,2)

package cn.spark.study.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds/*** &＃64;author Administrator*/
object WindowHotWord {def main(args: Array[String]): Unit &＃61; {val conf &＃61; new SparkConf().setMaster("local[2]") .setAppName("WindowHotWord")val ssc &＃61; new StreamingContext(conf, Seconds(1))val searchLogsDStream &＃61; ssc.socketTextStream("spark1", 9999) val searchWordsDStream &＃61; searchLogsDStream.map { _.split(" ")(1) } val searchWordPairsDStream &＃61; searchWordsDStream.map { searchWord &＃61;> (searchWord, 1) } val searchWordCountsDSteram &＃61; searchWordPairsDStream.reduceByKeyAndWindow((v1: Int, v2: Int) &＃61;> v1 &＃43; v2, Seconds(60), Seconds(10)) val finalDStream &＃61; searchWordCountsDSteram.transform(searchWordCountsRDD &＃61;> {val countSearchWordsRDD &＃61; searchWordCountsRDD.map(tuple &＃61;> (tuple._2, tuple._1)) val sortedCountSearchWordsRDD &＃61; countSearchWordsRDD.sortByKey(false) val sortedSearchWordCountsRDD &＃61; sortedCountSearchWordsRDD.map(tuple &＃61;> (tuple._1, tuple._2))val top3SearchWordCounts &＃61; sortedSearchWordCountsRDD.take(3)for(tuple <- top3SearchWordCounts) {println(tuple)}searchWordCountsRDD})finalDStream.print()ssc.start()ssc.awaitTermination()}}##在eclipse中启动程序##服务器上启动nc&＃xff0c;并输入内容
[root&＃64;spark1 ~]# nc -lk 9999
leo hello
leo hello
leo hello
leo word
leo word
leo word
leo hello
leo you
leo you##统计结果
(hello,4)
(word,3)
(you,2)