大数据-玩转数据-Flink 水印
一、Flink 中的水印
在Flink的流式操作中, 会涉及不同的时间概念:
1.1 处理时间
是指的执行操作的各个设备的时间,对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟。比如, 一个长度为1个小时的窗口将会包含设备时钟表示的1个小时内所有的数据。 假设应用程序在 9:15am分启动, 第1个小时窗口将会包含9:15am到10:00am所有的数据,然后下个窗口是10:00am-11:00am, 等等。处理时间是最简单时间语义, 数据流和设备之间不需要做任何的协调。他提供了最好的性能和最低的延迟。 但是, 在分布式和异步的环境下,处理时间没有办法保证确定性,容易受到数据传递速度的影响: 事件的延迟和乱序。在使用窗口的时候, 如果使用处理时间, 就指定时间分配器为处理时间分配器。
1.2 事件时间
是指的这个事件发生的时间。在event进入Flink之前, 通常被嵌入到了event中, 一般作为这个event的时间戳存在。在事件时间体系中, 时间的进度依赖于数据本身,和任何设备的时间无关。事件时间程序必须制定如何产生Event Time Watermarks(水印) 。假设所有数据都已到达,事件时间操作将按预期方式运行,即使在处理无序或迟到的事件或重新处理历史数据时,也会产生正确且一致的结果。例如,每小时事件时间窗口将包含带有事件时间戳的所有记录,这些记录落入该小时。在使用窗口的时候, 如果使用事件时间, 就指定时间分配器为事件时间分配器。从1.12开始, Flink内部已经把默认的语义改成了事件时间。
1.3 Flink中的WaterMark
支持event time的流式处理框架需要一种能够测量event time 进度的方式。 比如, 一个窗口算子创建了一个长度为1小时的窗口,那么这个算子需要知道事件时间已经到达了这个窗口的关闭时间,从而在程序中去关闭这个窗口。事件时间可以不依赖处理时间来表示时间的进度。例如,在程序中, 即使处理时间和事件时间有相同的速度, 事件时间可能会轻微的落后处理时间。另外一方面使用事件时间可以在几秒内处理已经缓存在Kafka中多周的数据,这些数据可以照样被正确处理, 就像实时发生的一样能够进入正确的窗口。这种在Flink中去测量事件时间的进度的机制就是watermark(水印)。
1.4 Flink中如何产生水印

二、代码集成
package com.lyh.flink08;import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
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 javax.naming.Context;
import java.time.Duration;public class WatorMark_01 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> stream = env.socketTextStream("hadoop100", 9999).map(new MapFunction<String, WaterSensor>() {@Overridepublic WaterSensor map(String value) throws Exception {String[] datas = value.split(",");return new WaterSensor(datas[0],Long.valueOf(datas[1]),Integer.valueOf(datas[2]));}});WatermarkStrategy<WaterSensor> wms = WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() { // 指定时间戳@Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {return element.getTs() * 1000;}});stream.assignTimestampsAndWatermarks(wms) // 指定水印和时间戳.keyBy(WaterSensor::getId).window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String key,Context ctx,Iterable<WaterSensor> elements,Collector<String> out) throws Exception {String msg = "当前key: " + key+ "窗口: [" + ctx.window().getStart() / 1000 + "," + ctx.window().getEnd()/1000 + ") 一共有 "+ elements.spliterator().estimateSize() + "条数据 ";out.collect(msg);}}).print();env.execute();}
}
三、测试结果


相关文章:
大数据-玩转数据-Flink 水印
一、Flink 中的水印 在Flink的流式操作中, 会涉及不同的时间概念: 1.1 处理时间 是指的执行操作的各个设备的时间,对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟。比如, 一个长度为1个小时的窗口将会包含设备…...
【Apollo】阿波罗自动驾驶系统:驶向未来的智能出行(含源码安装)
前言 Apollo (阿波罗)是一个开放的、完整的、安全的平台,将帮助汽车行业及自动驾驶领域的合作伙伴结合车辆和硬件系统,快速搭建一套属于自己的自动驾驶系统。 开放能力、共享资源、加速创新、持续共赢是 Apollo 开放平台的口号。百度把自己所拥有的强大、…...
网络-Netty
how pipeline.addLast(ChannelHandler)...
如何使用vue-smooth-dnd
Vue Smooth DnD是一个基于Vue的平滑易用的拖放库。它提供了简单易用的API和可自定义的样式。 要使用Vue Smooth DnD,可以按照以下步骤进行操作: 安装Vue Smooth DnD npm install vue-smooth-dnd --save 在组件中引入Vue Smooth DnD import VueSmoot…...
为AWS认证做好准备:一份全面的备考指南
随着云计算的快速发展,越来越多的专业人士选择获取AWS(亚马逊网络服务)认证。这个认证不仅可以证明你对AWS的理解和专业技能,还有助于你在云计算领域获得更好的工作机会。 以下是一份全面的备考指南,帮助你为AWS认证做…...
尚硅谷SpringMVC
九、HttpMessageConverter...
django的简易的图书管理系统jsp书店进销存源代码MySQL
本项目为前几天收费帮学妹做的一个项目,Java EE JSP项目,在工作环境中基本使用不到,但是很多学校把这个当作编程入门的项目来做,故分享出本项目供初学者参考。 一、项目描述 django的简易的图书管理系统 系统有1权限:…...
力扣125. 验证回文串
125. 验证回文串 如果在将所有大写字符转换为小写字符、并移除所有非字母数字字符之后,短语正着读和反着读都一样。则可以认为该短语是一个 回文串 。 字母和数字都属于字母数字字符。 给你一个字符串 s,如果它是 回文串 ,返回 true &…...
用WebStorm创建Mock数据
WebStorm是一款强大的集成式开发环境,它集成了许多实用的功能,包括Mock数据的创建。 下面是用WebStorm创建Mock数据的步骤: 打开WebStorm,选择一个项目或新建一个项目;在项目中创建一个名为“mock”的文件夹…...
Python钢筋混凝土结构计算.pdf-已知弯矩确定混凝土梁截面尺寸
计算原理 确定混凝土梁截面的合理尺寸通常需要考虑弯矩、受力要求和约束条件等多个因素。以下是一种常见的计算公式,用于基于已知弯矩确定混凝土梁截面的合理尺寸: 请注意,以上公式仅提供了一种常见的计算方法,并且具体的规范和设…...
【正点原子STM32连载】第二十四章 高级定时器PWM输入模式实验 摘自【正点原子】APM32F407最小系统板使用指南
1)实验平台:正点原子stm32f103战舰开发板V4 2)平台购买地址:https://detail.tmall.com/item.htm?id609294757420 3)全套实验源码手册视频下载地址: http://www.openedv.com/thread-340252-1-1.html# 第二…...
Adapter Tuning Overview:在CV,NLP,多模态领域的代表性工作
文章目录 Delta TuningAdapter Tuning in CVAdapter Tuning in NLP Delta Tuning Adapter Tuning in CV 题目: Learning multiple visual domains with residual adapters 机构:牛津VGG组 论文: https://arxiv.org/pdf/1705.08045.pdf Adapter Tuning in NLP …...
velocity一个基于Java的模板引擎
参考:https://blog.csdn.net/m0_51517236/article/details/126175283 http://www.51gjie.com/javaweb/896.html...
异步servlet
我们日常使用的 SpringMVC,基本上都不是异步 Servlet,而学习 WebFlux,异步 Servlet 是基础,WebFlux。 1.什么是异步 Servlet 先来说说什么是非异步 Servlet。 在 Servlet3.0 之前,Servlet 采用 Thread-Per-Request 的方…...
煤矿皮带运输智能监控算法 opencv
煤矿皮带运输智能监控算法通过opencvpython深度学习算法网络模型,煤矿皮带运输智能监控算法实时监测皮带运输过程中的各种异常情况,如跑偏、撕裂、堆料异常等,一旦检测到异常情况,立即发出告警并采取相应的措施,以保障…...
Docker搭建elasticsearch+kibana测试
最近需要做大数据画像,所以先简单搭建一个eskibana学习使用,记录一下搭建过程和遇到的问题以及解决办法 1.拉取es和kibana镜像 在拉取镜像之前先搜索一下 elasticsearch发现是存在elasticsearch镜像的,我一般习惯性拉取最新镜像,…...
QT(C++)-QTreeview节点折叠与展开
文章目录 1、前言2、QTreeview全部展开与折叠3、QTreeview某个节点展开与折叠3.1 节点折叠与展开的信号与槽3.2 槽函数的实现3.3 某个节点展开与折叠 1、前言 最近要用QT开发项目,对QT不是很熟,就根据网上的查到的知识和自己的摸索,将一些经…...
项目 - 后端技术栈转型方案
前言 某开发项目的后端技术栈比较老了,现在想换到新的技术栈上。使用更好的模式、设计思想、更合理的架构等,为未来的需求迭代做铺垫。怎么办呢?假设系统目前在线上运行着的,直接整体换的话耗时太久,且中间还有新的需…...
Oracle权限语句
授予权限:grant 权限 to 用户名; 撤销权限:revoke 权限 from 用户名; 常用: 创建用户: create user zhangsan identified by zhangsan; grant connect, resource to zhangsan; //授权zhangsan用户连接权限 grant create …...
微信小程序发布一个npm包
参考:https://developers.weixin.qq.com/miniprogram/dev/devtools/npm.html 同npm一样流程 npm install weixin_heath_apis...
后进先出(LIFO)详解
LIFO 是 Last In, First Out 的缩写,中文译为后进先出。这是一种数据结构的工作原则,类似于一摞盘子或一叠书本: 最后放进去的元素最先出来 -想象往筒状容器里放盘子: (1)你放进的最后一个盘子(…...
【根据当天日期输出明天的日期(需对闰年做判定)。】2022-5-15
缘由根据当天日期输出明天的日期(需对闰年做判定)。日期类型结构体如下: struct data{ int year; int month; int day;};-编程语言-CSDN问答 struct mdata{ int year; int month; int day; }mdata; int 天数(int year, int month) {switch (month){case 1: case 3:…...
阿里云ACP云计算备考笔记 (5)——弹性伸缩
目录 第一章 概述 第二章 弹性伸缩简介 1、弹性伸缩 2、垂直伸缩 3、优势 4、应用场景 ① 无规律的业务量波动 ② 有规律的业务量波动 ③ 无明显业务量波动 ④ 混合型业务 ⑤ 消息通知 ⑥ 生命周期挂钩 ⑦ 自定义方式 ⑧ 滚的升级 5、使用限制 第三章 主要定义 …...
电脑插入多块移动硬盘后经常出现卡顿和蓝屏
当电脑在插入多块移动硬盘后频繁出现卡顿和蓝屏问题时,可能涉及硬件资源冲突、驱动兼容性、供电不足或系统设置等多方面原因。以下是逐步排查和解决方案: 1. 检查电源供电问题 问题原因:多块移动硬盘同时运行可能导致USB接口供电不足&#x…...
【Zephyr 系列 10】实战项目:打造一个蓝牙传感器终端 + 网关系统(完整架构与全栈实现)
🧠关键词:Zephyr、BLE、终端、网关、广播、连接、传感器、数据采集、低功耗、系统集成 📌目标读者:希望基于 Zephyr 构建 BLE 系统架构、实现终端与网关协作、具备产品交付能力的开发者 📊篇幅字数:约 5200 字 ✨ 项目总览 在物联网实际项目中,**“终端 + 网关”**是…...
拉力测试cuda pytorch 把 4070显卡拉满
import torch import timedef stress_test_gpu(matrix_size16384, duration300):"""对GPU进行压力测试,通过持续的矩阵乘法来最大化GPU利用率参数:matrix_size: 矩阵维度大小,增大可提高计算复杂度duration: 测试持续时间(秒&…...
.Net Framework 4/C# 关键字(非常用,持续更新...)
一、is 关键字 is 关键字用于检查对象是否于给定类型兼容,如果兼容将返回 true,如果不兼容则返回 false,在进行类型转换前,可以先使用 is 关键字判断对象是否与指定类型兼容,如果兼容才进行转换,这样的转换是安全的。 例如有:首先创建一个字符串对象,然后将字符串对象隐…...
重启Eureka集群中的节点,对已经注册的服务有什么影响
先看答案,如果正确地操作,重启Eureka集群中的节点,对已经注册的服务影响非常小,甚至可以做到无感知。 但如果操作不当,可能会引发短暂的服务发现问题。 下面我们从Eureka的核心工作原理来详细分析这个问题。 Eureka的…...
RabbitMQ入门4.1.0版本(基于java、SpringBoot操作)
RabbitMQ 一、RabbitMQ概述 RabbitMQ RabbitMQ最初由LShift和CohesiveFT于2007年开发,后来由Pivotal Software Inc.(现为VMware子公司)接管。RabbitMQ 是一个开源的消息代理和队列服务器,用 Erlang 语言编写。广泛应用于各种分布…...
LabVIEW双光子成像系统技术
双光子成像技术的核心特性 双光子成像通过双低能量光子协同激发机制,展现出显著的技术优势: 深层组织穿透能力:适用于活体组织深度成像 高分辨率观测性能:满足微观结构的精细研究需求 低光毒性特点:减少对样本的损伤…...
