Flink(java版)
watermark
时间语义和 watermark
注意:数据进入flink的时间:如果用这个作为时间语义就不存在问题,但是开发中往往会用处理时间
作为时间语义这里就需要考虑延时的问题。
如上图,数据从kafka中获取出来,从多个分区中获取,这时候时间肯定有乱序,这时候就需要使用事
件时间。
场景:游戏连续过五关,给予奖励
地铁里面玩游戏,连过三关断网了,二分钟过了八关。这时候是用处理时间还是事件时间呢?
处理时间的优势:牺牲一定的数据准确性,没有延迟
package com.atguigu.apitest.window;/**import com.atguigu.apitest.beans.SensorReading;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;public class WindowTest3_EventTimeWindow {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//默认为当前机器的cpu的最大核数//env.setParallelism(1);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.getConfig().setAutoWatermarkInterval(100);// socket文本流DataStream<String> inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型,分配时间戳和watermarkDataStream<SensorReading> dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));})// 乱序数据设置时间戳和watermark.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {@Overridepublic long extractTimestamp(SensorReading element) {return element.getTimestamp() * 1000L;}});OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") {};// 基于事件时间的开窗聚合,统计15秒内温度的最小值SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id").timeWindow(Time.seconds(15)).allowedLateness(Time.minutes(1)).sideOutputLateData(outputTag).minBy("temperature");minTempStream.print("minTemp");minTempStream.getSideOutput(outputTag).print("late");env.execute();}
}
sensor_1,1547718199,35.8
sensor_6,1547718201,15.4
sensor_7,1547718202,6.7
sensor_10,1547718205,38.1
sensor_1,1547718207,36.3
sensor_1,1547718211,32.8
sensor_1,1547718212,37.1注意:第一个窗口是[1547718195,1547718210);
sensor_1,1547718213,33
sensor_1,1547718224,32.1
sensor_1,1547718225,31.6
sensor_1,1547718226,21.2
sensor_1,1547718227,33.6第二个窗口大小:第一个窗口是[1547718210,1547718225);
1.理想状态:来一条数据处理一条,每条数据代表对时间推进;如图到5之后就将【0,5)的窗口关闭并输出;2.乱序状态:原因:网络延迟、分布式、分区导致乱序数据产生;网络延迟和分布式处理造成的乱序都是几十毫秒和几百毫秒的范围的差距;这将回造成大多数延迟数据集中在几十毫秒和几百毫秒的范围内;3.解决方案:将时间事件放慢
flink的三重保证:1.设置watermaker将几百毫秒的数据全部输出;2.先输出一个近似的结果,但是不要关闭窗口后面延迟的时间还需要更新;3.当延时时间到了,窗口就关闭了;兜底方案使用侧输出流保证数据不丢失;注意:数据流中的 Watermark 用于表示 timestamp 小于 Watermark 的数据都已经到
达了,因此,window 的执行也是由 Watermark 触发的。
6 3 2 5 4 1
比如设置3秒的watermaker:
到达5:说明2秒之前的数据都到齐了,后面2,3都可以输出
到达6:说明3秒之前的数据都到齐了,大于等于3秒的数据才能输出意义:watermark 用来让程序自己平衡延迟和结果正确性:如果设置太大延迟太高,设置太
小数据就不准确,需要通过具体的业务场景去平衡这个值;
watermark 用来让程序自己平衡延迟和结果正确性:如果设置太大延迟太高,设置太小,乱序数据
没有搞定,数据就不准确,需要通过具体的业务场景去平衡这个值;如何找到watermaker:首先要了解乱序程度;
解决方案:通过机器学习构建一个模型,构建当前业务模型中的延迟状态的分布情况;
如图:大部分的延时数据都20ms和80ms之间的范围中,这时候设置80ms就搞定大部分乱序数据;
这时候还有很少的数据,如果对数据准确性要求比较高,这时候就需要设置窗口迟到机制去保证
数据的准备性;最后还有网络延迟的数据还是没有输出这时候就需要添加侧输出流作为兜底方案。
watermark 生成问题
默认:来一条生产一条watermaker,如果短时间数据量比较大,会造成watermaker都一样造成资
源浪费;周期性添加watermaker:每隔一段时间更新一下watermaker
周期性时间缺点:实时性不好;数据过于分散会造成资源浪费;如何选择:看数据的分布,过于集中使用周期性生成模式,数据稀疏,使用默认的模型;
状态编程
需求:我们可以利用 Keyed state,实现这样一个需求: 检测传感器的温度值,如果连续的两个温度差值超过 10 度,就输出报警
package com.atguigu.apitest.state;/*** Copyright (c) 2018-2028 尚硅谷 All Rights Reserved* <p>* Project: FlinkTutorial* Package: com.atguigu.apitest.state* Version: 1.0* <p>* Created by wushengran on 2020/11/10 16:33*/import com.atguigu.apitest.beans.SensorReading;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** @ClassName: StateTest3_KeyedStateApplicationCase* @Description:* @Author: wushengran on 2020/11/10 16:33* @Version: 1.0*/
public class StateTest3_KeyedStateApplicationCase {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// socket文本流DataStream<String> inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型DataStream<SensorReading> dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));});// 定义一个flatmap操作,检测温度跳变,输出报警SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy("id").flatMap(new TempChangeWarning(10.0));resultStream.print();env.execute();}// 实现自定义函数类public static class TempChangeWarning extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>>{// 私有属性,温度跳变阈值private Double threshold;public TempChangeWarning(Double threshold) {this.threshold = threshold;}// 定义状态,保存上一次的温度值private ValueState<Double> lastTempState;@Overridepublic void open(Configuration parameters) throws Exception {lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class));}@Overridepublic void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {// 获取状态Double lastTemp = lastTempState.value();// 如果状态不为null,那么就判断两次温度差值if( lastTemp != null ){Double diff = Math.abs( value.getTemperature() - lastTemp );if( diff >= threshold )out.collect(new Tuple3<>(value.getId(), lastTemp, value.getTemperature()));}// 更新状态lastTempState.update(value.getTemperature());}@Overridepublic void close() throws Exception {lastTempState.clear();}}
}
sensor_1,1547718206,36.3
sensor_1,1547718206,37.9
sensor_1,1547718206,48
sensor_6,1547718201,15.4
sensor_6,1547718201,35
sensor_1,1547718226,36
状态后端
状态后端: 1.本地的状态管理(如何存,上下文配置,怎么存,怎么写) 2.做快照容错,如何恢复数据
1. 测试环境:MemoryStateBackend
2. 生产环境:FsStateBackend
3. 数据非常大时候:RocksDBStateBackend
state.backend: filesystem //默认使用FsStateBackend
tate.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints
//配置一个checkpoint的hdfs的存储路径jobmanager.execution.failover-strategy: region //区域化重启state.backend.incremental: false //增量添加checkpoint
相关文章:

Flink(java版)
watermark 时间语义和 watermark 注意:数据进入flink的时间:如果用这个作为时间语义就不存在问题,但是开发中往往会用处理时间 作为时间语义这里就需要考虑延时的问题。 如上图,数据从kafka中获取出来,从多个分区中获取…...
什么是动态组件以及使用场景
文章目录 一、vue中的动态组件是什么?有什么用?二、使用demo1.tab页签中的使用2.模拟新闻页demo 三、用keep-alive包裹,保持状态总结 一、vue中的动态组件是什么?有什么用? 动态组件指可以动态切换组件的显示和隐藏。…...

CRM销售管理系统如何提高销售效率
CRM销售管理系统是帮助企业对销售活动进行管理、执行和优化的软件系统。它可以帮助企业提高销售效率,提高客户转化率,实现企业的业绩增长。那么,CRM销售管理系统好用吗? CRM销售管理系统的功能 线索管理: CRM系统可…...

纯小白安卓刷机1
文章目录 常见的英文意思刷机是什么?为什么要刷机?什么是BL锁(BootLoader锁)?我的机能够刷机吗?什么是Boot镜像/分区?什么是Recovery镜像/分区(缩写为rec)?什…...

C高级day4循环语句
1,思维导图 运行结果为: 运行结果为:...

Linux 操作系统云服务器安装部署 Tomcat 服务器详细教程
Tomcat 基本概述 Tomcat 服务器是Apache软件基金会(Apache Software Foundation)的 Jakarta 项目中的一个核心项目,由 Apache、Sun 和其他一些公司及个人共同开发而成。它是一个免费的开放源代码的 Web 应用服务器,属于轻量级应用…...

【易盾点选】
拿官网的点选做个例子吧,比较省事,水一篇~ 官网的接口目前都改成V3了,多了个dt参数,以及加密的一个函数也变动了下 点选坐标在这,加密函数未变,用逗号拼接 整个加密里的函数变了,直接重新…...
vue中打印指定dom元素
和window.print()效果一样,调出打印窗口,只是当前使用的插件是vue-print-nb 官网地址:vue-print-nb vue2中使用 安装插件 npm install vue-print-nb --save导入插件 import Print from vue-print-nb // 全局使用 Vue.use(Print);//or// 单…...

OpenCV(三十六):霍夫直线检测
1.检测直线的霍夫变换原理 2.检测直线函数HoughLines() 检测直线流程: Step1:将参数空间的坐标轴离散化。 Step2:将图像中每个非0像素通过映射关系求取在参数空间通过的方格 Step3:统计参数空间内每个方格出现的次数,选取次数大于某一值的方格作为表示直线的方格…...

文心一言 VS 讯飞星火 VS chatgpt (91)-- 算法导论8.3 4题
四、用go语言,说明如何在 O(n) 时间内,对 0 到 n 3 − 1 n^3-1 n3−1 区间内的n个整数进行排序。 文心一言: 在 O(n) 时间内对 0 到 n^3-1 区间内的 n 个整数进行排序,可以使用基数排序(Radix Sort)算法…...

keep-alive缓存三级及三级以上路由
需求需要缓存这个出入记录,当tab切换时不重新加载,当刷新页面时,或把这个关闭在重新打开时重新加载如图: (我这里用的是芋道源码的前端框架) keep-alive 1、include 包含页面组件name的这些组件页面,会被…...

vite vue项目 运行时 \esbuild\esbuild.exe 缺失 错误码 errno: -4058, code: ‘ENOENT‘,
vite vue项目运行 npm run dev 报错某个模块启动文件丢失信息 D:\PengYe_code\2\vite-vue3-admin>npm run dev> vite-vue3-admin1.0.2 dev > vitenode:events:504throw er; // Unhandled error event^Error: spawn D:\PengYe_code\2\vite-vue3-admin\node_modules\vi…...
favicon.ico网站图标不显示问题 Failed to load resource: net::ERR_FILE_NOT_FOU
上述问题主要由于网站的小图标无法显示导致的:可以检查如下部分: 1、是否存在一个favicon.ico文件在根目录下 2、如果存在,看是否写的相对路径:改为绝对路径 <link rel"shortcut icon" href"../favicon.ico&quo…...

微服务·架构组件之服务注册与发现-Nacos
微服务组件架构之服务注册与发现之Nacos Nacos服务注册与发现流程 服务注册:Nacos 客户端会通过发送REST请求的方式向Nacos Server注册自己的服务,提供自身的元数据,比如ip地址、端口等信息。 Nacos Server接收到注册请求后,就会…...

Linux驱动【day2】
mychrdev.c: #include <linux/init.h> #include <linux/module.h> #include <linux/fs.h> #include<linux/uaccess.h> #include<linux/io.h> #include"head.h" unsigned int major; // 保存主设备号 char kbuf[128]{0}; unsigned int…...

4、Nginx 配置实例-反向代理
文章目录 4、nginx 配置实例-反向代理4.1 反向代理实例一4.1.1 实验代码 4.3 反向代理实例二4.3.1 实验代码 【尚硅谷】尚硅谷Nginx教程由浅入深 志不强者智不达;言不信者行不果。 4、nginx 配置实例-反向代理 4.1 反向代理实例一 实现效果:使用 nginx…...

2023年世界机器人大会回顾
1、前记: 本次记录是我自己去世界机器人博览会参观的一些感受,所有回顾为个人感兴趣部分的机器人产品分享。整个参观下来最大的感受就是科学技术、特别是机器人技术和人工智能毫无疑问地、广泛的应用在我们日常生活的方方面面,在安全巡检、特…...

Mac系统 AndroidStudio Missing essential plugin:org.jetbrains.android报错
打开Android Studio,提示 Missing essential plugin:org.jetbrains.android错误,产生的原因是Kotlin被禁用。 解决的方法是删除disabled_plugins.txt,Mac OS对应的路径为: /Users/xzh/Library/Application Support/Google/AndroidStudio202…...

读书笔记:多Transformer的双向编码器表示法(Bert)-1
多Transformer的双向编码器表示法 Bidirectional Encoder Representations from Transformers,即Bert; 本笔记主要是对谷歌Bert架构的入门学习: 介绍Transformer架构,理解编码器和解码器的工作原理;掌握Bert模型架构…...

第二证券:股利支付率和留存收益率的关系?
股利付出率和留存收益率是股票出资中非常重要的目标,它们可以反映公司的盈余才能和未来开展的潜力。那么,二者之间究竟有什么联系呢? 一、股利付出率和留存收益率的定义 股利付出率是指公司向股东分配的股息占当期净利润的比例,通…...

springboot 百货中心供应链管理系统小程序
一、前言 随着我国经济迅速发展,人们对手机的需求越来越大,各种手机软件也都在被广泛应用,但是对于手机进行数据信息管理,对于手机的各种软件也是备受用户的喜爱,百货中心供应链管理系统被用户普遍使用,为方…...

微信小程序之bind和catch
这两个呢,都是绑定事件用的,具体使用有些小区别。 官方文档: 事件冒泡处理不同 bind:绑定的事件会向上冒泡,即触发当前组件的事件后,还会继续触发父组件的相同事件。例如,有一个子视图绑定了b…...

定时器任务——若依源码分析
分析util包下面的工具类schedule utils: ScheduleUtils 是若依中用于与 Quartz 框架交互的工具类,封装了定时任务的 创建、更新、暂停、删除等核心逻辑。 createScheduleJob createScheduleJob 用于将任务注册到 Quartz,先构建任务的 JobD…...

Cloudflare 从 Nginx 到 Pingora:性能、效率与安全的全面升级
在互联网的快速发展中,高性能、高效率和高安全性的网络服务成为了各大互联网基础设施提供商的核心追求。Cloudflare 作为全球领先的互联网安全和基础设施公司,近期做出了一个重大技术决策:弃用长期使用的 Nginx,转而采用其内部开发…...
css的定位(position)详解:相对定位 绝对定位 固定定位
在 CSS 中,元素的定位通过 position 属性控制,共有 5 种定位模式:static(静态定位)、relative(相对定位)、absolute(绝对定位)、fixed(固定定位)和…...

【Java_EE】Spring MVC
目录 Spring Web MVC 编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 编辑参数重命名 RequestParam 编辑编辑传递集合 RequestParam 传递JSON数据 编辑RequestBody …...
Java求职者面试指南:Spring、Spring Boot、MyBatis框架与计算机基础问题解析
Java求职者面试指南:Spring、Spring Boot、MyBatis框架与计算机基础问题解析 一、第一轮提问(基础概念问题) 1. 请解释Spring框架的核心容器是什么?它在Spring中起到什么作用? Spring框架的核心容器是IoC容器&#…...

CVE-2020-17519源码分析与漏洞复现(Flink 任意文件读取)
漏洞概览 漏洞名称:Apache Flink REST API 任意文件读取漏洞CVE编号:CVE-2020-17519CVSS评分:7.5影响版本:Apache Flink 1.11.0、1.11.1、1.11.2修复版本:≥ 1.11.3 或 ≥ 1.12.0漏洞类型:路径遍历&#x…...
AGain DB和倍数增益的关系
我在设置一款索尼CMOS芯片时,Again增益0db变化为6DB,画面的变化只有2倍DN的增益,比如10变为20。 这与dB和线性增益的关系以及传感器处理流程有关。以下是具体原因分析: 1. dB与线性增益的换算关系 6dB对应的理论线性增益应为&…...

算法:模拟
1.替换所有的问号 1576. 替换所有的问号 - 力扣(LeetCode) 遍历字符串:通过外层循环逐一检查每个字符。遇到 ? 时处理: 内层循环遍历小写字母(a 到 z)。对每个字母检查是否满足: 与…...