Flink和Kafka连接时的精确一次保证
Flink写入Kafka两阶段提交
端到端的 exactly-once(精准一次)
kafka -> Flink -> kafka
1)输入端
输入数据源端的 Kafka 可以对数据进行持久化保存,并可以重置偏移量(offset)
2)Flink内部
Flink 内部可以通过检查点机制保证状态和处理结果的 exactly-once 语义
3)输出端
两阶段提交(2PC)。
写入 Kafka 的过程实际上是一个两段式的提交:处理完毕得到结果,写入 Kafka 时是基于事务的“预提交”;等到检查点保存完毕,才会提交事务进行“正式提交”。
如果中间出现故障,事务进行回滚,预提交就会被放弃;恢复状态之后,也只能恢复所有已经确认提交的操作。
必须的配置
1)必须启用检查点
2)指定 KafkaSink 的发送级别为 DeliveryGuarantee.EXACTLY_ONCE
3)配置 Kafka 读取数据的消费者的隔离级别【默认kafka消费者隔离级别是读未提交,2PC第一阶段预提交数据也会被读到,下游消费者需要设置为读已提交】
4)事务超时配置
【配置的事务超时时间 transaction.timeout.ms 默认是1小时,而Kafka 集群配置的事务最大超时时间 transaction.max.timeout.ms 默认是15 分钟。在检查点保存时间很长时,有可能出现 Kafka 已经认为事务超时了,丢弃了预提交的数据;而Sink任务认为还可以继续等待。如果接下来检查点保存成功,发生故障后回滚到这个检查点的状态,这部分数据就被真正丢掉了。因此checkpoint 间隔 < 事务超时时间 < max的15分钟】
代码实战
kafka -> Flink -> kafka【Flink处理kafka来源数据再输出到kafka】
public class KafkaEOSDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 【1】、启用检查点,设置为精准一次env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);CheckpointConfig checkpointConfig = env.getCheckpointConfig();checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/chk");checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// 2.读取 kafkaKafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("hadoop102:9092").setGroupId("default").setTopics("topic_1").setValueOnlyDeserializer(new SimpleStringSchema()).setStartingOffsets(OffsetsInitializer.latest()).build();DataStreamSource<String> kafkasource = env.fromSource(kafkaSource,WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource");/*3.写出到 Kafka精准一次 写入 Kafka,需要满足以下条件,【缺一不可】1、开启 checkpoint2、sink 设置保证级别为 精准一次3、sink 设置事务前缀4、sink 设置事务超时时间: checkpoint 间隔 < 事务超时时间 < max的15分钟*/KafkaSink<String> kafkaSink = KafkaSink.<String>builder()// 指定 kafka 的地址和端口.setBootstrapServers("hadoop102:9092")// 指定序列化器:指定 Topic 名称、具体的序列化.setRecordSerializer(KafkaRecordSerializationSchema.<String>builder().setTopic("ws").setValueSerializationSchema(new SimpleStringSchema()).build())// 【3.1】 精准一次,开启 2pc.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)// 【3.2】 精准一次,必须设置 事务的前缀.setTransactionalIdPrefix("li-")// 【3.3】 设置事务超时时间.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "").build();kafkasource.sinkTo(kafkaSink);env.execute();}
}
后续读取“ws”这个 topic 的消费者,要设置事务的隔离级别为“读已提交”
public class KafkaEOSConsumer {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 消费 在前面使用【两阶段提交】写入的 TopicKafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("hadoop102:9092").setGroupId("default").setTopics("ws").setValueOnlyDeserializer(new SimpleStringSchema()).setStartingOffsets(OffsetsInitializer.latest())// 作为 下游的消费者,要设置事务的隔离级别为 【读已提交】.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed").build();env.fromSource(kafkaSource,WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource").print();env.execute();}
}
处理程序以及消费程序如上设置才能真正实现端到端精准一次的保证。
相关文章:
Flink和Kafka连接时的精确一次保证
Flink写入Kafka两阶段提交 端到端的 exactly-once(精准一次) kafka -> Flink -> kafka 1)输入端 输入数据源端的 Kafka 可以对数据进行持久化保存,并可以重置偏移量(offset) 2)Flink内…...
UE4动作游戏实例RPG Action解析三:实现效果,三连击Combo,射线检测,显示血条,火球术
一、三连Combo 实现武器三连击,要求: 1.下一段Combo可以随机选择, 2.在一定的时机才能再次检测输入 3. 等当前片段播放完才播放下一片段 1.1、蒙太奇设置 通过右键-新建蒙太奇片段,在蒙太奇里创建三个片段,并且移除相关连接,这样默认只会播放第一个片段 不同片段播…...
Linux/麒麟系统上部署Vue+SpringBoot前后端分离项目
目录 1. 前端准备工作 1.1 在项目根目录创建两份环境配置文件 1.2 环境配置 2. 后端准备工作 2.1 在项目resources目录创建两份环境配置文件 2.2 环境配置 3. 前后端打包 3.1 前端打包 3.2 后端打包 4、服务器前后端配置及部署 4.1 下载、安装、启动Nginx 4.2 前端项目部署…...
STM32在FreeRTOS下的us延时
STM32在FreeRTOS下的us延时 前言 freeRTOS下跑SPI时需要微秒级别的延时,但是freeRTOS只提供了毫秒级的,记录一下实现us延时的方法。 前期分析 最简单的方式就是开个定时器或者干脆直接计算一下用nop做都可以实现us延时,但是显然还是使用滴…...
软件测试/人工智能丨深入人工智能软件测试:PyTorch引领新时代
在人工智能的浪潮中,软件测试的角色变得愈发关键。本文将介绍在人工智能软件测试中的一些关键技术,以及如何借助PyTorch深度学习框架来推动测试的创新与升级。 PyTorch:深度学习的引擎 PyTorch作为一种开源的深度学习框架,为软件…...
Android 当中的 Fragment 协作解耦方式
Android 当中的 Fragment 协作解耦方式 文章目录 Android 当中的 Fragment 协作解耦方式第一章 前言介绍第01节 遇到的问题第02节 绘图说明 第二章 核心代码第01节 代理人接口第02节 中间人 Activity第03节 开发者A第04节 开发者B第05节 测试类 第一章 前言介绍 第01节 遇到的…...
城市网吧视频智能监控方案,实现视频远程集中监控
网吧环境较为复杂,电脑设备众多且人员流动性大,极易发生人员或消防事故,亟需改变,TSINGSEE青犀AI智能网吧视频监管方案可以帮助实现对网吧环境和用户活动的实时监控和管理。 1、视频监控系统 在网吧内部布置高清摄像头࿰…...
C#WPF视频播放器实例
本文实例演示C#WPF视频播放器 实例如下: 修改mainwindow的代码 <Windowx:Class="PlayerDemo.MainWindow"xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"xml…...
【uniapp】Google Maps
话不多说 直接上干货 提前申请谷歌地图账号一、新建地图 使用h5获取当前定位或者使用三方uniapp插件 var coords ""navigator.geolocation.getCurrentPosition(function(position) {coords {lat: position.coords.latitude,lng: position.coords.longitude};lats …...
C语言变量与常量
跟着肯哥(不是我)学C语言的变量和常量、跨文件访问、栈空间 栈空间还不清楚,期待明天的课程内容 C变量 变量(Variable)是用于存储和表示数据值的名称。 主要包括四个环节:定义、初始化、声明、使用 在我刚…...
AI创作系统ChatGPT网站源码/支持DALL-E3文生图/支持最新GPT-4-Turbo模型+Prompt应用
一、AI创作系统 SparkAi创作系统是基于OpenAI很火的ChatGPT进行开发的Ai智能问答系统和Midjourney绘画系统,支持OpenAI-GPT全模型国内AI全模型。本期针对源码系统整体测试下来非常完美,可以说SparkAi是目前国内一款的ChatGPT对接OpenAI软件系统。那么如…...
二维码智慧门牌管理系统升级,异常门牌聚合解决方案助力高效管理
文章目录 前言一、异常门牌聚合解决方案 前言 在今天的数字化时代,智慧城市已成为发展趋势,其中二维码智慧门牌管理系统扮演着至关重要的角色。通过对门牌信息进行数字化管理,该系统极大提升了城市管理的效率和便捷性。然而,随着…...
【XTDrone Ubuntu20.04】XTDrone+ Ubuntu20.04 + PX4安装
XTDrone仿真平台配置 文章目录 XTDrone仿真平台配置依赖安装 ROS一键安装Marvos安装PX4 安装安装QTGroundControlXTDrone下载安装 环境: VMWare 16.0 Ubuntu 22.04 (因为没人配过)Ubuntu 20.04 参考文章: 仿真平台基础配置 (yuq…...
河北大学选择ZStack Cube超融合一体机打造实训云平台
河北大学通过云轴科技ZStack Cube超融合一体机构建校园实训云平台,部署测试仅耗时1天,该平台能够更快地为学生提供高性能、高可用的云主机、云存储和云网络服务;同时也能满足日常运维管理要求,为学生提供更好的实训环境。 河北省…...
IDEA远程一键部署SpringBoot到Docker
IDEA是Java开发利器,Spring Boot是Java生态中最流行的微服务框架,docker是时下最火的容器技术,那么它们结合在一起会产生什么化学反应呢? 一、开发前准备 1. Docker安装 可以参考:https://docs.docker.com/install/ 2…...
索引三星结构
三星索引的定义,可以先给我们对索引优化提供一个大概的思路: 满足第1颗星: 取出所有的等值谓词的列,作为索引最开头的列——以任意顺序都可以。 满足第2颗星: 将order by加入到索引列,不要改变这些列的顺…...
rust 笔记 高级错误处理
文章目录 错误处理组合器or() 和 and()or_else() 和 and_then()filtermap() 和 map_err()map_or() 和 map_or_else()ok_or() and ok_or_else() 自定义错误类型错误转换 From 特征 归一化不同的错误类型Box<dyn Error>自定义错误类型 简化错误处理thiserroranyhow 错误处理…...
python+Django 使用apscheduler实现定时任务 管理调度
apscheduler实现定时任务 管理调度 在Django 项目中经常会用到定时任务去处理一些业务处理 使用 APScheduler 可以轻松地实现定时任务的管理和调度。你可以通过以下步骤来创建、启动、停止和删除定时任务: 1.创建调度器对象: from apscheduler.schedu…...
Java编程中,异步操作流程中,最终一致性以及重试补偿的设计与实现
一、背景 微服务设计中,跨服务的调用,由于网络或程序故障等各种原因,经常会出现调用失败而需要重试。另外,在异步操作中,我们提供接口让外部服务回调。回调过程中,也可能出现故障。 这就要求我们主动向外…...
吴恩达《机器学习》8-7:多元分类
在机器学习领域,经常会遇到不止两个类别的分类问题。这时,需要使用多类分类技术。本文将深入探讨多类分类,并结合学习内容中的示例,了解神经网络在解决这类问题时的应用。 一、理解多类分类 多类分类问题是指当目标有多个类别时…...
linux之kylin系统nginx的安装
一、nginx的作用 1.可做高性能的web服务器 直接处理静态资源(HTML/CSS/图片等),响应速度远超传统服务器类似apache支持高并发连接 2.反向代理服务器 隐藏后端服务器IP地址,提高安全性 3.负载均衡服务器 支持多种策略分发流量…...
linux 下常用变更-8
1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行,YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID: YW3…...
ios苹果系统,js 滑动屏幕、锚定无效
现象:window.addEventListener监听touch无效,划不动屏幕,但是代码逻辑都有执行到。 scrollIntoView也无效。 原因:这是因为 iOS 的触摸事件处理机制和 touch-action: none 的设置有关。ios有太多得交互动作,从而会影响…...
Mac下Android Studio扫描根目录卡死问题记录
环境信息 操作系统: macOS 15.5 (Apple M2芯片)Android Studio版本: Meerkat Feature Drop | 2024.3.2 Patch 1 (Build #AI-243.26053.27.2432.13536105, 2025年5月22日构建) 问题现象 在项目开发过程中,提示一个依赖外部头文件的cpp源文件需要同步,点…...
技术栈RabbitMq的介绍和使用
目录 1. 什么是消息队列?2. 消息队列的优点3. RabbitMQ 消息队列概述4. RabbitMQ 安装5. Exchange 四种类型5.1 direct 精准匹配5.2 fanout 广播5.3 topic 正则匹配 6. RabbitMQ 队列模式6.1 简单队列模式6.2 工作队列模式6.3 发布/订阅模式6.4 路由模式6.5 主题模式…...
Java求职者面试指南:Spring、Spring Boot、MyBatis框架与计算机基础问题解析
Java求职者面试指南:Spring、Spring Boot、MyBatis框架与计算机基础问题解析 一、第一轮提问(基础概念问题) 1. 请解释Spring框架的核心容器是什么?它在Spring中起到什么作用? Spring框架的核心容器是IoC容器&#…...
【电力电子】基于STM32F103C8T6单片机双极性SPWM逆变(硬件篇)
本项目是基于 STM32F103C8T6 微控制器的 SPWM(正弦脉宽调制)电源模块,能够生成可调频率和幅值的正弦波交流电源输出。该项目适用于逆变器、UPS电源、变频器等应用场景。 供电电源 输入电压采集 上图为本设计的电源电路,图中 D1 为二极管, 其目的是防止正负极电源反接, …...
从面试角度回答Android中ContentProvider启动原理
Android中ContentProvider原理的面试角度解析,分为已启动和未启动两种场景: 一、ContentProvider已启动的情况 1. 核心流程 触发条件:当其他组件(如Activity、Service)通过ContentR…...
Neko虚拟浏览器远程协作方案:Docker+内网穿透技术部署实践
前言:本文将向开发者介绍一款创新性协作工具——Neko虚拟浏览器。在数字化协作场景中,跨地域的团队常需面对实时共享屏幕、协同编辑文档等需求。通过本指南,你将掌握在Ubuntu系统中使用容器化技术部署该工具的具体方案,并结合内网…...
何谓AI编程【02】AI编程官网以优雅草星云智控为例建设实践-完善顶部-建立各项子页-调整排版-优雅草卓伊凡
何谓AI编程【02】AI编程官网以优雅草星云智控为例建设实践-完善顶部-建立各项子页-调整排版-优雅草卓伊凡 背景 我们以建设星云智控官网来做AI编程实践,很多人以为AI已经强大到不需要程序员了,其实不是,AI更加需要程序员,普通人…...
