Flink 的时间属性及原理解析

FlinkAPI大体上可以划分为三个层次:处于最底层的ProcessFunction中间一层的DataStream API最上层的SQL/Table API,这三层中的每一层都非常依赖于时间属性。时间在Flink中的地位如下图所示:
[点击并拖拽以移动] ​

时间属性是流处理中最重要的一个方面,是流处理系统的基石之一,贯穿这三层API。在DataStream API这一层中因为封装方面的原因,我们能够接触到时间的地方不是很多,所以我们将重点放在底层的ProcessFunction和最上层的SQL/Table API

Flink 时间语义

不同的应用场景拥有不同的时间语义,Flink作为一个先进的分布式流处理引擎,它本身支持不同的时间语义。其核心是Processing Time(窗口时间即处理时间) 和Event TimeRow Time,事件发生时间),这两类时间主要的不同点如下表所示:

Processing TimeEvent Time
真实世界的时间数据世界的时间
处理数据节点的本地时间记录携带的 Timestamp
处理简单处理复杂
结果不确定(无法重现)结果确定(可重现)

Processing Time是来模拟我们真实世界的时间 ,其实就算是处理数据的节点本地时间也不一定是完完全全的真实世界的时间,所以说它是用来模拟真实世界的时间。而 Event Time是数据世界的时间,即我们要处理的数据流世界里的时间。关于他们的获取方式,Process Time是通过直接去调用本地机器的时间,而Event Time则是根据每一条处理记录所携带的时间戳来判定。
这两种时间在Flink内部的处理以及用户的实际使用方面,难易程度都是不同的。相对而言的Processing Time处理起来更加的简单,而 Event Time要更麻烦一些。而在使用Processing Time的时候,我们得到的处理结果(或者说流处理应用的内部状态)是不确定的。而因为在Flink 内部对Event Time做了各种保障,使用Event Time的情况下,无论重放数据多少次,都能得到一个相对确定可重现的结果。

因此在判断应该使用Processing Time还是Event Time的时候,可以遵循一个原则:当你的应用遇到某些问题要从上一个checkpoint或者 savepoint进行重放,是不是希望结果完全相同。如果希望结果完全相同,就只能用Event Time;如果接受结果不同,则可以用Processing TimeProcessing Time的一个常见的用途是,根据现实时间来统计整个系统的吞吐,比如要计算现实时间一个小时处理了多少条数据,这种情况只能使用Processing Time

时间的特性

Processing Time:递增;
Event Time:一定程度的乱序;

时间的一个重要特性是:时间只能递增,不会来回穿越。 在使用时间的时候我们要充分利用这个特性。假设我们有这么一些记录,然后我们来分别看一下Processing Time还有Event Time对于时间的处理。
【1】对于Processing Time,因为我们是使用的是本地节点的时间(假设这个节点的时钟同步没有问题),我们每一次取到的Processing Time肯定都是递增的,递增就代表着有序,所以说我们相当于拿到的是一个有序的数据流。
【2】而在用Event Time的时候因为时间是绑定在每一条的记录上的,由于网络延迟、程序内部逻辑、或者其他一些分布式系统的原因,数据的时间可能会存在一定程度的乱序,比如下图的例子。在Event Time场景下,我们把每一个记录所包含的时间称作Record Timestamp。如果Record Timestamp所得到的时间序列存在乱序,我们就需要去处理这种情况。
[点击并拖拽以移动] ​

如果单条数据之间是乱序,我们就考虑对于整个序列进行更大程度的离散化。简单地讲,就是把数据按照一定的条数组成一些小批次,但这里的小批次并不是攒够多少条就要去处理,而是为了对他们进行时间上的划分。经过这种更高层次的离散化之后,我们会发现最右边方框里的时间就是一定会小于中间方框里的时间,中间框里的时间也一定会小于最左边方框里的时间。
[点击并拖拽以移动] ​

这个时候我们在整个时间序列里插入一些类似于标志位的特殊的处理数据,这些特殊的处理数据叫做watermark。一个watermark本质上就是一个timestamp数值,表示后到来的数据再也没有小于或等于这个时间watermark的了

Timestamp 和 Watermark 行为概览

[点击并拖拽以移动] ​

接下来我们重点看一下Event Time里的Record Timestamp(简写成timestamp)和watermark的一些基本信息。绝大多数的分布式流计算引擎对于数据都是进行了 DAG 图的抽象,它有自己的数据源,有处理算子,还有一些数据汇。数据在不同的逻辑算子之间进行流动。watermarktimestamp有自己的生命周期,主要分为watermarktimestamp的产生、他们在不同的节点之间的传播、以及在每一个节点上的处理

Timestamp分配和Watermark生成: Flink支持两种watermark生成方式。第一种是在SourceFunction中产生,相当于把整个的timestamp分配和watermark生成的逻辑放在流处理应用的源头。我们可以在SourceFunction里面通过这两个方法产生watermark
【1】通过collectWithTimestamp方法发送一条数据,其中第一个参数就是我们要发送的数据,第二个参数就是这个数据所对应的时间戳;也可以调用emitWatermark去产生一条watermark,表示接下来不会再有时间戳小于等于这个数值记录。
【2】另外,有时候我们不想在SourceFunction里生成timestamp或者watermark,或者说使用的SourceFunction本身不支持,我们还可以在使用DataStreamAPI的时候指定,调用的DataStream.assignTimestampsAndWatermarks这个方法,能够接收不同的timestampwatermark的生成器。

总体上而言生成器可以分为两类: 第一类是定期生成器;第二类是根据一些在流处理数据流中遇到的一些特殊记录生成的。

定期生成根据特殊记录生成
现实时间驱动数据驱动
没个一段时间调用生成方法分一次分配Timestamp都会调用生成的方法
实现AssignerWithPeriodicWatermarks实现 AssignerWithPunctuatedWatermarks

两者的区别主要有三个方面,首先定期生成是现实时间驱动的,这里的定期生成主要是指watermark(因为timestamp是每一条数据都需要有的),即定期会调用生成逻辑去产生一个watermark。而根据特殊记录生成是数据驱动的,即是否生成watermark不是由现实时间来决定,而是当看到一些特殊的记录就表示接下来可能不会有符合条件的数据再发过来了,这个时候相当于每一次分配Timestamp之后都会调用用户实现的watermark生成方法,用户需要在生成方法中去实现watermark的生成逻辑。

在分配timestamp和生成watermark的过程中,虽然在SourceFunctionDataStream中都可以指定,但是还是建议生成的工作越靠近 DataSource越好。这样会方便让程序逻辑里面更多的 operator 去判断某些数据是否乱序。Flink 内部提供了很好的机制去保证这些timestampwatermark被正确地传递到下游的节点。

Watermark 传播

具体的传播策略基本上遵循这三点:
【1】watermark会以广播的形式在算子之间进行传播。比如说上游的算子连接了三个下游的任务,它会把自己当前的收到的watermark以广播的形式传到下游。

广播特点: 主机之间“一对所有”的通讯模式,网络对其中每一台主机发出的信号都进行无条件复制并转发,所有主机都可以接收到所有信息(不管你是否需要)

【2】如果在程序里面收到了一个Long.MAX_VALUE这个数值的watermark,就表示对应的那一条流的一个部分不会再有数据发过来了,它相当于就是一个终止的标志。
【3】对于单流而言,这个策略比较好理解,而对于有多个输入的算子,watermark的计算就有讲究了,一个原则是:单输入取其大,多输入取小

[点击并拖拽以移动] ​

举个例子,上图蓝色代表一个算子的一个任务,然后它有三个输入,分别是W1W2W3,这三个输入可以理解成任何输入,这三个输入可能是属于同一个流,也可能是属于不同的流。然后在计算watermark的时候,对于单个输入而言是取他们的最大值,因为我们都知道 watermark应该遵循一个单调递增的一个原则。对于多输入,它要统计整个算子任务的watermark时,就会取这三个计算出来的watermark的最小值。即一个多个输入的任务,它的watermark受制于最慢的那条输入流。这一点类似于木桶效应,整个木桶中装的水会受制于最矮的那块板。

watermark在传播的时候有一个特点是,它的传播是幂等的。多次收到相同的watermark,甚至收到之前的watermark都不会对最后的数值产生影响,因为对于单个输入永远是取最大的,而对于整个任务永远是取一个最小的。同时我们可以注意到这种设计其实有一个局限,具体体现在它没有区分你这个输入是一条流多个partition还是来自于不同的逻辑上的流的JOIN。对于同一个流的不同partition,我们对他做这种强制的时钟同步是没有问题的,因为一开始就把一条流拆散成不同的部分,但每一个部分之间共享相同的时钟。但是如果算子的任务是在做类似于JOIN操作,那么要求两个输入的时钟强制同步其实没有什么道理的,因为完全有可能是把一条离现在时间很近的数据流和一个离当前时间很远的数据流进行JOIN,这个时候对于快的那条流,因为它要等慢的那条流,所以说它可能就要在状态中去缓存非常多的数据,这对于整个集群来说是一个很大的性能开销。

ProcessFunction

在了解watermark的处理之前,先简单了解ProcessFunction,因为watermark在任务里的处理逻辑分为内部逻辑外部逻辑。外部逻辑其实就是通过ProcessFunction来体现的,需要使用 Flink提供的时间相关的API的话就只能写在ProcessFunction里。

ProcessFunction和时间相关的功能主要有三点:
【1】根据你当前系统使用的时间语义不同,你可以去获取当前你正在处理这条记录的Record Timestamp,或者当前的Processing Time
【2】它可以获取当前算子的时间,可以把它理解成当前的watermark
【3】为了在 ProcessFunction 中去实现一些相对复杂的功能,允许注册一些timer(定时器)。比如说在watermark达到某一个时间点的时候就触发定时器,所有的这些回调逻辑也都是由用户来提供,涉及到如下三个方法,registerEventTimeTimerregisterProcessingTimeTimeronTimer。在onTimer方法中就需要去实现自己的回调逻辑,当条件满足时回调逻辑就会被触发。

一个简单的应用是,我们在做一些时间相关的处理的时候,可能需要缓存一部分数据,但这些数据不能一直去缓存下去,所以需要有一些过期的机制,我们可以通过timer去设定这么一个时间,指定某一些数据可能在将来的某一个时间点过期,从而把它从状态里删除掉。所有的这些和时间相关的逻辑在Flink内部都是由自己的Time Service(时间服务)完成的。

Watermark 处理

一个算子的实例在收到watermark的时候,首先要更新当前的算子时间,这样的话在ProcessFunction里方法查询这个算子时间的时候,就能获取到最新的时间。第二步它会遍历计时器队列,这个计时器队列就是我们刚刚说到的timer,你可以同时注册很多timerFlink会把这些Timer按照触发时间放到一个优先队列中。第三步Flink得到一个时间之后就会遍历计时器的队列,然后逐一触发用户的回调逻辑。通过这种方式,Flink的某一个任务就会将当前的watermark发送到下游的其他任务实例上,从而完成整个watermark的传播,从而形成一个闭环。
[点击并拖拽以移动] ​

Table API 中的时间

下面看一看Table/SQL API中的时间。为了让时间参与到Table/SQL这一层的运算中,我们需提前把时间属性放到表的schema中,这样的话我们才能够在SQL语句或者Table的逻辑表达式里面使用时间去完成需求。

Table中指定时间列: 其实之前社区就怎么在Table/SQL中去使用时间这个问题做过一定的讨论,是把获取当前Processing Time的方法是作为一个特殊的UDF,还是把这一个列物化到整个的schema里面,最终采用了后者。我们这里就分开来讲一讲Processing TimeEvent Time在使用的时候怎么在Table中指定。

从DataStream转化通过TableSource 转化
Processing TimetEnv.fromDataStream(stream,“f1,f2,f3.proctime”)TableSource实现DefinedProctimeAttributes接口
Event Time原始 DataStream 必须有 timestamp 及 watermark数据中存在类型为 long或 timestamp的时间字段
Event TimetEnv.fromDataStream(stream,“f1,f2,f3.rowtime”) tEnv.fromDataStream(stream,“f1.rowtime,f2,f3”)TableSource实现DefinedProctimeAttributes接口

对于Processing Time,我们知道要得到一个Table对象(或者注册一个Table)有两种手段:
● 可以从一个DataStream转化成一个Table
● 直接通过TableSource去生成这么一个Table
对于第一种方法而言,我们只需要在你已有的这些列中(例子中f1f2就是两个已有的列),在最后用“列名.proctime”这种写法就可以把最后的这一列注册为一个Processing Time,以后在写查询的时候就可以去直接使用这一列。如果Table是通过TableSource生成的,就可以通过实现这一个DefinedRowtimeAttributes接口,然后就会自动根据你提供的逻辑去生成对应的Processing Time
相对而言,在使用EventTime时则有一个限制,因为EventTime不像Processing Time那样是随拿随用。如果要从DataStream去转化得到一个Table,必须要提前保证原始的DataStream里面已经存在了RecordTimestampwatermark。如果想通过TableSource生成的,也一定要保证要接入的数据里面存在一个类型为long或者timestamp的时间字段。具体来说,如果要从DataStream去注册一个表,和proctime类似,只需要加上“列名.rowtime”就可以。需要注意,如果要用Processing Time,必须保证要新加的字段是整个schema中的最后一个字段,而Event Time的时候其实可以去替换某一个已有的列,然后Flink会自动的把这一列转化成需要的rowtime这个类型。

如果是通过TableSource生成的,只需要实现DefinedRowtimeAttributes接口就可以了。需要说明的一点是,在DataStream API这一侧其实不支持同时存在多个Event Time(rowtime),但是在Table这一层理论上可以同时存在多个rowtime。因为DefinedRowtimeAttributes接口的返回值是一个对于rowtime描述的List,即其实可以同时存在多个rowtime列,在将来可能会进行一些其他的改进,或者基于去做一些相应的优化。

时间列和 Table 操作

指定完了时间列之后,当我们要真正去查询时就会涉及到一些具体的操作。这里我列举的这些操作都是和时间列紧密相关,或者说必须在这个时间列上才能进行的。比如说 Over窗口聚合Group by窗口聚合 这两种窗口聚合,在写SQL提供参数的时候只能允许你在这个时间列上进行这种聚合。第三个就是时间窗口聚合,你在写条件的时候只支持对应的时间列。最后就是排序,我们知道在一个无尽的数据流上对数据做排序几乎是不可能的事情,但因为这个数据本身到来的顺序已经是按照时间属性来进行排序,所以说如果要对一个 DataStream转化成Table进行排序的话,只能是按照时间列进行排序,当然同时也可以指定一些其他的列,但是时间列这个是必须的,并且必须放在第一位。
[点击并拖拽以移动] ​

为什么说这些操作只能在时间列上进行?
因为我们有的时候可以把到来的数据流就看成是一张按照时间排列好的一张表,而我们任何对于表的操作,其实都是必须在对它进行一次顺序扫描的前提下完成的。大家都知道数据流的特性之一就是一过性,某一条数据处理过去之后,将来其实不太好去访问它。当然因为 Flink中内部提供了一些状态机制,我们可以在一定程度上去弱化这个特性,但是最终还是不能超越的,限制状态不能太大。所有这些操作为什么只能在时间列上进行,因为这个时间列能够保证我们内部产生的状态不会无限的增长下去,这是一个最终的前提。

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

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

相关文章

python flask图书管理系统带文档

python flask图书管理系统带文档。功能:登录,图书的增删改查,读者管理,借阅记录,有文档。 技术:python3,flask,mysql,html。 包含源码数据库文件文档。 源码下载地址: https://download.csd…

芯课堂 | MCU之TIMER精准延时

引言 华芯微特公司SWM系列单片机提供的TIMER个数和功能有些微差别,为了让您更加简单的使用这一功能,下面小编将以SWM190为例,给大家展示如何使用SWM系列产品的TIMER功能。 TIMER精准延时 一、TIMER简介 TIMER是一种定时器工具,…

实例:NodeJS 操作 Kafka

本人是C#出身的程序员,c#很简单就能实现,有需要的可以加我私聊。但是就目前流行的开发语言,尤其是面向web方向应用的,我感觉就是Nodejs最简单了。下面介绍: 本文将会介绍在windows环境下启动Kafka,并通过n…

滑动窗口最大值(力扣239题)

单调递减队列: 在解决题目之前,我们先来了解一下单调递减队列,它其实就是在队列的基础上多加了一些限制,如下图: 要求队列中的元素必须按从大到小的顺序排列。 如果向单调递减队列中加入数字 1,可以直接加入…

一起玩儿物联网人工智能小车(ESP32)——25. 利用超声波传感器测量距离

摘要:本文介绍如何利用超声波传感器测量障碍物的距离 测量距离是智能小车经常要用到的功能,今天就来介绍一个最常用的测量距离的传感器——超声波传感器。 超声波传感器的测距原理是利用超声波发射器向某个方向发射超声波,与此同时&#xff…

【Emgu.CV教程】第22篇 、色彩处理之ApplyColorMap()伪色彩应用

这篇文章讲的内容比较轻松,技术含量比较低。从我个人的角度讲,ApplyColorMap()函数实现了类似PhotoShop的一些酷炫效果,既把原始彩色图转换为21种风格各异的彩色图像,比如秋天风格、热力图风格等等,但是,在…

【Java期末】学生成绩管理系统

诚接计算机专业编程任务(C语言、C、Python、Java、HTML、JavaScript、Vue等)10/15R,如有需要请私信我,或者加我的企鹅号:1404293476 本文资源下载地址:https://download.csdn.net/download/weixin_47040861/88697244 —————…

Win11/10家庭版升专业版/企业版

2024.1.4 下载地址:HEU_KMS_Activator GitHubhttps://github.com/zbezj/HEU_KMS_Activator/tags 下载最新版即可 (升级时需要断网)打开程序 --> 其他 --> 选择professional --> 在点击四叶草图标即可 升级完后再次打开软件 -->…

Kali Linux实现UEFI和传统BIOS(Legacy)引导启动

默认Kali linux安装会根据当前启动的引导模式进行安装 例:以UEFI引导启动安装程序,安装后仅能在UEFI引导模式下进入系统 安装Kali系统 这边基于VirtualBox虚拟机镜像实战操作 首先创建一个Kali虚拟机 这里需要注意,把启动 EFI (只针对某些操…

Vue v-html中内容图片过大自适应处理

之前图片如下&#xff0c;图片已经超出了页面的展示范围 对v-html增加样式处理 <div class"body padding-l scroll " v-html"docData.content"> </div><style scoped>.body >>> img {max-width: 100% ;} </style>…

CCNP课程实验-04-BGP_CFG

目录 实验条件网络拓朴 基础配置需求实现IGP部分1. 按照图示配置OSPF区域&#xff0c;RID为Loopback 0地址。其中Area 146要配置为OSPF的特殊区域。2. 配置其它路由协议&#xff0c;重分布使得路由互相注入&#xff0c;实现全网互通。3. R1配置策略路由&#xff0c;使得R2经R1去…

Vue 框架前导:详解 Ajax

Ajax Ajax 是异步的 JavaScript 和 XML。简单来说就是使用 XMLHttpRequest 对象和服务器通信。可以使用 JSON、XML、HTML 和 text 文本格式来发送和接收数据。具有异步的特性&#xff0c;可在不刷新页面的情况下实现和服务器的通信&#xff0c;交换数据或者更新页面 01. 体验 A…

18、BLIP

简介 github BLIP提出了一种基于预训练的方法&#xff0c;通过联合训练视觉和语言模型来提升多模态任务的性能。 BLIP(Bootstrapping Language-Image Pretraining)是salesforce在2022年提出的多模态框架&#xff0c;是理解和生成的统一&#xff0c;引入了跨模态的编码器和解码…

词嵌入位置编码的实现(基于pytorch)

背景介绍 在transformers架构当中&#xff0c;对于词向量的输入需要加上原本词对应的位置信息&#xff0c;作为输入到模型中训练的input&#xff0c;那具体的位置编码如何实现呢&#xff1f;本篇博客就跟大家一起分享一下对应的步骤 位置编码的公式 对于词向量的位置编码的方…

西电期末1017.有序序列插值

一.题目 二.分析与思路 简单题。主要考察简单的排序&#xff0c;最后的插入数据同样不用具体实现&#xff0c;只需在输出时多输出一下即可&#xff0c;注意顺序&#xff01;&#xff01; 三.代码实现 #include<bits/stdc.h>//万能头 int main() {int n;scanf("%d…

HTML5+CSS3+Vue小实例:彩色圆环溶解加载动画

实例:彩色圆环溶解加载动画 技术栈:HTML+CSS+Vue.js 效果: 源码: 【HTML】 <!DOCTYPE html> <html lang="zh-CN"> <head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge&…

【已解决】Invalid bound statement (not found)

报错讯息 org.apache.ibatis.binding.BindingException: Invalid bound statement (not found): com.casey.mapper.SysRoleMapper.getUserRoleCode at org.apache.ibatis.binding.MapperMethod S q l C o m m a n d . < i n i t > ( M a p p e r M e t h o d . j a v a :…

Springcloud 微服务实战笔记 Ribbon

使用 Configurationpublic class CustomConfiguration {BeanLoadBalanced // 开启负载均衡能力public RestTemplate restTemplate() {return new RestTemplate();}}可看到使用Ribbon&#xff0c;非常简单&#xff0c;只需将LoadBalanced注解加在RestTemplate的Bean上&#xff0…

视频通话录制 方案 教程

一些基本概要 1 市面流行的音视频产品&#xff0c;都包含一对一、一对多、多对多通话&#xff0c;云端录制等这些基本功能&#xff0c;有些厂商支持本地服务录制。 2 本地服务录制&#xff0c;云端录制区别 两个录制区别在于&#xff0c;本地服务端录制的 SDK 要求部署在 …

【Proteus仿真】【Arduino单片机】电蒸锅温度控制系统

文章目录 一、功能简介二、软件设计三、实验现象联系作者 一、功能简介 本项目使用Proteus8仿真Arduino单片机控制器&#xff0c;使用LCD1602液晶、按键开关、蜂鸣器、DS18B20温度传感器&#xff0c;液位传感器、继电器控制加热保温装置等。 主要功能&#xff1a; 系统运行后&…