大数据-玩转数据-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...
Java 语言特性(面试系列2)
一、SQL 基础 1. 复杂查询 (1)连接查询(JOIN) 内连接(INNER JOIN):返回两表匹配的记录。 SELECT e.name, d.dept_name FROM employees e INNER JOIN departments d ON e.dept_id d.dept_id; 左…...
关于 WASM:1. WASM 基础原理
一、WASM 简介 1.1 WebAssembly 是什么? WebAssembly(WASM) 是一种能在现代浏览器中高效运行的二进制指令格式,它不是传统的编程语言,而是一种 低级字节码格式,可由高级语言(如 C、C、Rust&am…...
中医有效性探讨
文章目录 西医是如何发展到以生物化学为药理基础的现代医学?传统医学奠基期(远古 - 17 世纪)近代医学转型期(17 世纪 - 19 世纪末)现代医学成熟期(20世纪至今) 中医的源远流长和一脉相承远古至…...
安全突围:重塑内生安全体系:齐向东在2025年BCS大会的演讲
文章目录 前言第一部分:体系力量是突围之钥第一重困境是体系思想落地不畅。第二重困境是大小体系融合瓶颈。第三重困境是“小体系”运营梗阻。 第二部分:体系矛盾是突围之障一是数据孤岛的障碍。二是投入不足的障碍。三是新旧兼容难的障碍。 第三部分&am…...
Selenium常用函数介绍
目录 一,元素定位 1.1 cssSeector 1.2 xpath 二,操作测试对象 三,窗口 3.1 案例 3.2 窗口切换 3.3 窗口大小 3.4 屏幕截图 3.5 关闭窗口 四,弹窗 五,等待 六,导航 七,文件上传 …...
uniapp 开发ios, xcode 提交app store connect 和 testflight内测
uniapp 中配置 配置manifest 文档:manifest.json 应用配置 | uni-app官网 hbuilderx中本地打包 下载IOS最新SDK 开发环境 | uni小程序SDK hbulderx 版本号:4.66 对应的sdk版本 4.66 两者必须一致 本地打包的资源导入到SDK 导入资源 | uni小程序SDK …...
数学建模-滑翔伞伞翼面积的设计,运动状态计算和优化 !
我们考虑滑翔伞的伞翼面积设计问题以及运动状态描述。滑翔伞的性能主要取决于伞翼面积、气动特性以及飞行员的重量。我们的目标是建立数学模型来描述滑翔伞的运动状态,并优化伞翼面积的设计。 一、问题分析 滑翔伞在飞行过程中受到重力、升力和阻力的作用。升力和阻力与伞翼面…...
goreplay
1.github地址 https://github.com/buger/goreplay 2.简单介绍 GoReplay 是一个开源的网络监控工具,可以记录用户的实时流量并将其用于镜像、负载测试、监控和详细分析。 3.出现背景 随着应用程序的增长,测试它所需的工作量也会呈指数级增长。GoRepl…...
JavaScript 标签加载
目录 JavaScript 标签加载script 标签的 async 和 defer 属性,分别代表什么,有什么区别1. 普通 script 标签2. async 属性3. defer 属性4. type"module"5. 各种加载方式的对比6. 使用建议 JavaScript 标签加载 script 标签的 async 和 defer …...
简单聊下阿里云DNS劫持事件
阿里云域名被DNS劫持事件 事件总结 根据ICANN规则,域名注册商(Verisign)认定aliyuncs.com域名下的部分网站被用于非法活动(如传播恶意软件);顶级域名DNS服务器将aliyuncs.com域名的DNS记录统一解析到shado…...
