【Flink系列四】Window及Watermark

3.1、window

在 Flink 中 Window 可以将无限流切分成有限流，是处理有限流的核心组件，现在 Flink 中 Window 可以是时间驱动的（Time Window），也可以是数据驱动的（Count Window）。

Flink中的窗口可以分成：滚动窗口(Tumbling Window,无重叠)，滑动窗口(Sliding Window,可能有重叠)，会话窗口(Session Window，活动间隙)，全局窗口(Gobal Window)

3.1.1、Tumbling Windows 滚动窗口

滚动窗口的assigner分发元素到指定大小的窗口。滚动窗口的大小是固定的，且各自范围之间不重叠。

// 滚动event-time窗口
input.keyBy(<key selector>).window(TumblingEventTimeWindows.of(Time.seconds(5))).<windowed transformation>(<window function>);// 滚动processing-time窗口
input.keyBy(<key selector>).window(TumblingProcessingTimeWindows.of(Time.second(5))).<windowed transformation>(<window function>);// 长度为一天的滚动event-time窗口， 偏移量为-8小时
input.keyBy(<key selector>).window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))).<windowed transformation>(<window function>);

如上一个例子所示，滚动窗口的 assigners 也可以传入可选的 offset 参数。这个参数可以用来对齐窗口。 比如说，不设置 offset 时，长度为一小时的滚动窗口会与 linux 的 epoch 对齐。 你会得到如 1:00:00.000 - 1:59:59.999、2:00:00.000 - 2:59:59.999 等。 如果你想改变对齐方式，你可以设置一个 offset。如果设置了 15 分钟的 offset， 你会得到 1:15:00.000 - 2:14:59.999、2:15:00.000 - 3:14:59.999 等。 一个重要的 offset 用例是根据 UTC-0 调整窗口的时差。比如说，在中国你可能会设置 offset 为 Time.hours(-8)。

3.1.2、Sliding Windows滑动窗口

滑动窗口的assigner 分发元素到指定大小的窗口，窗口大小通过 window size 参数设置。 滑动窗口需要一个额外的滑动距离（滑动步长window slide）参数来控制生成新窗口的频率。 因此，如果 slide 小于窗口大小，滑动窗口可以允许窗口重叠。这种情况下，一个元素可能会被分发到多个窗口。

// 滑动 event-time 窗口
input.keyBy(<key selector>).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>);// 滑动 processing-time 窗口
input.keyBy(<key selector>).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>);// 滑动 processing-time 窗口，偏移量为 -8 小时
input.keyBy(<key selector>).window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8))).<windowed transformation>(<window function>);

3.1.3、Session Windows 会话窗口

会话窗口的 assigner 会把数据按活跃的会话分组。 与滚动窗口和滑动窗口不同，会话窗口不会相互重叠，且没有固定的开始或结束时间。 会话窗口在一段时间没有收到数据之后会关闭，即在一段不活跃的间隔之后。 会话窗口的 assigner 可以设置固定的会话间隔（session gap）或 用 session gap extractor 函数来动态地定义多长时间算作不活跃。 当超出了不活跃的时间段，当前的会话就会关闭，并且将接下来的数据分发到新的会话窗口。

// 设置了固定间隔的event-time会话窗口
input.keyBy(<key selector>).window(EventTimeSessionWindows.withGap(Time.minutes(10))).<windowed transformation>(<window function>);// 设置了动态间隔的event-time会话窗口
input.keyBy(<key selector>).window(EventTimeSessionWindows.withDynamicGap((element)-> {// 决定并返回会话间隔})).<windowed transformation>(<window function>);// 设置了固定间隔的 processing-time session 窗口
input.keyBy(<key selector>).window(ProcessingTimeSessionWindows.withGap(Time.minutes(10))).<windowed transformation>(<window function>);// 设置了动态间隔的 processing-time 会话窗口
input.keyBy(<key selector>).window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {// 决定并返回会话间隔}))

3.1.4、Global Windows 全局窗口

全局窗口的 assigner 将拥有相同 key 的所有数据分发到一个全局窗口。 这样的窗口模式仅在你指定了自定义的 trigger 时有用。 否则，计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据。

input.keyBy(<key selector>).window(GlobalWindows.create()).<windowed transformation>(<window function>);
3.1.5、Triggers窗口触发

Trigger决定了一个窗口(由window assigner定义)何时可以被window function处理。一般来说，watermark的时间戳>=window endTime并且在窗口内有数据，就会触发窗口的计算。每个WindowAssigner都有一个默认的Trigger。如果默认trigger无法满足需求，可以在trigger(...)调用中指定自定义的trigger。

  • onElement() 每次往 window 增加一个元素的时候都会触发
  • onEventTime() 当 event-time timer 被触发的时候会调用
  • onProcessingTime() 当 processing-time timer 被触发的时候会调用
  • onMerge() 对两个 trigger 的 state 进行 merge 操作
  • clear() window 销毁的时候被调用

上面的接口中前三个会返回一个 TriggerResult，TriggerResult 有如下几种可能的选择：

  • CONTINUE 不做任何事情
  • FIRE 触发 window
  • PURGE 清空整个 window 的元素并销毁窗口
  • FIRE_AND_PURGE 触发窗口，然后销毁窗口

3.2、time和watermark

3.2.1、time

在 Flink 中 Time 可以分为三种Event-Time，Processing-Time 以及 Ingestion-Time，三者的关系我们可以从下图中得知：

3.2.2、watermark

Flink提出了watermark，专门处理EventTime窗口计算，其本质其实就是一个时间戳。因为对于迟到数据late element，不可能一直无限期等待，必须有一个机制来保证一个特定的时间后，必须取触发window去进行计算，这种机制就是watermark

watermark本质上也是一种时间戳，由Apache Flink Source或者自定义的Watermark生成器按照需求Punctuated或者Periodic两种方式生成的一种系统Event，与普通数据流Event一样流转到对应的下游算子，接收到Watermark Event的算子以此不断调整自己管理的EventTime clock。 Apache Flink 框架保证Watermark单调递增，算子接收到一个Watermark时候，框架知道不会再有任何小于该Watermark的时间戳的数据元素到来了，所以Watermark可以看做是告诉Apache Flink框架数据流已经处理到什么位置(时间维度)的方式。 Watermark的产生和Apache Flink内部处理逻辑如下图所示: 

目前Apache Flink 有两种生产Watermark的方式，如下：

  • Punctuated - 数据流中每一个递增的EventTime都会产生一个Watermark。 在实际的生产中Punctuated方式在TPS很高的场景下会产生大量的Watermark在一定程度上对下游算子造成压力，所以只有在实时性要求非常高的场景才会选择Punctuated的方式进行Watermark的生成。
  • Periodic - 周期性的（一定时间间隔或者达到一定的记录条数）产生一个Watermark。在实际的生产中Periodic的方式必须结合时间和积累条数两个维度继续周期性产生Watermark，否则在极端情况下会有很大的延时。

参阅:Apache Flink 漫谈系列(03) - Watermark-阿里云开发者社区

我们可以考虑一个这样的例子：某 App 会记录用户的所有点击行为，并回传日志（在网络不好的情况下，先保存在本地，延后回传）。A 用户在 11:02 对 App 进行操作，B 用户在 11:03 操作了 App，但是 A 用户的网络不太稳定，回传日志延迟了，导致我们在服务端先接受到 B 用户 11:03 的消息，然后再接受到 A 用户 11:02 的消息，消息乱序了。那我们怎么保证基于 event-time 的窗口在销毁的时候，已经处理完了所有的数据呢？这就是 watermark 的功能所在。watermark 会携带一个单调递增的时间戳 t，watermark(t) 表示所有时间戳不大于 t 的数据都已经到来了，未来小于等于t的数据不会再来，因此可以放心地触发和销毁窗口了。下图中给了一个乱序数据流中的 watermark 例子

3.2.3、迟到的数据

上面的 watermark 让我们能够应对乱序的数据，但是真实世界中我们没法得到一个完美的 watermark 数值 — 要么没法获取到，要么耗费太大，因此实际工作中我们会使用近似 watermark — 生成 watermark(t) 之后，还有较小的概率接受到时间戳 t 之前的数据，在 Flink 中将这些数据定义为 “late elements”, 同样我们可以在 window 中指定是允许延迟的最大时间（默认为 0），可以使用下面的代码进行设置

设置allowedLateness之后，迟来的数据同样可以触发窗口，进行输出，利用 Flink 的 side output 机制，我们可以获取到这些迟到的数据，使用方式如下：

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

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

相关文章

Python----多态

Python----多态

1、什么是多态 多态指的是一类事物有多种形态。 定义&#xff1a;多态是一种使用对象的方式&#xff0c;子类重写父类方法&#xff0c;调用不同子类对象的相同父类方法&#xff0c;可以产生不同的执行结果。 ① 多态依赖继承 ② 子类方法必须要重写父类方法 首先定义一个父类…
阅读更多...
VC++使用GetProcessTimes获取进程创建时间、销毁时间、用户态时间、内核态时间

VC++使用GetProcessTimes获取进程创建时间、销毁时间、用户态时间、内核态时间

一、GetProcessTimes函数简介&#xff08;微软MSDN&#xff09; 微软提供了一个非常有用的API函数GetProcessTimes用来获取进程创建时间、销毁时间、用户态时间、内核态时间&#xff0c;msdn连接为&#xff1a;GetProcessTimes 函数 (processthreadsapi.h) 其函数原型为&#…
阅读更多...
【计算机组成体系结构】SRAM和DRAM

【计算机组成体系结构】SRAM和DRAM

RAM — Random Access Memory 随机访问存储器 —指定某一存储单元地址的时候&#xff0c;存储单元的读取速度并不会因为存储单元的物理位置改变 SRAM即为 Static RAM 静态随机访问存储器 — 用于主存DRAM即为 Dynamic RAM 动态随机访问存储器 — 用于Cache 一、SRAM和DRAM的特…
阅读更多...
智能外呼常见场景有哪些？

智能外呼常见场景有哪些？

智能外呼常见场景是什么&#xff1f; 智能外呼在各种场景下都有应用&#xff0c;以下是一些常见的场景&#xff1a; 营销推广 通过智能外呼向潜在客户进行产品或服务的宣传和推广&#xff0c;收集客户对产品或服务的反馈。根据客户的反馈自动调整宣传策略&#xff0c;从而提…
阅读更多...
JavaSE基础50题：15. 斐波那契数列（迭代实现）

JavaSE基础50题：15. 斐波那契数列（迭代实现）

概述 求斐波那契数列的第n项&#xff08;迭代实现&#xff09;。 斐波那契&#xff1a;1 1 2 3 5 8…… 代码 public class P15 {public static int fib(int n) {if (n 1 || n 2) {return 1;}int f1 1;int f2 1;int f3 0;for (int i 3; i < n; i) {f3 f1 f2;f1 …
阅读更多...
Java TCP（一对一）聊天简易版

Java TCP（一对一）聊天简易版

客户端 import java.io.*; import java.net.Socket; import java.util.Date; import javax.swing.*;public class MyClient {private JFrame jf;private JButton jBsend;private JTextArea jTAcontent;private JTextField jText;private JLabel JLcontent;private Date data;p…
阅读更多...
字符选择的题解

字符选择的题解

目录 原题描述&#xff1a; 题目描述 样例输入1 样例输出1 样例输入2 样例输出2 题目大意&#xff1a; 主要思路&#xff1a; change的设计&#xff1a; dp的转移&#xff1a; dp初始化&#xff1a; dp的结算&#xff1a; 注意事项&#xff1a; 代码&#xff08;有…
阅读更多...
使用智能AI文心一言处理采集数据

使用智能AI文心一言处理采集数据

简数采集器支持调用百度智能AI文心一言大模型API接口&#xff0c;可对采集的数据进行研究分析&#xff0c;内容创作。 文心一言API使用方法如下&#xff1a; 目录 1. 采集数据 2. 申请API 3. 对接文心一言API 4. 设置文心一言API的执行指令 5. 使用文心一言API处理采集数…
阅读更多...
Java 第21章 网络通信

Java 第21章 网络通信

网络程序设计基础 网络程序设计编写的是与其他计算机进行通信的程序。Java 已经将网络程序所需要的元素封装成不同的类&#xff0c;用户只要创建这些类的对象&#xff0c;使用相应的方法&#xff0c;即使不具备有关的网络支持&#xff0c;也可以编写出高质量的网络通信程序。 …
阅读更多...
OpenEuler_22.03升级mongdb到7.0.4

OpenEuler_22.03升级mongdb到7.0.4

使用命令&#xff1a;lscpu&#xff0c;查看cpu架构为aarch64为arm架构的一种执行状态。 所以我们直接下载arm的包安装即可。无需自己编译源码。 下载地址&#xff1a;https://www.mongodb.com/try/download/community 下载解压 wget https://fastdl.mongodb.org/linux/mong…
阅读更多...
高精度时钟芯片SD2405

高精度时钟芯片SD2405

概要 SD2405是一款非常优秀的RTC解决方案&#xff0c;为了能让用户在Arduino上有一款方便易用的时钟模块。该模块是一款内置晶振&#xff0c;支持IIC串行接口的高精度时钟模块&#xff1b;内置一次性工业级电池&#xff0c;可保证外部掉电的情况下&#xff0c;可以继续工作5~8…
阅读更多...
c语言词法分析器

c语言词法分析器

词法分析器&#xff08;也称为词法解析器或词法扫描器&#xff09;是编译器的一个组成部分&#xff0c;它的任务是将输入的源代码&#xff08;字符流&#xff09;分解成称为“标记”的序列&#xff0c;其中每个标记对应于源代码中的一个单词或符号。 以下是一个简单的C语言词法…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023