Window Operations
Spark Streaming还提供了窗口计算，允许您在数据的滑动窗口上应用转换。 下图说明了这个滑动窗口。
如图所示，每当窗口滑过源DStream时，落在窗口内的源RDD被组合并进行操作以产生窗口DStream的RDD。在这种具体情况下，操作应用于最近3个时间单位的数据，并以2个时间单位滑动。这表明任何窗口操作都需要指定两个参数。 窗口长度 - 窗口的持续时间（图中的3）。 滑动间隔 - 执行窗口操作的间隔（图中的2）。 这两个参数必须是源DStream的批间隔的倍数（图中的1）。
我们以一个例子来说明窗口操作。为了扩展以前的wordcount示例， 每隔10秒，统计前30秒的单词数。为此，我们必须在最近30秒的数据中对（word，1）对的对DStream应用reduceByKey操作。这是使用reduceByKeyAndWindow操作完成的。
// Reduce function adding two integers, defined separately for clarity
Function2 reduceFunc = new Function2() {
@Override public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
};
##reduceByKey 改为了reduceByKeyAndWindow
// Reduce last 30 seconds of data, every 10 seconds
JavaPairDStream windowedWordCounts = pairs.reduceByKeyAndWindow(reduceFunc,
Durations.seconds(30), //窗口长度30s
Durations.seconds(10)); //滑动间隔10s更多的Window操作-Window Operations
二、wordcount案例 2.1、分词，mapToPair()没有改变 2.2、统计单词数据由reduceByKey变为了reduceByKeyAndWindow 内部逻辑，还是对相同word进行累加 @Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
Durations.seconds(60), Durations.seconds(10) // 然后对这一个RDD根据每个搜索词出现频率进行排序然后获取排名前3热点搜索词,这里不用transform用transformToPair返回就是键值对
package com.chb.spark.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;
public class WindowBasedTopWord {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("wordcount").setMaster("local[2]");
JavaStreamingContext jssc = new JavaStreamingContext(conf,Durations.seconds(5));
// 这里日志简化, yasaka hello, lily world,这里日志简化主要是学习怎么使用Spark Streaming的
JavaReceiverInputDStream searchLog = jssc.socketTextStream("spark001", 9999);
// 将搜索日志转换成只有一个搜索词即可
JavaDStream searchWordDStream = searchLog.map(new Function(){
private static final long serialVersionUID = 1L;
@Override
public String call(String searchLog) throws Exception {
return searchLog.split(" ")[1];
}
});
// 将搜索词映射为(searchWord, 1)的Tuple格式
JavaPairDStream searchWordPairDStream = searchWordDStream.mapToPair(new PairFunction(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(String word) throws Exception {
return new Tuple2(word,1);
}
}) ;
JavaPairDStream searchWordCountsDStream =
searchWordPairDStream.reduceByKeyAndWindow(new Function2(){
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
}, Durations.seconds(60), Durations.seconds(10));
// 到这里就已经每隔10秒把之前60秒收集到的单词统计计数(Durations.seconds(5),每隔batch的时间间隔为5s, 所以共有12个RDD),执行transform操作因为一个窗口60秒数据会变成一个RDD
// 然后对这一个RDD根据每个搜索词出现频率进行排序然后获取排名前3热点搜索词,这里不用transform用transformToPair返回就是键值对
JavaPairDStream finalDStream = searchWordCountsDStream.transformToPair(
new Function(){
private static final long serialVersionUID = 1L;
@Override
public JavaPairRDD call(
JavaPairRDD searchWordCountsRDD) throws Exception {
// 反转
JavaPairRDD countSearchWordsRDD = searchWordCountsRDD
.mapToPair(new PairFunction(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(
Tuple2 tuple) throws Exception {
return new Tuple2(tuple._2,tuple._1);
}
});
//排序
JavaPairRDD sortedCountSearchWordsRDD = countSearchWordsRDD.
sortByKey(false);
//再次反转
JavaPairRDD sortedSearchWordsRDD = sortedCountSearchWordsRDD
.mapToPair(new PairFunction(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(
Tuple2 tuple) throws Exception {
return new Tuple2(tuple._2,tuple._1);
}
});
//获取前三个word
List topSearchWordCounts = sortedSearchWordsRDD.take(3);
//打印
for(Tuple2 wordcount : topSearchWordCounts){
System.out.println(wordcount._1 + " " + wordcount._2);
}
return searchWordCountsRDD;
}
} );
// 这个无关紧要,只是为了触发job的执行,所以必须有action操作
finalDStream.print();
jssc.start();
jssc.awaitTermination();
jssc.close();
}
}
