Flink解决延迟数据问题

总结:

水印:对于迟到数据不长

allowedLateness: 迟到时间很长

侧道输出:对于迟到时间特别长

对于延迟数据的理解:

水印机制(水位线、watermark)机制可以帮助我们在短期延迟下,允许乱序数据的到来。

这个机制很好的处理了那些因为网络等情况短期延迟的数据,让窗口等它们一会儿。

但是水印机制无法长期的等待下去,因为水印机制简单说就是让窗口一直等在那里,等达到水印时间才会触发计算和关闭窗口

这个等待不能一直等,因为会一直缓着数据不计算。

一般水印也就是几秒钟最多几分钟而已(看业务)

那么,在现实世界中,延迟数据除了有短期延迟外,长期延迟也是很常见的。

比如:

l 客户端断网,等了好几个小时才恢复

l 车联网系统进入隧道后没有信号无法上报数据

l 手机欠费没有网

等等,这些场景下数据的迟到就不是简单的网络堵塞造成的几秒延迟了

而是小时、天级别的延迟

对于水印来说,这样的长期延迟数据是无法很好处理的。

那么有没有什么办法去处理这些长期延迟的数据呢?让其可以找到其所属的窗口正常完成计算,哪怕晚了几个小时。

这个场景的解决方式就是:延迟数据处理机制(allowedLateness方法)

水印:乱序数据处理(时间很短的延迟)

延迟处理:长期延迟数据的处理机制。

延迟数据的处理:

waterMark和Window机制解决了流式数据的乱序问题,对于因为延迟而顺序有误的数据,可以根据eventTime进行业务处理,对于延迟的数据Flink也有自己的解决办法,

主要的办法是给定一个允许延迟的时间,在该时间范围内仍可以接受处理延迟数据

设置允许延迟的时间是通过allowedLateness(lateness: Time)设置

保存延迟数据则是通过sideOutputLateData(outputTag: OutputTag[T])保存

获取延迟数据是通过DataStream.getSideOutput(tag: OutputTag[X])获取

1)allowedLateness(lateness: Time)

当我们对流设置窗口后得到的WindowedSteam对象就可以使用allowedLateness方法

该方法传入一个Time值,设置允许的长期延迟(迟到)的时间。

和watermark不同。

未设置allowedLateness(为0),当watermark满足条件,会触发窗口的 执行 + 关闭

当设置了allowedLateness,当watermark满足条件后,只会触发窗口的执行,不会触发窗口关闭

也就是,watermark满足条件后会正常触发窗口计算,将已有的数据完成计算。

但是,不会关闭窗口。如果在allowedLateness允许的时间内仍有这个窗口的数据进来,那么每进来一条,会和已经计算过的(被watermark触发的)数据一起在计算一次

水印:短期延迟,达到条件后触发计算并且关闭窗口(触发+关闭同时进行)

水印+allowedLateness : 短期延迟+ 等待长期延迟效果, 达到水印条件后,会触发窗口计算,但是不关闭窗口。事件时间延迟达到水印+allowedLateness之和后会关闭窗口。

2) 侧道输出-SideOutput

Flink 通过watermark在短时间内允许了乱序到来的数据

通过延迟数据处理机制,可以处理长期迟到的数据。

但总有那么些数据来的晚的太久了。允许迟到1天的设置,它迟到了2天才来。

对于这样的迟到数据,水印无能为力,设置allowedLateness也无能为力,那对于这样的数据Flink就只能任其丢掉了吗?

不会,Flink的两个迟到机制尽量确保了数据不会错过了属于他们的窗口,但是真的迟到太久了,Flink也有一个机制将这些数据收集起来

保存成为一个DataStream,然后,交由开发人员自行处理。

那么这个机制就叫做侧输出机制(Side Output)

侧输出机制:可以将错过水印又错过allowedLateness允许的时间的数据,单独的存放到一个DataStream中,然后开发人员可以自定逻辑对这些超级迟到数据进行处理。

处理主要使用两个方式:

对窗口对象调用sideOutputLateData(OutputTag outputTag)方法,将数据存储到一个地方

对DataStream对象调用getSideOutput(OutputTag outputTag)方法,取出这些被单独处理的数据的DataStream

注意,取到的是一个DataStream,这意味着你可以对这些超级迟到数据继续写 如keyBy, window等处理逻辑。

sideOutputLateData方法:


使用方式:
先定义OutputTag对象(注意,必须new一个匿名内部类形式的OutputTag对象的实例)
然后调用sideOutputLateData方法
// side output   OutputTag对象必须是匿名内部类的形式创建出来, 本质上得到的是OutputTag对象的一个匿名子类
OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("side output"){};
WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> sideOutputLateData =allowedLateness.sideOutputLateData(outputTag);

DataStream.getSideOutput方法:

用法:
DataStream<Tuple2<String, Long>> sideOutput = result.getSideOutput(outputTag);
// 对得到的保存超级迟到数据的DataStream进行处理
sideOutput.print("late>>>");

代码演示

使用Watermark + AllowedLateness + SideOutput ,即使用侧道输出机制来单独收集延迟/迟到/乱序严重的数据,避免数据丢失!

package com.bigdata.day05;import com.bigdata.day04.OrderInfo;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
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;import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Random;
import java.util.UUID;class MyOrderSource2 implements SourceFunction<OrderInfo> {@Overridepublic void run(SourceContext<OrderInfo> ctx) throws Exception {Random random = new Random();while(true){OrderInfo orderInfo = new OrderInfo();orderInfo.setOrderId(UUID.randomUUID().toString().replace("-",""));// 在这个地方可以模拟迟到数据long orderTime = System.currentTimeMillis() - 1000*random.nextInt(100);orderInfo.setOrdertime(orderTime);int money = random.nextInt(10);System.out.println("订单产生的时间:"+ DateFormatUtils.format(orderTime,"yyyy-MM-dd HH:mm:ss")+",金额:"+money);orderInfo.setMoney(money);orderInfo.setUserId(random.nextInt(2));ctx.collect(orderInfo);Thread.sleep(500);}}@Overridepublic void cancel() {}
}
public class Demo01 {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);env.setParallelism(1);// 每隔五秒统计每个用户的前面5秒的订单的总金额//2. source-加载数据DataStreamSource<OrderInfo> streamSource = env.addSource(new MyOrderSource2());//-2.告诉Flink最大允许的延迟时间/乱序时间为多少SingleOutputStreamOperator<OrderInfo> orderDSWithWatermark = streamSource.assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))//-3.告诉Flink哪一列是事件时间.withTimestampAssigner((order, time) -> order.getOrdertime()));OutputTag<OrderInfo> outputTag = new OutputTag<OrderInfo>("side output"){};//3. transformation-数据处理转换SingleOutputStreamOperator<String> result = orderDSWithWatermark.keyBy(orderInfo -> orderInfo.getUserId()).window(TumblingEventTimeWindows.of(Time.seconds(5))).allowedLateness(Time.seconds(4)).sideOutputLateData(outputTag).apply(new WindowFunction<OrderInfo, String, Integer, TimeWindow>() {@Overridepublic void apply(Integer key,  // 代表分组key值     五旬老太守国门TimeWindow window, // 代表窗口对象Iterable<OrderInfo> input, // 分组过之后的数据 [1,1,1,1,1]Collector<String> out  // 用于输出的对象) throws Exception {SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");long start = window.getStart();long end = window.getEnd();int sum = 0;// 专门存放迟到的订单时间for (OrderInfo orderInfo : input) {sum += orderInfo.getMoney();}out.collect(key + ",窗口开始:" + dateFormat.format(new Date(start)) + ",结束时间:" + dateFormat.format(new Date(end)) + "," + sum);//out.collect(key+",窗口开始:"+start+",结束时间:"+end+","+sum);}});result.print("流中的数据,包含迟到的数据:");result.getSideOutput(outputTag).print("严重迟到的数据:");//4. sink-数据输出//5. execute-执行env.execute();}
}

虽然我们添加了延迟的效果,就是说侧道输出,侧道输出不能触发窗口的执行,窗口的执行只能通过水印时间触发 ,允许迟到的数据,不放入到当前窗口中,而是作为一个触发条件看到,它需要放入到它对应的窗口中。

只考虑 1 个并行度的问题

订单发生的真实事件:窗口5秒,间隔5秒,允许迟到 3秒 最晚允许迟到4秒

10:44:00 第一个区间就应该是10:44:00 10:44:05

10:44:01

10:44:02

10:44:03

10:44:04

10:44:05

10:44:07 第一个区间就应该是10:44:05 10:44:10

10:44:22 第一个区间就应该是10:44:20 10:44:25

10:44:30

10:44:28

10:44:20

通过上面这个图可以知道,44:07没有办法触发00~05的执行,但是07不放入00~05区间,而是放入10:44:05 10:44:10

44:22 一个数据触发了两个区间的执行 00~05 05~10

假如有一个订单时44:10产生的,放入哪个区间?应该放入10~15这个区间

 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/62032.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

数据结构(ArrayList顺序表)

一、引言 1.什么是顺序表 定义&#xff1a; 顺序表是一种基于阵列实现的线性表结构&#xff0c;用连续的存储空间保存表中的数据元素&#xff0c;并按顺序排列。 底层依赖阵列&#xff0c;支持随机访问。元素之间没有额外的连接信息&#xff0c;如指针或链表节点。通过动态扩容…

HTTPS的单向认证和双向认证是什么?有什么区别?

目录标题 单向认证流程双向认证流程区别 单向认证流程 HTTPS单向认证‌是指只有服务器向客户端证明其身份。在这种认证方式中&#xff0c;服务器会向客户端发送一个由可信证书颁发机构&#xff08;CA&#xff09;签发的SSL证书&#xff0c;客户端会验证这个证书以确保服务器的…

Python语法基础(一)

&#x1f308;个人主页&#xff1a;羽晨同学 &#x1f4ab;个人格言:“成为自己未来的主人~” 关键字参数和不定长参数 关键字参数主要体现在函数的调用上 使用关键字参数的好处是&#xff0c;可以不按照形参的参数传参 比如说&#xff0c;我们定义一个函数 def c1(nam…

音视频技术扫盲之预测编码的基本原理探究

预测编码是一种数据压缩技术&#xff0c;广泛应用于图像、视频和音频编码等领域。其基本原理是利用数据的相关性&#xff0c;通过对当前数据的预测和实际值与预测值之间的差值进行编码&#xff0c;从而实现数据压缩的目的。 一、预测编码的基本概念 预测编码主要包括预测器和…

Android BottomNavigationView 底部导航栏使用详解

一、BottomNavigationView简介 BottomNavigationView是官方提供可以实现底部导航的组件&#xff0c;最多支持5个item&#xff0c;主要用于功能模块间的切换&#xff0c;默认会包含动画效果。 官方介绍地址&#xff1a;BottomNavigationView 二、使用BottomNavigationView a…

【大数据学习 | Spark-Core】Spark提交及运行流程

spark的集群运行结构 我们要选择第一种使用方式 命令组成结构 spark-submit [选项] jar包 参数 standalone集群能够使用的选项。 --master MASTER_URL #集群地址 --class class_name #jar包中的类 --executor-memory MEM #executor的内存 --executor-cores NUM # executor的…

React中事件处理和合成事件:理解与使用

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…

[241125] X-CMD 发布 v0.4.14:webtop-Linux 容器桌面;支持 PowerShell 环境;轻松搭建并测试蜜罐环境等

目录 X-CMD 发布 v0.4.14&#x1f4c3;Changelog&#x1f427; webtop -- Linux 桌面容器&#x1f5a5;️ pwsh&#x1f4bb; elv|fish|nu|onsh|tcsh&#x1f40b; endlessh&#x1f40b; cowrie&#x1f4f2; mosh&#x1f4bb; mac -- Mac 实用功能&#x1f386; ascii&#…

Jmeter中的测试片段和非测试原件

1&#xff09;测试片段 1--测试片段 功能特点 重用性&#xff1a;将常用的测试元素组合成一个测试片段&#xff0c;便于在多个线程组中重用。模块化&#xff1a;提高测试计划的模块化程度&#xff0c;使测试计划更易于管理和维护。灵活性&#xff1a;可以通过模块控制器灵活地…

linux实时操作系统xenomai看门狗(watchdog)机制及作用介绍

版权声明&#xff1a;本文为本文为博主原创文章&#xff0c;转载请注明出处 https://www.cnblogs.com/wsg1100。如有错误&#xff0c;欢迎指正。 文章目录 一、前言PREEMPT-RT&#xff08;RT Throttling&#xff09; 一、xenomai watchdog介绍二、xenomai watchdog工作原理三、…

【C语言】字符串左旋的三种解题方法详细分析

博客主页&#xff1a; [小ᶻ☡꙳ᵃⁱᵍᶜ꙳] 本文专栏: C语言 文章目录 &#x1f4af;前言&#x1f4af;题目描述&#x1f4af;方法一&#xff1a;逐字符移动法&#x1f4af;方法二&#xff1a;使用辅助空间法&#x1f4af;方法三&#xff1a;三次反转法&#x1f4af;方法对…

【大模型】LLaMA-Factory的环境配置、微调模型与测试

前言 【一些闲扯】 时常和朋友闲聊&#xff0c;时代发展这么快&#xff0c;在时代的洪流下&#xff0c;我们个人能抓住些什么呢。我问了大模型&#xff0c;文心一言是这样回答的&#xff1a; 在快速发展的时代背景下&#xff0c;个人确实面临着诸多挑战&#xff0c;但同时也充满…

Web 表单开发全解析:从基础到高级掌握 HTML 表单设计

文章目录 前言一、什么是 Web 表单?二、表单元素详解总结前言 在现代 Web 开发中,表单 是用户与后端服务交互的重要桥梁。无论是用户登录、注册、搜索,还是提交反馈,表单都无处不在。在本文中,我们将从基础入手,全面解析表单的核心知识点,并通过示例带你轻松掌握表单开…

nodepad配置c/c++ cmd快速打开创建项目文件

前提:下载MinGw,并且配置环境变量 点击阅读次篇文章配置MinGw 无论是哪个编译器&#xff0c;执行c文件都是经历以下步骤: 编译文件生成exe文件执行该exe文件 我们先手动完成这两部 手动编译文件使用指令 gcc {你的c文件} -o {生成文件名}生成exe文件 第二步运行exe直接点击该文…

打造优秀技术文档的三大方向

✅作者简介&#xff1a;2022年博客新星 第八。热爱国学的Java后端开发者&#xff0c;修心和技术同步精进。 &#x1f34e;个人主页&#xff1a;Java Fans的博客 &#x1f34a;个人信条&#xff1a;不迁怒&#xff0c;不贰过。小知识&#xff0c;大智慧。 &#x1f49e;当前专栏…

Xcode15(iOS17.4)打包的项目在 iOS12 系统上启动崩溃

0x00 启动崩溃 崩溃日志&#xff0c;只有 2 行&#xff0c;看不出啥来。 0x01 默认配置 由于我开发时&#xff0c;使用的 Xcode 14.1&#xff0c;打包在另外一台电脑 Xcode 15.3 Xcode 14.1 Build Settings -> Asset Catalog Compliter - Options Xcode 15.3 Build S…

如何使用GCC手动编译stm32程序

如何不使用任何IDE&#xff08;集成开发环境&#xff09;编译stm32程序? 集成开发环境将编辑器、编译器、链接器、调试器等开发工具集成在一个统一的软件中&#xff0c;使得开发人员可以更加简单、高效地完成软件开发过程。如果我们不使用KEIL,IAR等集成开发环境&#xff0c;…

QUICK 调试camera-xml解析

本文主要介绍如何在QUICK QCS6490使能相机模组。QCS6490的相机基于CameraX的框架&#xff0c;只需通过配置XML文件&#xff0c;设置相机模组的相关参数&#xff0c;就可以点亮相机。本文主要介绍Camera Sensor Module XML和Camera Sensor XML配置的解析&#xff0c;这中间需要c…

数据结构 (11)串的基本概念

一、串的定义 1.串是由一个或者多个字符组成的有限序列&#xff0c;一般记为&#xff1a;sa1a2…an&#xff08;n≥0&#xff09;。其中&#xff0c;s是串的名称&#xff0c;用单括号括起来的字符序列是串的值&#xff1b;ai&#xff08;1≤i≤n&#xff09;可以是字母、数字或…

汽车渲染领域:Blender 和 UE5 哪款更适用?两者区别?

在汽车渲染领域&#xff0c;选择合适的工具对于实现高质量的视觉效果至关重要。Blender和UE5&#xff08;Unreal Engine 5&#xff09;作为两大主流3D软件&#xff0c;各自在渲染动画方面有着显著的差异。本文将从核心定位与用途、工作流程、渲染技术和灵活性、后期处理与合成四…