flink的(Sliding)滑动窗口

未分组全局执行的滑动窗口(Sliding)
滑动窗口一般用于股票的可视化实现,他不会忽然高了忽然低了,他会有一个过渡缓冲区.

public class Sliding1 {public static void main(String[] args)throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Integer> inte = lines.map(Integer::parseInt);AllWindowedStream<Integer, TimeWindow> sliding = inte.windowAll(SlidingProcessingTimeWindows.of(Time.seconds(20), Time.seconds(10)));SingleOutputStreamOperator<Integer> summed = sliding.sum(0);summed.print();env.execute();}
}
在8888端口一次输入:
123
123
123
123
123
在打印台打印出:
log4j:WARN No appenders could be found for logger (org.apache.flink.api.java.ClosureCleaner).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
4> 123
1> 246
2> 246
3> 246
4> 246

分组全局执行的滑动窗口(Sliding)

public class Sliding2 {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> line = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> wordandone = line.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String word) throws Exception {return Tuple2.of(word, 1);}});KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = wordandone.keyBy(0);WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> time = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));SingleOutputStreamOperator<Tuple2<String, Integer>> summed = time.sum(1);summed.print();env.execute();}
}
888端口输入:
shuju
shuju
shuju
shuju
打印台:
2> (shuju,1)
2> (shuju,2)
2> (shuju,2)
2> (shuju,2)

flink的(Sliding)滑动窗口相关推荐

Flink之Watermark滑动窗口案例
只要水印watermark的时间大于等于窗口的结束时间,并且窗口内有数据存在,就会触发对应窗口计算. 除此之外,如果flink配置了allowedLateness参数,只要水印watermark的时间 ...
Flink 滚动窗口、滑动窗口详解
1 滚动窗口(Tumbling Windows) 滚动窗口有固定的大小,是一种对数据进行"均匀切片"的划分方式.窗口之间没有重叠,也不会有间隔,是"首尾相接"的 ...
2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口
目录案例二基于数量的滚动和滑动窗口需求代码实现案例二基于数量的滚动和滑动窗口需求需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗 ...
2021年大数据Flink（十九）：案例一基于时间的滚动和滑动窗口
目录案例一基于时间的滚动和滑动窗口需求代码实现案例一基于时间的滚动和滑动窗口需求 nc -lk 9999 有如下数据表示: 信号灯编号和通过该信号灯的车的数量 9,3 9,2 9,7 4 ...
LeetCode 滑动窗口（Sliding Window）类问题总结
导语滑动窗口类问题是面试当中的高频题,问题本身其实并不复杂,但是实现起来细节思考非常的多,想着想着可能因为变量变化,指针移动等等问题,导致程序反复删来改去,有思路,但是程序写不出是这类问题最大的障碍 ...
python实现滑动窗口平均_数据流滑动窗口平均值 · sliding window average from data stream...
［抄题］: 给出一串整数流和窗口大小,计算滑动窗口中所有整数的平均值. MovingAverage m = new MovingAverage(3); m.next(1) = 1 // 返回 1.00 ...
[Swift]LeetCode480. 滑动窗口中位数 | Sliding Window Median
★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★ ➤微信公众号:山青咏芝(shanqingyongzhi) ➤博客园地址:山青咏芝(https://www.cnblog ...
Sliding Window(滑动窗口)
Sliding Window 题目描述 An array of size n ≤ 106 is given to you. There is a sliding window of size k wh ...
滑动窗口类型(Sliding window)
Sliding window,滑动窗口类型介绍部分来自:https://www.zhihu.com/question/36738189/answer/908664455 滑动窗口类型的题目经常是用来 ...

flink的(Sliding)滑动窗口

flink的(Sliding)滑动窗口

flink的(Sliding)滑动窗口相关推荐

最新文章

热门文章