版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/zhangfengBX/article/details/80659612
Window函数，可以统计最近一段时间的数据，使用Window函数加载成DStream：DStream.window("窗口长度","滑动间隔")
reduceByKeyAndWindow
窗口长度：必须是BathInterval的整数倍
滑动间隔：必须是BatchInterval的整数倍
/**
* 1、local的模拟线程数必须大于等于2 因为一条线程被receiver(接受数据的线程)占用，另外一个线程是job执行
* 2、Durations时间的设置，就是我们能接受的延迟度，这个我们需要根据集群的资源情况以及监控每一个job的执行时间来调节出最佳时间。
* 3、 创建JavaStreamingContext有两种方式 （sparkconf、sparkcontext）
* 4、业务逻辑完成后，需要有一个output operator
* 5、JavaStreamingContext.start()straming框架启动之后是不能在次添加业务逻辑
* 6、JavaStreamingContext.stop()无参的stop方法会将sparkContext一同关闭，stop(false) ,默认为true，会一同关闭
* 7、JavaStreamingContext.stop()停止之后是不能在调用start
*/
public class WindowOperator {
@SuppressWarnings("resource")
public static void main(String[] args) {
SparkConf conf = new SparkConf();
//local[2]模拟线程数
conf.setMaster("local[2]").setAppName("ww");
//在此处设置最小间隔时间durations
JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5));
jsc.sparkContext().setLogLevel("WARN");
jsc.checkpoint("./check");
//设置监听的节点和ip端口
JavaReceiverInputDStream<String> sts = jsc.socketTextStream("CentOS16",9999);
//设置文件保留时间，之后得而RDD使用Window时，只会取到最近60秒的数据
JavaDStream<String> window = sts.window(Durations.seconds(60),Durations.seconds(5));
JavaDStream<String> flatMap = sts.flatMap(new FlatMapFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> call(String arg0) throws Exception {
return Arrays.asList(arg0.split(" "));
}
});
JavaPairDStream<String,Integer> mapToPair = flatMap.mapToPair(new PairFunction<String, String, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, Integer> call(String arg0) throws Exception {
return new Tuple2<String, Integer>(arg0,1);
}
});
/**
* 每隔5秒，计算最近60秒内的数据，那么这个窗口大小就是60秒，里面有12个rdd，在没有计算之前，这些rdd是不会进行计算的。
* 那么在计算的时候会将这12个rdd聚合起来，然后一起执行reduceByKeyAndWindow操作 ，
* reduceByKeyAndWindow是针对窗口操作的而不是针对DStream操作的。
*/
// JavaPairDStream<String, Integer> reduceByKeyAndWindow =
// mapToPair.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
// private static final long serialVersionUID = 1L;
// @Override
// public Integer call(Integer arg0, Integer arg1) throws Exception {
// return arg0+arg1;
// }
// }, Durations.seconds(30),Durations.seconds(5));
/**
* window窗口操作优化，设置加上后面10s的数据，然后减去统计在内的最前面10s的数据
* 可以看成只产生了两个rdd，在计算之后，这些rdd才开始计算
*/
JavaPairDStream<String, Integer> reduceByKeyAndWindow = mapToPair.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer arg0, Integer arg1) throws Exception {
return arg0+arg1;
}
}, new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer arg0, Integer arg1) throws Exception {
return arg0-arg1;
}
//下面的两个时间必须是前面设置的durations属性的整数倍
},Durations.seconds(60),Durations.seconds(10));
reduceByKeyAndWindow.print();
jsc.start();
jsc.awaitTermination();
jsc.stop(false);
jsc.close();
}
}
