关于flink重新提交任务,重复消费kafka的坑
异常现象1
按照以下方式设置backend目录和checkpoint目录,fsbackend目录有数据,checkpoint目录没数据
env.getCheckpointConfig().setCheckpointStorage(PropUtils.getValueStr(Constant.ENV_FLINK_CHECKPOINT_PATH));
env.setStateBackend(new FsStateBackend(PropUtils.getValueStr(Constant.ENV_FLINK_STATEBACKEND_PATH)));
原因
我以为checkpoint和fsbackend要同时设置,其实,1.14.3版本,setCheckpointStorage和stateBackend改成了分着设置
我上边代码这样设置,相当于首先指定了以下checkpoint按照默认的backend存储,然后又指定了按照fsbackend存储,因此首先指定的checkpoint目录没有数据。
正解
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(PropUtils.getValueStr(Constant.ENV_FLINK_CHECKPOINT_PATH));
State Backends | Apache Flink

异常现象2
开启checkpoint eos,开启容错,每次任务重新提交都会重新消费kafka已经完成了checkpoint的数据
原因
我以为只要开启这两个配置就可以保证已经checkpoint的kafka数据不会被重复消费,其实不然
checkpoint是flink内部的容错机制,他能保证在设置了失败重启策略之后(setRestartStrategy),如果发生异常导致失败重试之后自动从最新checkpoint恢复。不是手动重启。。。手动重启默认不会进行加载状态数据,所以每次都会从头消费

正解
flink任务 -s 指定恢复点提交,这个恢复点可以是checkpoint也可以时savepoint。
# 启动
/home/cuadmin/flink-1.14.3/bin/flink run -d \
-c cn.flink.ApplicationMaster \/home/cuadmin/portal-flink-2021.0.1-SNAPSHOT-shaded.jar# 备份,创建savepoint
/home/cuadmin/flink-1.14.3/bin/flink savepoint 19f4bb5d103ea8695712d4d1a797893f /home/cuadmin/flink-1.14.3/savepoint# 指定savepoint启动
/home/cuadmin/flink-1.14.3/bin/flink run -d \
-c cn.flink.ApplicationMaster \
-s file:/home/cuadmin/flink-1.14.3/savepoint/savepoint-033556-251a9e55ed25 \
/home/cuadmin/portal-flink-2021.0.1-SNAPSHOT-shaded.jar
异常现象4
这是错误的
# 指定savepoint启动
/home/cuadmin/flink-1.14.3/bin/flink run -d \
-c cn.flink.ApplicationMaster \/home/cuadmin/portal-flink-2021.0.1-SNAPSHOT-shaded.jar
-s file:/home/cuadmin/flink-1.14.3/savepoint/savepoint-033556-251a9e55ed25 \
按照上述命令执行,这个地方显示恢复点的加载情况,这里没显示,代表恢复点没有执行成功
原因
-s的位置有问题,我之前以为没有顺序,把-s 放到了命令最后,结果没报错,也没识别。。
正解
-s 位置要正确
# 指定savepoint启动
/home/cuadmin/flink-1.14.3/bin/flink run -d \
-c cn.flink.ApplicationMaster \
-s file:/home/cuadmin/flink-1.14.3/savepoint/savepoint-033556-251a9e55ed25 \
/home/cuadmin/portal-flink-2021.0.1-SNAPSHOT-shaded.jar
异常现象3
我记得savepoint和checkpoint是都可以用来flink -s 进行恢复点恢复的。但是每次都提示恢复失败,提示文件找不到,savepoint就可以。。。
原因
cancel job会将 checkpoint的数据删掉。。。

正解
测试的时候,直接stop-cluster,这样checkpoint数据就不会被删除了
保留 Checkpoint
Checkpoint 在默认的情况下仅用于恢复失败的作业,并不保留,当程序取消时 checkpoint 就会被删除。当然,你可以通过配置来保留 checkpoint,这些被保留的 checkpoint 在作业失败或取消时不会被清除。这样,你就可以使用该 checkpoint 来恢复失败的作业。
CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
ExternalizedCheckpointCleanup 配置项定义了当作业取消时,对作业 checkpoint 的操作:
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:当作业取消时,保留作业的 checkpoint。注意,这种情况下,需要手动清除该作业保留的 checkpoint。ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION:当作业取消时,删除作业的 checkpoint。仅当作业失败时,作业的 checkpoint 才会被保留。
总结
1、savepoint的数据要比checkpoint更加稳定,比如你可以通过移动(拷贝)savepoint 目录到任意地方,然后再进行恢复。checkpoint就不可以,因为他有很多相对路径配置。
2、savepoint和checkpoint一般都能作为恢复点使用,例外情况是使用 RocksDB 状态后端的增量 Checkpoint。他们使用了一些 RocksDB 内部格式,而不是 Flink 的本机 Savepoint 格式。这使他们成为了与 Savepoint 相比,更轻量级的 Checkpoint 机制的第一个实例。
3、任务因为偶然原因内部重启(task级别),通过失败重试机制+checkpoint自动进行重放,任务因重启、断电、死机等外部因素(cluster级别),通过-s 指定checkpoint/savepoint恢复点进行手动重放。这样就可以保证状态数据的稳定
相关文章:
关于flink重新提交任务,重复消费kafka的坑
异常现象1 按照以下方式设置backend目录和checkpoint目录,fsbackend目录有数据,checkpoint目录没数据 env.getCheckpointConfig().setCheckpointStorage(PropUtils.getValueStr(Constant.ENV_FLINK_CHECKPOINT_PATH)); env.setStateBackend(new FsStat…...
Win11右键恢复Win10老版本
Win11右键恢复Win10老版本 最近自己更新了windows11的OS,整体感觉都是不错的,但是就是每次右键菜单我都要再次点击下展开更多选项,这对追求极简主义的我,就是不爽, 手动恢复win10操作吧! 第一种:创建文件(简单快速) 1.新建一个resoreRightKey.reg文件,并在里面填入如下代码 W…...
ur机械臂30003端口socket通信踩坑(double类型数据怎么解析)
坑的由来 都知道在网络通信时要把网络字节序转化为主机字节序才行,但是c里的标准库函数ntohl默认是转换32位字节序的数据,也就是说默认是转换float类型的数据;而ur机械臂30003端口发送的是double类型的数据,没法直接用ntohl进行转…...
代理IP与Socks5代理的技术奇妙之旅
随着数字化时代的崛起,网络工程师们日益承担着维护网络稳定性和保护数据安全的重任。在这个充满挑战的世界里,代理IP与Socks5代理技术成为了他们的秘密武器,本文将带您踏上一段技术奇妙之旅,深入了解这两项技术在不同领域中的应用…...
自动化测试定位不到元素?可能是 frame 在搞鬼
很多人在用Splinter或Selenium定位页面元素的时候会遇到定位不到的问题,明明元素就在那儿,就是定位不到,这种情况很有可能是frame在搞鬼。 说白了就是网站上的网页A,又嵌入了其他网页B。你访问了网页A,在里面可以看到…...
uni-app 开发中,监听 input 键盘事件获取不到按下按键值怎么办?
uniapp 开发 H5 时,无法监听按钮键盘事件的原因以及解决方法。 问题描述: 不少 uni-app 开发者在使用 input 组件时,监听 keyup 事件时,获取不到键盘的 keyCode。编写的代码如下: <template><input keyup&…...
【juc】countdownlatch实现并发网络请求
目录 一、截图示例二、代码示例2.1 测试代码2.2 接口代码 一、截图示例 二、代码示例 2.1 测试代码 package com.learning.countdownlatch;import lombok.extern.slf4j.Slf4j; import org.springframework.web.client.RestTemplate;import java.util.Arrays; import java.uti…...
在供应链管理中,如何做好库存分析?库存分析有哪些监控指标?
在供应链管理中,库存分析是其重要的一环。库存分析的方法繁杂且广泛,选择正确的方法才能更好的进行库存分析,下面就为大家盘点一些常用的库存分析方法和监控指标,全程干货,建议收藏! 01 如何进行库存分析&…...
黑豹程序员-架构师学习路线图-百科:Database数据库
文章目录 1、什么是Database2、发展历史3、数据库排行网4、总结 1、什么是Database 当今世界是一个充满着数据的互联网世界,各处都充斥着大量的数据。即这个互联网世界就是数据世界。 支撑这个数据世界的基石就是数据库,数据库也可以称为数据的仓库。 …...
你相信光吗?黑灯工厂重新相信“光”
你知道“黑灯工厂”吗?望文生义,所谓黑灯工厂,就是可以不需要照明的工厂。全程流水线自动化生产,无人干预、无人值守,工厂变成黑匣子,原材料进去,成品出来,流水线上百分百自动化。 完…...
Ubuntu 20.04使用源码安装nginx 1.14.0
nginx安装及使用(详细版)是一篇参考博文。 http://nginx.org/download/可以选择下载源码的版本。 sudo wget http://nginx.org/download/nginx-1.14.0.tar.gz下载源代码。 sudo tar xzf nginx-1.14.0.tar.gz进行解压。 cd nginx-1.14.0进入到源代码…...
springboot框架拦截器中HttpServletRequest 请求如何区分是图片上传流还是普通的字符流?
在Spring Boot框架中的拦截器(Interceptor)中,可以通过检查Content-Type请求头来区分图片上传流和普通的字符流。 当客户端发送POST请求并携带文件时,Content-Type请求头通常会包含multipart/form-data或者类似的值。这表明该请求…...
简单聊聊 TCP 协议
简单聊聊 TCP 协议 如何实现可靠传输 ?完全可靠存在比特差错存在丢包流水线可靠数据传输协议回退N步 (GBN)选择重传 (ARQ) 小结 TCPTCP 连接报文段结构序号和确认号 可靠数据传输避免重传超时时间加倍快速重传回退N步还是选择重传 流量控制连接管理拥塞控制拥塞原因拥塞控制方…...
钡铼BL124PN:简单快速转换Profinet到Ethernet/IP
钡铼技术BL124PN是一款高性能的Profinet转Ethernet/IP网关设备。该网关专为工业自动化领域设计,用于实现不同协议之间的互连和通信。BL124PN采用可靠稳定的硬件和先进的通信技术,具有以下主要特点: 协议转换能力:BL124PN能够将Pr…...
【golang】go 空结构体 详解 空结构体内容占用及大小
一、空结构体基础 空结构实例 和 空结构体变量 本质是一样的 1、所有空结构体地址都是一样的2、大小都为0(最独特的) package mainimport ("fmt""time""unsafe" )type EST struct { }func main() {// 一、基础// 空结构…...
身为产品经理该如何向客户推广API商品数据接口
在当今数字化的时代,API(Application Programming Interface,应用程序编程接口)已成为各种软件应用程序之间交互数据的主要方式。API商品数据接口作为一种特殊类型的API,能够让不同的系统之间共享商品数据,…...
【数据结构】460. LFU 缓存
460. LFU 缓存 解题思路 get操作 返回key对应的val 然后增加对应的freq插入操作 如果key已经存在 直接进行更新 如果不存在 但是容器已经满了 直接进行删除freq最小的Key 之后进行插入 class LFUCache {// key到 val的映射 KVHashMap<Integer,Integer> keyToVal;// …...
文字转语音播报模块(一):阿里云nls服务使用示例
一、业务场景 最近笔者在业务中涉及到语音告警的模块,需要讲告警内容以文件或流形式返回给前端进行语音播报,具体的分析与处理如下 二、业务分析 首先告警内容提示信息这里做的处理是通过专门字段去存储、编辑,根据拟定好的代码逻辑判断是…...
Vscode配置C#编程环境(win10)
目录 1、安装好Vscode 2、下载安装.NetCore SDK 3、配置C#环境 3.1 打开Vscode并下载扩展 3.2 Vscode中打开文件夹并配置环境 3.3 调试运行 1、安装好Vscode 2、下载安装.NetCore SDK 官网如下,下载完成后双击打开一路走到底就行.NetCore SDK官网 软件显示安…...
python:xlrd 读取 Excel文件,显示在 tkinterTable 表格中
pip install xlrd xlrd-1.2.0-py2.py3-none-any.whl (103 kB) 摘要: Library for developers to extract data from Microsoft Excel (tm) spreadsheet files pip install tkinterTable tkintertable-1.3.3.tar.gz (58 kB) 摘要: Extendable table class for Tkinter 源代…...
XML Group端口详解
在XML数据映射过程中,经常需要对数据进行分组聚合操作。例如,当处理包含多个物料明细的XML文件时,可能需要将相同物料号的明细归为一组,或对相同物料号的数量进行求和计算。传统实现方式通常需要编写脚本代码,增加了开…...
Python爬虫实战:研究MechanicalSoup库相关技术
一、MechanicalSoup 库概述 1.1 库简介 MechanicalSoup 是一个 Python 库,专为自动化交互网站而设计。它结合了 requests 的 HTTP 请求能力和 BeautifulSoup 的 HTML 解析能力,提供了直观的 API,让我们可以像人类用户一样浏览网页、填写表单和提交请求。 1.2 主要功能特点…...
第19节 Node.js Express 框架
Express 是一个为Node.js设计的web开发框架,它基于nodejs平台。 Express 简介 Express是一个简洁而灵活的node.js Web应用框架, 提供了一系列强大特性帮助你创建各种Web应用,和丰富的HTTP工具。 使用Express可以快速地搭建一个完整功能的网站。 Expre…...
day52 ResNet18 CBAM
在深度学习的旅程中,我们不断探索如何提升模型的性能。今天,我将分享我在 ResNet18 模型中插入 CBAM(Convolutional Block Attention Module)模块,并采用分阶段微调策略的实践过程。通过这个过程,我不仅提升…...
【大模型RAG】Docker 一键部署 Milvus 完整攻略
本文概要 Milvus 2.5 Stand-alone 版可通过 Docker 在几分钟内完成安装;只需暴露 19530(gRPC)与 9091(HTTP/WebUI)两个端口,即可让本地电脑通过 PyMilvus 或浏览器访问远程 Linux 服务器上的 Milvus。下面…...
【SQL学习笔记1】增删改查+多表连接全解析(内附SQL免费在线练习工具)
可以使用Sqliteviz这个网站免费编写sql语句,它能够让用户直接在浏览器内练习SQL的语法,不需要安装任何软件。 链接如下: sqliteviz 注意: 在转写SQL语法时,关键字之间有一个特定的顺序,这个顺序会影响到…...
智能AI电话机器人系统的识别能力现状与发展水平
一、引言 随着人工智能技术的飞速发展,AI电话机器人系统已经从简单的自动应答工具演变为具备复杂交互能力的智能助手。这类系统结合了语音识别、自然语言处理、情感计算和机器学习等多项前沿技术,在客户服务、营销推广、信息查询等领域发挥着越来越重要…...
如何更改默认 Crontab 编辑器 ?
在 Linux 领域中,crontab 是您可能经常遇到的一个术语。这个实用程序在类 unix 操作系统上可用,用于调度在预定义时间和间隔自动执行的任务。这对管理员和高级用户非常有益,允许他们自动执行各种系统任务。 编辑 Crontab 文件通常使用文本编…...
解读《网络安全法》最新修订,把握网络安全新趋势
《网络安全法》自2017年施行以来,在维护网络空间安全方面发挥了重要作用。但随着网络环境的日益复杂,网络攻击、数据泄露等事件频发,现行法律已难以完全适应新的风险挑战。 2025年3月28日,国家网信办会同相关部门起草了《网络安全…...
PostgreSQL——环境搭建
一、Linux # 安装 PostgreSQL 15 仓库 sudo dnf install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-$(rpm -E %{rhel})-x86_64/pgdg-redhat-repo-latest.noarch.rpm# 安装之前先确认是否已经存在PostgreSQL rpm -qa | grep postgres# 如果存在࿰…...


