案例二基于数量的滚动和滑动窗口

需求

需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口

需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口

代码实现

package cn.it.window;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** Author lanosn* Desc* nc -lk 9999* 有如下数据表示:* 信号灯编号和通过该信号灯的车的数量
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4* 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口* 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口*/
public class WindowDemo02_CountWindow {public static void main(String[] args) throws Exception {//1.envStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2.SourceDataStreamSource<String> socketDS = env.socketTextStream("node1", 9999);//3.Transformation//将9,3转为CartInfo(9,3)SingleOutputStreamOperator<CartInfo> cartInfoDS = socketDS.map(new MapFunction<String, CartInfo>() {@Overridepublic CartInfo map(String value) throws Exception {String[] arr = value.split(",");return new CartInfo(arr[0], Integer.parseInt(arr[1]));}});//分组//KeyedStream<CartInfo, Tuple> keyedDS = cartInfoDS.keyBy("sensorId");// * 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口//countWindow(long size, long slide)SingleOutputStreamOperator<CartInfo> result1 = cartInfoDS.keyBy(CartInfo::getSensorId)//.countWindow(5L, 5L).countWindow( 5L).sum("count");// * 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口//countWindow(long size, long slide)SingleOutputStreamOperator<CartInfo> result2 = cartInfoDS.keyBy(CartInfo::getSensorId).countWindow(5L, 3L).sum("count");//4.Sink//result1.print();/*
1,1
1,1
1,1
1,1
2,1
1,1*/result2.print();/*
1,1
1,1
2,1
1,1
2,1
3,1
4,1*///5.executeenv.execute();}@Data@AllArgsConstructor@NoArgsConstructorpublic static class CartInfo {private String sensorId;//信号灯idprivate Integer count;//通过该信号灯的车的数量}
}

2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口相关推荐

2021年大数据Flink（十九）：案例一基于时间的滚动和滑动窗口
目录案例一基于时间的滚动和滑动窗口需求代码实现案例一基于时间的滚动和滑动窗口需求 nc -lk 9999 有如下数据表示: 信号灯编号和通过该信号灯的车的数量 9,3 9,2 9,7 4 ...
2021年大数据Flink（十八）：Flink Window操作
目录 Flink-Window操作为什么需要Window Window的分类按照time和count分类按照slide和size分类总结 Window ...
2021年大数据Flink（十二）：流批一体API Transformation
目录 Transformation 官网API列表基本操作-略 map flatMap keyBy filter sum reduce 代码演示合并-拆分 union和connect split. ...
2021年大数据Flink（十五）：流批一体API Connectors Kafka
目录 Kafka pom依赖参数设置参数说明 Kafka命令代码实现-Kafka Consumer 代码实现-Kafka Producer 代码实现-实时ETL Kafka pom依赖 Flin ...
2021年大数据Flink（十）：流处理相关概念
目录流处理相关概念数据的时效性流处理和批处理流批一体API DataStream API 支持批执行模式 API 编程模型流处理相关概念数据的时效 ...
2021年大数据Flink（十六）：流批一体API Connectors Redis
目录 Redis API 使用RedisCommand设置数据结构类型时和redis结构对应关系需求代码实现 Redis API 通过flink 操作redis 其实我们可以通过传统的redis ...
2021年大数据Flink（十四）：流批一体API Connectors JDBC
目录 Connectors JDBC 代码演示 Connectors JDBC Apache Flink 1.12 Documentation: JDBC Connector 代码演示 package ...
2021年大数据Hive（十二）：Hive综合案例！！！
全网最详细的大数据Hive文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章前言 Hive综合案例一.需求描述二.项目表的字段三.进 ...
2021年大数据HBase（十二）：Apache Phoenix 二级索引
全网最详细的大数据HBase文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章前言 Apache Phoenix 二级索引一.索引分类 ...

2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口

案例二基于数量的滚动和滑动窗口

需求

代码实现

2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口相关推荐

最新文章

热门文章

2021年大数据Flink（二十）：案例二 基于数量的滚动和滑动窗口

案例二 基于数量的滚动和滑动窗口

需求

代码实现

2021年大数据Flink（二十）：案例二 基于数量的滚动和滑动窗口相关推荐

最新文章

热门文章

2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口

案例二基于数量的滚动和滑动窗口

2021年大数据Flink（二十）：案例二基于数量的滚动和滑动窗口相关推荐