案例三会话窗口

需求

设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算

代码实现


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;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;/*** Author lanson* Desc* nc -lk 9999* 有如下数据表示:* 信号灯编号和通过该信号灯的车的数量
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4* 需求:设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算(前提是上一个窗口得有数据!)*/
public class WindowDemo03_SessionWindow {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]));}});//需求:设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算(前提是上一个窗口得有数据!)SingleOutputStreamOperator<CartInfo> result = cartInfoDS.keyBy(CartInfo::getSensorId).window(ProcessingTimeSessionWindows.withGap(Time.seconds(10))).sum("count");//4.Sinkresult.print();//5.executeenv.execute();}@Data@AllArgsConstructor@NoArgsConstructorpublic static class CartInfo {private String sensorId;//信号灯idprivate Integer count;//通过该信号灯的车的数量}
}

2021年大数据Flink（二十一）：案例三会话窗口相关推荐

2021年大数据Flink（十一）：流批一体API Source
目录 Source 预定义Source 基于集合的Source 基于文件的Source 基于Socket的Source 自定义Source 随机生成数据 MySQL Sou ...
2021年大数据HBase（十一）：Apache Phoenix的视图操作
全网最详细的大数据HBase文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章前言 Apache Phoenix的视图操作一.应用场景 ...
2021年大数据Hive（十一）：Hive调优
全网最详细的大数据Hive文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章前言 Hive调优一.本地模式 1.空key处理二.SQL ...
2021年大数据Kafka（十一）：❤️Kafka的消费者负载均衡机制和数据积压问题❤️
全网最详细的大数据Kafka文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章 Kafka的消费者负载均衡机制和数据积压问题一.kafka ...
2021年大数据Flink（三十九）：Table与SQL 总结 Flink-SQL常用算子
目录总结 Flink-SQL常用算子 SELECT WHERE DISTINCT GROUP BY UNION 和 UNION ALL JOI ...
2021年大数据Flink（三十六）：Table与SQL 案例三
目录案例三需求编码步骤代码实现-方式1 代码实现-方式2 案例三需求使用Flink SQL来统计5秒内每个用户的订单总数.订单的最大金额.订单的最小金额也就是每隔5秒统计最近5秒的每 ...
2021年大数据Hadoop（十一）：HDFS的元数据辅助管理
2021大数据领域优质创作博客,带你从入门到精通,该博客每天更新,逐渐完善大数据各个知识体系的文章,帮助大家更高效学习. 有对大数据感兴趣的可以关注微信公众号:三帮大数据目录 HDFS的元数据辅助管 ...
2021年大数据Flink（四十八）：扩展阅读 Streaming File Sink
目录扩展阅读 Streaming File Sink 介绍场景描述 Bucket和SubTask.PartFile 案例演示扩展阅读配置详解 PartFile PartFile序列化编码 ...
2021年大数据Flink（四十五）：扩展阅读双流Join
目录扩展阅读双流Join 介绍 Window Join Interval Join 代码演示1 代码演示2 重点注意扩展阅读双流Join 介绍 https:// ...

2021年大数据Flink（二十一）：案例三会话窗口

案例三会话窗口

需求

代码实现

2021年大数据Flink（二十一）：案例三会话窗口相关推荐

最新文章

热门文章

2021年大数据Flink（二十一）：​​​​​​​案例三 会话窗口

案例三 会话窗口

需求

代码实现

2021年大数据Flink（二十一）：​​​​​​​案例三 会话窗口相关推荐

最新文章

热门文章

2021年大数据Flink（二十一）：案例三会话窗口

案例三会话窗口

2021年大数据Flink（二十一）：案例三会话窗口相关推荐