当前位置: 首页 > news >正文

Flink问题解决及性能调优-【Flink不同并行度引起sink2es报错问题】

最近需求,仅想提高sink2es的qps,所以仅调节了sink2es的并行度,但在调节不同算子并行度时遇到一些问题,找出问题的根本原因解决问题,并分析整理。

实例代码

--SET table.exec.state.ttl=86400s; --24 hour,默认: 0 ms
SET table.exec.state.ttl=2592000s; --30 days,默认: 0 msCREATE TABLE kafka_table (mid bigint,db string,sch string,tab string,opt string,ts bigint,ddl string,err string,src map<string,string>,cur map<string,string>,cus map<string,string>,account_id AS IF(cur['account_id'] IS NOT NULL , cur['account_id'], src ['account_id']),publish_time AS IF(cur['publish_time'] IS NOT NULL , cur['publish_time'], src ['publish_time']),msg_status AS IF(cur['msg_status'] IS NOT NULL , cur['msg_status'], src ['msg_status']),send_type AS IF(cur['send_type'] IS NOT NULL , cur['send_type'], src ['send_type'])--event_time as cast(IF(cur['update_time'] IS NOT NULL , cur['update_time'], src ['update_time']) AS TIMESTAMP(3)), -- TIMESTAMP(3)/TIMESTAMP_LTZ(3)--WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE     --SECOND
) WITH ('connector' = 'kafka','topic' = 't1','properties.bootstrap.servers' = 'xx.xx.xx.xx:9092','properties.group.id' = 'g1','scan.startup.mode' = 'earliest-offset',  --group-offsets/earliest-offset/latest-offset--  'properties.enable.auto.commit',= 'true' -- default:false, 如果为false,则在发生checkpoint时触发offset提交'format' = 'json'
);CREATE TABLE es_sink(send_type      STRING,account_id     STRING,publish_time   STRING,grouping_id       INTEGER,init           INTEGER,init_cancel    INTEGER,push          INTEGER,succ           INTEGER,fail           INTEGER,init_delete    INTEGER,update_time    STRING,PRIMARY KEY (group_id,send_type,account_id,publish_time) NOT ENFORCED
)
with ('connector' = 'elasticsearch-6','index' = 'es_sink','document-type' = 'es_sink','hosts' = 'http://xxx:9200','format' = 'json','filter.null-value'='true','sink.bulk-flush.max-actions' = '1000','sink.bulk-flush.max-size' = '10mb'
);CREATE view  tmp as
selectsend_type,account_id,publish_time,msg_status,case when UPPER(opt) = 'INSERT' and msg_status='0'  then 1 else 0 end AS init,case when UPPER(opt) = 'UPDATE' and send_type='1' and msg_status='4' then 1 else 0 end AS init_cancel,case when UPPER(opt) = 'UPDATE' and msg_status='3' then 1 else 0 end AS push,case when UPPER(opt) = 'UPDATE' and (msg_status='1' or msg_status='5') then 1 else 0 end AS succ,case when UPPER(opt) = 'UPDATE' and (msg_status='2' or msg_status='6') then 1 else 0 end AS fail,case when UPPER(opt) = 'DELETE' and send_type='1' and msg_status='0' then  1 else 0 end AS init_delete,event_time,opt,ts
FROM kafka_table
where (UPPER(opt) = 'INSERT' and msg_status='0' )
or        (UPPER(opt) = 'UPDATE' and msg_status in ('1','2','3','4','5','6'))
or        (UPPER(opt) = 'DELETE' and send_type='1' and msg_status='0');--send_type=1          send_type=0
--初始化->0             初始化->0
--取消->4
--推送->3               推送->3
--成功->1               成功->5
--失败->2               失败->6CREATE view  tmp_groupby as
selectCOALESCE(send_type,'N') AS send_type
,COALESCE(account_id,'N') AS account_id
,COALESCE(publish_time,'N') AS publish_time
,case when send_type is null and account_id is null and publish_time is null then 1when send_type is not null and account_id is null and publish_time is null then 2when send_type is not null and account_id is not null and publish_time is null then 3when send_type is not null and account_id is not null and publish_time is not null then 4end grouping_id
,sum(init) as init
,sum(init_cancel) as init_cancel
,sum(push) as push
,sum(succ) as succ
,sum(fail) as fail
,sum(init_delete) as init_delete
from tmp
--GROUP BY GROUPING SETS ((send_type,account_id,publish_time), (send_type,account_id),(send_type), ())
GROUP BY ROLLUP (send_type,account_id,publish_time); --等同于以上INSERT INTO es_sink
selectsend_type,account_id,publish_time,grouping_id,init,init_cancel,push,succ,fail,init_delete,CAST(LOCALTIMESTAMP AS STRING) as update_time
from tmp_groupby

发现问题

由于groupby或join聚合等算子操作的并行度与sink2es算子操作的并行度不同,上游算子同一个key的数据可能会下发到下游多个不同算子中。
导致sink2es出现多个subtask同时操作同一个key(这里key作为主键id),报错如下:

...Caused by: [test1/cfPTBYhcRIaYTIh3oavCvg][[test1][2]] ElasticsearchException[Elasticsearch exception [type=version_conflict_engine_exception, reason=[test1][4_1_92_2024-01-15 16:30:00]: version conflict, required seqNo [963], primary term [1]. current document has seqNo [964] and primary term [1]]]at org.elasticsearch.ElasticsearchException.innerFromXContent(ElasticsearchException.java:510)at org.elasticsearch.ElasticsearchException.fromXContent(ElasticsearchException.java:421)at org.elasticsearch.action.bulk.BulkItemResponse.fromXContent(BulkItemResponse.java:135)at org.elasticsearch.action.bulk.BulkResponse.fromXContent(BulkResponse.java:198)at org.elasticsearch.client.RestHighLevelClient.parseEntity(RestHighLevelClient.java:653)at org.elasticsearch.client.RestHighLevelClient.lambda$performRequestAsyncAndParseEntity$3(RestHighLevelClient.java:549)at org.elasticsearch.client.RestHighLevelClient$1.onSuccess(RestHighLevelClient.java:580)at org.elasticsearch.client.RestClient$FailureTrackingResponseListener.onSuccess(RestClient.java:621)at org.elasticsearch.client.RestClient$1.completed(RestClient.java:375)at org.elasticsearch.client.RestClient$1.completed(RestClient.java:366)at org.apache.http.concurrent.BasicFuture.completed(BasicFuture.java:122)at org.apache.http.impl.nio.client.DefaultClientExchangeHandlerImpl.responseCompleted(DefaultClientExchangeHandlerImpl.java:177)at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.processResponse(HttpAsyncRequestExecutor.java:436)at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.inputReady(HttpAsyncRequestExecutor.java:326)at org.apache.http.impl.nio.DefaultNHttpClientConnection.consumeInput(DefaultNHttpClientConnection.java:265)at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:81)at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:39)at org.apache.http.impl.nio.reactor.AbstractIODispatch.inputReady(AbstractIODispatch.java:114)at org.apache.http.impl.nio.reactor.BaseIOReactor.readable(BaseIOReactor.java:162)at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvent(AbstractIOReactor.java:337)at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvents(AbstractIOReactor.java:315)at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:276)at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104)at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:588)... 1 more[CIRCULAR REFERENCE:[test1/cfPTBYhcRIaYTIh3oavCvg][[test1][2]] ElasticsearchException[Elasticsearch exception [type=version_conflict_engine_exception, reason=[test1][4_1_92_2024-01-15 16:30:00]: version conflict, required seqNo [963], primary term [1]. current document has seqNo [964] and primary term [1]]]]

问题原因

Flink中存在八种分区策略,常用Operator Chain链接方式有三种分区器:

  • forward:上下游并行度相同,且不发生shuffle,直连的分区器
  • hash:将数据按照key的Hash值下发到下游的算子中
  • rebalance:数据会被循环或者随机下发到下游算子中,改变并行度若无keyby,默认使用RebalancePartitioner分区策略

rebalance分区器,可能会将上游算子的同一个key随机下发到下游不同算子中,因而引起报错,如下图:
在这里插入图片描述在这里插入图片描述模型如下:

在这里插入图片描述

解决方案

  • 分组聚合算子与sink2es算子配置成相同的并行度,即使用forward分区器,如下图:
    在这里插入图片描述在这里插入图片描述
    模型如下:
    ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/215b7ba208b142da803a78d85b0f474e.png

  • sink2es算子的并行度配置为1,如下图:

operator chain为forward分区器模型如下:
在这里插入图片描述

总结

归根结底就是需要保证:上游subtask中同一个key只能下发到下游一个subtask中

相关文章:

Flink问题解决及性能调优-【Flink不同并行度引起sink2es报错问题】

最近需求&#xff0c;仅想提高sink2es的qps&#xff0c;所以仅调节了sink2es的并行度&#xff0c;但在调节不同算子并行度时遇到一些问题&#xff0c;找出问题的根本原因解决问题&#xff0c;并分析整理。 实例代码 --SET table.exec.state.ttl86400s; --24 hour,默认: 0 ms …...

瑞_数据结构与算法_二叉搜索树

文章目录 1 什么是二叉搜索树1.1 二叉搜索树的特征1.2 前驱后继 2 二叉搜索树的Java实现2.1 定义二叉搜索树节点类BSTNode泛型key改进 2.2 实现查找方法get(int key)递归实现非递归实现 ★非递归实现 泛型key版本 2.3 实现查找最小方法min()递归实现非递归实现 ★ 2.4 实现查找…...

Linux 命令行访问名字中包含空格的文件或文件夹

Linux 命令行访问名字中包含空格的文件或文件夹 References 在 Windows 下命名文件或文件夹名有空格是可以的&#xff0c;甚至在 Windows 和 Ubuntu 虚拟机共享的文件中也可以这么做&#xff0c;但是在 Ubuntu 中空格要用下划线代替&#xff0c;养成好习惯。Linux 会把空格当成…...

Dart/Flutter工具模块:the_utils

Flutter笔记 Dart/Flutter工具模块&#xff1a;the_utils 作者&#xff1a;李俊才 &#xff08;jcLee95&#xff09;&#xff1a;https://blog.csdn.net/qq_28550263 邮箱 &#xff1a;291148484163.com 本文地址&#xff1a;https://blog.csdn.net/qq_28550263/article/detail…...

矩阵号:日入100+,八大提示词(Prompt)使用技巧

最近在搞头条矩阵&#xff0c;发现自己的指令写的太烂了&#xff0c;一个指令将会决定你的写作质量。 收益比较拉垮&#xff0c;50个号收益好的&#xff0c;也就这么几个号。 于是我扒了一些提示词的操作技巧&#xff0c;分享一下自己的学习心得。 先说理论知识&#xff0c;实…...

爬虫工作量由小到大的思维转变---<第三十九章 Scrapy-redis 常用的那个RetryMiddleware>

前言: 为什么要讲这个RetryMiddleware呢?因为他很重要~ 至少在你装配代理ip或者一切关于重试的时候需要用到!----最关键的是:大部分的教学视频里面,没有提及这个!!!! 正文: 源代码分析 这个RetryMiddleware是来自: from scrapy.downloadermiddlewares.retry import Retry…...

【MongoDB】mongodb安装及启动踩坑点

mongodb的安装&#xff0c;基本上参考文章[1]。 但是在过程中&#xff0c;有一些踩坑点。 1&#xff0c;高版本mongodb不自带mongo脚本 在文章1中&#xff0c;作者在解压后&#xff0c;直接使用了mongo脚本&#xff0c;而我下载的mongodb版本要更高&#xff0c;在解压后&…...

动态规划——采矿的小奇【集训笔记】

题目描述 假期小奇去采矿场体验生活&#xff0c;工头为每个员工发放了一个最多能装 M 公斤的背包&#xff0c;经过一天的辛苦小奇开采出了 n 块矿石&#xff0c;它们的重量分别是W1&#xff0c;W2&#xff0c;...,Wn,经过预估它们的价值分别为C1,C2,...,Cn&#xff0c;那么请你…...

wpf控件Expander集合下的像素滚动

项目场景&#xff1a;Expander集合滚动 如下图&#xff0c;有一个Expander集合&#xff0c;且设置 ScrollViewer.VerticalScrollBarVisibility "Auto" 每个Expaner下包含有若干元素&#xff0c;当打开Expader(即IsExpanded "true"&#xff09;时&#…...

docker 基础手册

文章目录 docker 基础手册docker 容器技术镜像与容器容器与虚拟机docker 引擎docker 架构docker 底层技术docker 二进制安装docker 镜像加速docker 相关链接docker 生态 docker 基础手册 docker 容器技术 开源的容器项目&#xff0c;使用 Go 语言开发原意“码头工人”&#x…...

记一次SPI机制导致的BUG定位【不支持:http://javax.xml.XMLConstants/property/accessExternalDTD】

1、前因 今天在生产环境启用了某个功能&#xff0c;结果发现有个文件上传华为云OBS失败了&#xff0c;报错如下&#xff1a; Caused by: java.lang.IllegalArgumentException: 不支持&#xff1a;http://javax.xml.XMLConstants/property/accessExternalDTDat org.apache.xal…...

Kali如何启动SSH服务并实现无公网ip环境远程连接

文章目录 1. 启动kali ssh 服务2. kali 安装cpolar 内网穿透3. 配置kali ssh公网地址4. 远程连接5. 固定连接SSH公网地址6. SSH固定地址连接测试 简单几步通过[cpolar 内网穿透](cpolar官网-安全的内网穿透工具 | 无需公网ip | 远程访问 | 搭建网站)软件实现ssh 远程连接kali! …...

谷粒商城配置虚拟机

一、创建虚拟机 之前有在VM里面建一个ubuntu的虚拟机&#xff0c;准备拿来直接用&#xff0c;网络设置为NAT模式&#xff0c;查看我的虚拟机是虚拟机&#xff1a;192.168.248.128 主机&#xff1a; 192.168.2.12。可以互相ping通。 二、linux安装docker Docker docker是虚拟…...

Java中文乱码浅析及解决方案

Java中文乱码浅析及解决方案 一、GBK和UTF-8编码方式二、idea和eclipse的默认编码方式三、解码和编码方法四、代码实现编码解码 五、额外知识扩展 一、GBK和UTF-8编码方式 如果采用的是UTF-8的编码方式&#xff0c;那么1个英文字母 占 1个字节&#xff0c;1个中文占3个字节如果…...

【前端基础--3】

文字样式 1.文字颜色 color 取值方式&#xff1a; 英文单词 red green blue十六进制的颜色值 #000000 也可以写为#000&#xff08;如aabbcc可以简写为abc&#xff09;rgb三原色取值 color&#xff1a;rgb(220,32,215) 取值范围都在0~255之间 2.文字大小 font-size …...

Obsidian笔记软件结合cpolar实现安卓移动端远程本地群晖WebDAV数据同步

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…...

51单片机电子密码锁Proteus仿真+程序+视频+报告

目录 视频 设计分析 系统结构 仿真图 资料内容 资料下载地址&#xff1a;51单片机电子密码锁Proteus仿真程序视频报告 视频 单片机电子密码锁Proteus仿真程序视频 设计分析 (1)能够从键盘中输入密码&#xff0c;并相应地在显示器上显示‘*’&#xff1b; (2)能够判断密码…...

[BSidesCF 2020]Had a bad day

先看url&#xff0c;发现可能有注入 http://655c742e-b427-485c-9e15-20a1e7ef1717.node5.buuoj.cn:81/index.php?categorywoofers 试试能不能查看index.php直接?categoryindex.php不行&#xff0c;试试伪协议 把.php去掉试试 base64解码 <?php$file $_GET[category];…...

[笔记]事务简介-springboot

在Spring Boot中&#xff0c;事务的管理通常通过注解来实现&#xff0c;使得配置变得简单而直观。这种方式与Spring Boot的设计理念一致&#xff0c;即减少显式配置&#xff0c;增加自动配置。以下是如何在Spring Boot项目中应用和管理事务的详细说明&#xff1a; Spring Boot中…...

初识计算机网络 | 计算机网络的发展 | 协议初识

1.计算机网络的发展 “矛盾是普遍存在的&#xff0c;矛盾是事物联系的实质内容和事物发展的根本动力&#xff01;” 计算机在诞生之初&#xff0c;在军事上用来计算导弹的弹道轨迹&#xff01;在发展的过程中&#xff08;商业的推动&#xff0c;国家政策推动&#xff09;&…...

地震勘探——干扰波识别、井中地震时距曲线特点

目录 干扰波识别反射波地震勘探的干扰波 井中地震时距曲线特点 干扰波识别 有效波&#xff1a;可以用来解决所提出的地质任务的波&#xff1b;干扰波&#xff1a;所有妨碍辨认、追踪有效波的其他波。 地震勘探中&#xff0c;有效波和干扰波是相对的。例如&#xff0c;在反射波…...

渲染学进阶内容——模型

最近在写模组的时候发现渲染器里面离不开模型的定义,在渲染的第二篇文章中简单的讲解了一下关于模型部分的内容,其实不管是方块还是方块实体,都离不开模型的内容 🧱 一、CubeListBuilder 功能解析 CubeListBuilder 是 Minecraft Java 版模型系统的核心构建器,用于动态创…...

Redis数据倾斜问题解决

Redis 数据倾斜问题解析与解决方案 什么是 Redis 数据倾斜 Redis 数据倾斜指的是在 Redis 集群中&#xff0c;部分节点存储的数据量或访问量远高于其他节点&#xff0c;导致这些节点负载过高&#xff0c;影响整体性能。 数据倾斜的主要表现 部分节点内存使用率远高于其他节…...

AspectJ 在 Android 中的完整使用指南

一、环境配置&#xff08;Gradle 7.0 适配&#xff09; 1. 项目级 build.gradle // 注意&#xff1a;沪江插件已停更&#xff0c;推荐官方兼容方案 buildscript {dependencies {classpath org.aspectj:aspectjtools:1.9.9.1 // AspectJ 工具} } 2. 模块级 build.gradle plu…...

算法岗面试经验分享-大模型篇

文章目录 A 基础语言模型A.1 TransformerA.2 Bert B 大语言模型结构B.1 GPTB.2 LLamaB.3 ChatGLMB.4 Qwen C 大语言模型微调C.1 Fine-tuningC.2 Adapter-tuningC.3 Prefix-tuningC.4 P-tuningC.5 LoRA A 基础语言模型 A.1 Transformer &#xff08;1&#xff09;资源 论文&a…...

GruntJS-前端自动化任务运行器从入门到实战

Grunt 完全指南&#xff1a;从入门到实战 一、Grunt 是什么&#xff1f; Grunt是一个基于 Node.js 的前端自动化任务运行器&#xff0c;主要用于自动化执行项目开发中重复性高的任务&#xff0c;例如文件压缩、代码编译、语法检查、单元测试、文件合并等。通过配置简洁的任务…...

【Elasticsearch】Elasticsearch 在大数据生态圈的地位 实践经验

Elasticsearch 在大数据生态圈的地位 & 实践经验 1.Elasticsearch 的优势1.1 Elasticsearch 解决的核心问题1.1.1 传统方案的短板1.1.2 Elasticsearch 的解决方案 1.2 与大数据组件的对比优势1.3 关键优势技术支撑1.4 Elasticsearch 的竞品1.4.1 全文搜索领域1.4.2 日志分析…...

小木的算法日记-多叉树的递归/层序遍历

&#x1f332; 从二叉树到森林&#xff1a;一文彻底搞懂多叉树遍历的艺术 &#x1f680; 引言 你好&#xff0c;未来的算法大神&#xff01; 在数据结构的世界里&#xff0c;“树”无疑是最核心、最迷人的概念之一。我们中的大多数人都是从 二叉树 开始入门的&#xff0c;它…...

rknn toolkit2搭建和推理

安装Miniconda Miniconda - Anaconda Miniconda 选择一个 新的 版本 &#xff0c;不用和RKNN的python版本保持一致 使用 ./xxx.sh进行安装 下面配置一下载源 # 清华大学源&#xff08;最常用&#xff09; conda config --add channels https://mirrors.tuna.tsinghua.edu.cn…...

加密通信 + 行为分析:运营商行业安全防御体系重构

在数字经济蓬勃发展的时代&#xff0c;运营商作为信息通信网络的核心枢纽&#xff0c;承载着海量用户数据与关键业务传输&#xff0c;其安全防御体系的可靠性直接关乎国家安全、社会稳定与企业发展。随着网络攻击手段的不断升级&#xff0c;传统安全防护体系逐渐暴露出局限性&a…...