Flink 统计接入的数据量-滚动窗口和状态的使用
1、概述
在生产场景值,经常需要和上游、下游对数,离线场景可以直接 group by 再 count ,但是实时场景中,如果使用 kafka 作为中间件,中间经过几个 job 的过滤转化后,再对照像 Doris 或 Clickhouse 中最终层的数据,如果出现缺失,很难判断是哪一层缺失的。
2、使用 侧流输出+处理时间的滚动窗口+状态进行数据量级统计
package com.flink.feature.windowcount;import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
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.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;/*** 1、先输入数据* 1* 1* 1* 1* 输出结果* main=>:8> 业务处理=>1* main=>:1> 业务处理=>1* main=>:2> 业务处理=>1* main=>:3> 业务处理=>1* 每10秒每个key接受到的数据量=>:2> (1698913020000,1698913030000,窗口统计=>1,4)** 2、再输入数据* 1* 2* 2* 3* 3* 4* 4* 输出结果* main=>:4> 业务处理=>1* main=>:5> 业务处理=>2* main=>:6> 业务处理=>2* main=>:7> 业务处理=>3* main=>:8> 业务处理=>3* main=>:1> 业务处理=>4* main=>:2> 业务处理=>4* 每10秒每个key接受到的数据量=>:2> (1698913030000,1698913040000,窗口统计=>1,1)* 每10秒每个key接受到的数据量=>:7> (1698913030000,1698913040000,窗口统计=>4,2)* 每10秒每个key接受到的数据量=>:6> (1698913030000,1698913040000,窗口统计=>2,2)* 每10秒每个key接受到的数据量=>:6> (1698913030000,1698913040000,窗口统计=>3,2)*/public class UseWindowValidateData {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();OutputTag<Tuple2<String,Integer>> windowCountTag = new OutputTag<Tuple2<String,Integer>>("window_count"){};DataStreamSource<String> source = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<String> process = source.process(new ProcessFunction<String, String>() {@Overridepublic void processElement(String input, ProcessFunction<String, String>.Context ctx, Collector<String> collector) throws Exception {ctx.output(windowCountTag,new Tuple2<>("窗口统计=>"+input,1));collector.collect("业务处理=>" + input);}});process.getSideOutput(windowCountTag).keyBy(new KeySelector<Tuple2<String,Integer>, String>() {@Overridepublic String getKey(Tuple2<String,Integer> tp) throws Exception {return tp.f0;}}).window(TumblingProcessingTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple4<String,String,String,Integer>, String, TimeWindow>() {private MapState<String, Integer> mapState;@Overridepublic void open(Configuration parameters) throws Exception {MapStateDescriptor<String, Integer> stateDescriptor = new MapStateDescriptor<>("map-state", String.class, Integer.class);mapState = getRuntimeContext().getMapState(stateDescriptor);}@Overridepublic void process(String key, ProcessWindowFunction<Tuple2<String, Integer>, Tuple4<String, String, String, Integer>, String, TimeWindow>.Context ctx, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple4<String, String, String, Integer>> out) throws Exception {for (Tuple2<String, Integer> tp : elements) {Integer res = mapState.get(tp.f0);if (res == null) {res = 0;}res += 1;mapState.put(tp.f0, res);}out.collect(new Tuple4<>(String.valueOf(ctx.window().getStart()),String.valueOf(ctx.window().getEnd()),key,mapState.get(key)));// 每个窗口计算后清空状态mapState.clear();}}).print("每10秒每个key接受到的数据量=>");process.print("main=>");env.execute();}
}
3、测试结果
1)先输入数据
1
1
1
1
输出结果
main=>:8> 业务处理=>1
main=>:1> 业务处理=>1
main=>:2> 业务处理=>1
main=>:3> 业务处理=>1
每10秒每个key接受到的数据量=>:2> (1698913020000,1698913030000,窗口统计=>1,4)
2)再输入数据
1
2
2
3
3
4
4
输出结果
main=>:4> 业务处理=>1
main=>:5> 业务处理=>2
main=>:6> 业务处理=>2
main=>:7> 业务处理=>3
main=>:8> 业务处理=>3
main=>:1> 业务处理=>4
main=>:2> 业务处理=>4
每10秒每个key接受到的数据量=>:2> (1698913030000,1698913040000,窗口统计=>1,1)
每10秒每个key接受到的数据量=>:7> (1698913030000,1698913040000,窗口统计=>4,2)
每10秒每个key接受到的数据量=>:6> (1698913030000,1698913040000,窗口统计=>2,2)
每10秒每个key接受到的数据量=>:6> (1698913030000,1698913040000,窗口统计=>3,2)
相关文章:
Flink 统计接入的数据量-滚动窗口和状态的使用
1、概述 在生产场景值,经常需要和上游、下游对数,离线场景可以直接 group by 再 count ,但是实时场景中,如果使用 kafka 作为中间件,中间经过几个 job 的过滤转化后,再对照像 Doris 或 Clickhouse 中最终层…...
SpringBoot快速整合canal1.1.5(TCP模式)
SpringBoot快速整合canal1.1.5(TCP模式) 安装并配置MySQL主从⭐ 1:Docker安装MySQL8.0.28 docker pull mysql:8.0.282:创建目录: mkdir -p /usr/local/mysql8/data mkdir -p /usr/local/mysql8/log mkdir -p /usr/…...
docker打包container成image,然后将image上传到docker hub
第一步:停止正在运行的容器 docker stop <container_name> eg: docker stop xuanjie_mlir 第二步:将对应的container打包成image docker commit <container_id> <镜像名:版本> eg:docker commit 005672e6d97a…...
设计模式—创建型模式之原型模式
设计模式—创建型模式之原型模式 原型模式(Prototype Pattern)用于创建重复的对象,同时又能保证性能。 本体给外部提供一个克隆体进行使用。 比如我们做一个SjdwzMybatis,用来操作数据库,从数据库里面查出很多记录&…...
Zygote进程通信为什么用Socket而不是Binder?
Zygote进程是Android系统中的一个特殊进程,它在系统启动时被创建,并负责孵化其他应用进程。它的主要作用是预加载和共享应用进程的资源,以提高应用启动的速度。 在Android系统中,常用的进程通信方式有以下几种: Intent…...
API接口加密,解决自动化中登录问题
一、加密方式 AES:对称加密,快RAS:非对称加密,慢AESRAS:安全高效 加密过程:字符串》字节流》加密的字节流(算法),解密有可能出现乱码,所以不能直接转成字符…...
COCOS2DX3.17.2 Android升级targetSDK30问题解决方案
一、luajit不兼容问题 不兼容版本:【2.1.0-bate2、2.1.0-bate3都存在异常】 出问题系统:Android11;Android10的系统部分机型有问题,部分机型正常 异常点1:c调用lua接口,pushObjiect的时候crash 异常点2…...
HarmonyOS鸿蒙原生应用开发设计- 隐私声明
HarmonyOS设计文档中,为大家提供了独特的隐私声明,开发者可以根据需要直接引用。 开发者直接使用官方提供的隐私声明内容,既可以符合HarmonyOS原生应用的开发上架运营规范,又可以防止使用别人的内容产生的侵权意外情况等ÿ…...
【面试精选】00后卷王带你三天刷完软件测试面试八股文
前言 本人普通本科计算机专业,做测试也有3年的时间了,讲下我的经历,我刚毕业就进了一个小自研薪资还不错,有10.5k(个人觉得我很优秀),在里面呆了两年,积累了一些的经验和技能&#…...
k-means算法c++实现
计算数据集中的元素与各个簇的中心的距离,将它赋给最近的簇,然后重新计算每个簇的平均值,再将元素按离平均值点最近的原则重新分配直到没有出现重新分配 该算法要事先给出k的值,即划分为几个簇。 vector<int> datoclu(dat…...
oracle查询哪些用户下有表
oracle查询哪些用户下有表,排除系统用户。 在实际业务中 oracle数据库中创建了很多的用户 但实际都是无表的,利用SQL语句将这些有表的用户查询出来 并显示用户名、表名、创建表的时间等信息。 select * from dba_objects where object_type = TABLE and owner not in ( AN…...
机器人连杆惯量参数辨识(估计)
杆的转动惯量的计算公式是Imr^2。在经典力学中,转动惯量(又称质量惯性矩,简称惯矩)通常以I 或J表示,SI 单位为 kgm。对于一个质点,I mr,其中 m 是其质量,r 是质点和转轴的垂直距离。…...
一座 “数智桥梁”,华为助力“天堑变通途”
《水调歌头游泳》中的一句话,“一桥飞架南北,天堑变通途”,广为人们所熟知,其中展现出的,是中国人对美好出行的无限向往。 天堑变通途从来不易。 中国是当今世界上交通运输最繁忙、最快捷的国家之一,交通行…...
C#知识总结 基础篇(上)
本篇内容参考C#图解教程 本篇内容偏向基础,适合0基础的朋友快速上手,也适合有一定C语言(或其他语言如C,java)基础的人快速上手C#。同时适合unity引擎的初学者,更加详细的了解C#语言。 本文内容基本涵盖C#基础内容&am…...
照片编辑软件Affinity Photo 2 for Mac v2.1.1中文激活版 2024年最新中文版下载
照片编辑软件Affinity Photo 2 for Mac v2.1.1中文激活版是一款功能强大的专业级图像编辑软件,由Serif公司开发。它提供了广泛的工具和功能,适用于摄影师、设计师和艺术家。 照片编辑软件Affinity Photo 2 for Mac v2.1.1中文激活版软件介绍 TIFF&#…...
TPAMI 2023 | Temporal Perceiver:通用时序边界检测方法
本文介绍一下今年我们组被T-PAMI 2023收录的时序边界检测工作 Temporal Perceiver: A General Architecture for Arbitrary Boundary Detection。 论文名称: Temporal Perceiver: A General Architecture for Arbitrary Boundary Detection 论文链接&…...
Unity-UV展开工具
using System.Collections; using System.Collections.Generic; using UnityEngine; using UnityEditor;public class unfold : EditorWindow {[MenuItem("Gq_Tools/展开")]public static void ShowWin(){EditorWindow.CreateInstance<unfold>().Show();}priva…...
springboot actuator jvm监控丢失
1、背景 系统接入了监控prometheus和grafana,某天grafana突然发现只有几台机器可以看到指标。 随便点击一个地址http://192.168.0.76:8681/lms/actuator/prometheus访问指标,发现JVM相关指标全部丢失 2、解决方法 从网上查找相关资料,逐一…...
UDP服务端和客户端通信代码开发流程
一、UDP通信 TCP:传输控制协议,面向连接的,稳定的,可靠的,安全的数据集流传递 稳定和可靠:丢包重传 数据有序:序号和确认序号 流量控制:稳定窗口 UDP:用户数据报协议 面向无连接的,不稳定的,不可靠,不安…...
数据库实验:SQL的数据定义与单表查询
目录 实验目的实验内容实验要求实验过程实验步骤实例代码结果示意 数据库的实验,对关系型数据库MySQL进行一些实际的操作 实验目的 (1) 掌握DBMS的数据定义功能 (2) 掌握SQL语言的数据定义语句 (3) 掌握RDBMS的数据单表查询功能 (4) 掌握SQL语言的数据单表查询语句…...
Java 8 Stream API 入门到实践详解
一、告别 for 循环! 传统痛点: Java 8 之前,集合操作离不开冗长的 for 循环和匿名类。例如,过滤列表中的偶数: List<Integer> list Arrays.asList(1, 2, 3, 4, 5); List<Integer> evens new ArrayList…...
无法与IP建立连接,未能下载VSCode服务器
如题,在远程连接服务器的时候突然遇到了这个提示。 查阅了一圈,发现是VSCode版本自动更新惹的祸!!! 在VSCode的帮助->关于这里发现前几天VSCode自动更新了,我的版本号变成了1.100.3 才导致了远程连接出…...
java调用dll出现unsatisfiedLinkError以及JNA和JNI的区别
UnsatisfiedLinkError 在对接硬件设备中,我们会遇到使用 java 调用 dll文件 的情况,此时大概率出现UnsatisfiedLinkError链接错误,原因可能有如下几种 类名错误包名错误方法名参数错误使用 JNI 协议调用,结果 dll 未实现 JNI 协…...
2025盘古石杯决赛【手机取证】
前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来,实在找不到,希望有大佬教一下我。 还有就会议时间,我感觉不是图片时间,因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...
华硕a豆14 Air香氛版,美学与科技的馨香融合
在快节奏的现代生活中,我们渴望一个能激发创想、愉悦感官的工作与生活伙伴,它不仅是冰冷的科技工具,更能触动我们内心深处的细腻情感。正是在这样的期许下,华硕a豆14 Air香氛版翩然而至,它以一种前所未有的方式&#x…...
HDFS分布式存储 zookeeper
hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架,允许使用简单的变成模型跨计算机对大型集群进行分布式处理(1.海量的数据存储 2.海量数据的计算)Hadoop核心组件 hdfs(分布式文件存储系统)&a…...
MySQL 8.0 事务全面讲解
以下是一个结合两次回答的 MySQL 8.0 事务全面讲解,涵盖了事务的核心概念、操作示例、失败回滚、隔离级别、事务性 DDL 和 XA 事务等内容,并修正了查看隔离级别的命令。 MySQL 8.0 事务全面讲解 一、事务的核心概念(ACID) 事务是…...
Spring AOP代理对象生成原理
代理对象生成的关键类是【AnnotationAwareAspectJAutoProxyCreator】,这个类继承了【BeanPostProcessor】是一个后置处理器 在bean对象生命周期中初始化时执行【org.springframework.beans.factory.config.BeanPostProcessor#postProcessAfterInitialization】方法时…...
Redis上篇--知识点总结
Redis上篇–解析 本文大部分知识整理自网上,在正文结束后都会附上参考地址。如果想要深入或者详细学习可以通过文末链接跳转学习。 1. 基本介绍 Redis 是一个开源的、高性能的 内存键值数据库,Redis 的键值对中的 key 就是字符串对象,而 val…...
6.计算机网络核心知识点精要手册
计算机网络核心知识点精要手册 1.协议基础篇 网络协议三要素 语法:数据与控制信息的结构或格式,如同语言中的语法规则语义:控制信息的具体含义和响应方式,规定通信双方"说什么"同步:事件执行的顺序与时序…...
