Flink SQL 基于Update流出现空值无法过滤问题

问题背景

  • 问题描述
基于Flink-CDC ，Flink SQL的实时计算作业在运行一段时间后，突然发现插入数据库的计算结果发生部分主键属性发生失败，导致后续计算结果无法插入，
超过失败次数失败的情况
  • 问题报错
	Caused by: java.sql.BatchUpdateException: Batch entry 0 INSERT INTO dm_hljy.dws_table_name (op_date, school_year, campus_name, school_name, depart_name, total_opfare, ids, update_time) VALUES ('2024-03-11 00:00:00+08', '2023', 'xxxx', 'xxxx学校', 'xxxx小学部', '203333300000', '57', '2024-03-21 09:31:08.47+08') ON DUPLICATE KEY UPDATE school_year=VALUES(school_year), total_opfare=VALUES(total_opfare), ids=VALUES(ids), update_time=VALUES(update_time) was aborted: ERROR: dn_6007_6008: null value in column "depart_name" violates not-null constraint  Call getNextException to see other errors in the batch.at com.huawei.gauss200.jdbc.jdbc.BatchResultHandler.handleCompletion(BatchResultHandler.java:171) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.core.v3.QueryExecutorImpl.executeBatch(QueryExecutorImpl.java:586) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.jdbc.PgStatement.executeBatch(PgStatement.java:883) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1580) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.executeBatch(FieldNamedPreparedStatementImpl.java:65) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor.executeBatch(TableSimpleStatementExecutor.java:64) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.executor.TableBufferReducedStatementExecutor.executeBatch(TableBufferReducedStatementExecutor.java:101) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.attemptFlush(JdbcOutputFormat.java:266) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.flush(JdbcOutputFormat.java:236) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.lambda$open$0(JdbcOutputFormat.java:159) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_332]at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_332]at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_332]at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_332]at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_332]at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_332]... 1 moreCaused by: com.huawei.gauss200.jdbc.util.PSQLException: ERROR: dn_6007_6008: null value in column "depart_name" violates not-null constraintat com.huawei.gauss200.jdbc.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2856) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2587) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.core.v3.QueryExecutorImpl.executeBatch(QueryExecutorImpl.java:575) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.jdbc.PgStatement.executeBatch(PgStatement.java:883) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at com.huawei.gauss200.jdbc.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1580) ~[huaweicloud-dws-jdbc-8.1.1.1-200.jar:?]at org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.executeBatch(FieldNamedPreparedStatementImpl.java:65) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor.executeBatch(TableSimpleStatementExecutor.java:64) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.executor.TableBufferReducedStatementExecutor.executeBatch(TableBufferReducedStatementExecutor.java:101) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.attemptFlush(JdbcOutputFormat.java:266) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.flush(JdbcOutputFormat.java:236) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.lambda$open$0(JdbcOutputFormat.java:159) ~[flink-connector-jdbc-1.15.0-h0.cbu.mrs.320.r33.jar:1.15.0-h0.cbu.mrs.320.r33]at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_332]at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_332]at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_332]at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_332]at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_332]at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_332

在这里插入图片描述

定位

定位思路

1.方向一：怀疑数据库插入存在数据处理时，造成数据处理出现空值的情况，即数据本身不为空，但是数据插入却出现了空
2.方向二：Flink-SQL在消费kafka数据时存在了空值，故加工的数据计算结果存在空值

定位过程

  • 因插入数据库定位比较麻烦，且数据库已经设置该字段为主属性，故出现插入时处理为空值的概率较小。故先从较为简单的Flink SQL查询数据
  • 定位方法一，查询该字段为空的记录,待作业执行完成后，未查询到空值对应记录
 select  select * from table_name where depart_name is null or depart_name = '' or char_length(depart_name) = 0;
  • 因考虑到使用Flink-CDC进行变更数据捕获，故对应的update流存在-U,+U,-D,+I记录，因此随着插入记录存在空值被记录进去的情况，故采用view的方式，先将宽表的加工、关联方式创建为view，然后进行空值的过滤。实施如下
create view view_prd as 
select a.* ,b.*  from a join b on a.id = b.idselect * from view_prd where depart_name is null or depart_name = '' or char_length(depart_name) = 0;
  • 通过查询结果，发现存在最后一条记录存在空值的原因，往源头定位，发现该字段之前为空，后面进行更新填充到值出现-U记录，导致数据插入持续失败
    在这里插入图片描述

原因

  • 因为flink-SQL消费的数据时kafka topic，flink以upsert-kafka形式的connector进行写入，故存在changelog 流中数据更新存在-U，+U的记录（按照Key进行区分唯一条记录），value 为空(-U)的记录kafka也，导致出现空值，
    在这里插入图片描述

解决

通过在DWS宽表创建一层View（如上)，在写入DWS宽表的kafka topic之前，现将该字段空值过滤，即可排除空值涉及记录被纳入结果指标计算的范围中

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/782628.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

智慧公厕：让公共厕所变得更智能、更卫生、更舒适的解决方案

智慧公厕：让公共厕所变得更智能、更卫生、更舒适的解决方案

近年来，随着城市发展的不断壮大，公共设施的建设也越来越受到重视。而公共厕所作为城市基础设施的一部分，是城市文明程度的重要体现。然而，传统的公共厕所在使用、运行、管理、养护等方面存在诸多问题，严重影响了市民的…
阅读更多...
特征选择集大成的包-arfs（python）

特征选择集大成的包-arfs（python）

特征选择集大成的包-arfs（python） 一、介绍 arfs介绍文档https://arfs.readthedocs.io/en/latest/Introduction.html 英文好的朋友可以阅读作者写的介绍： All relevant feature selection means trying to find all features carrying info…
阅读更多...
YOLOv5改进系列：升级版ResNet的新主干网络DenseNet

YOLOv5改进系列：升级版ResNet的新主干网络DenseNet

一、论文理论 论文地址：Densely Connected Convolutional Networks 1.理论思想 DenseNet最大化前后层信息交流，通过建立前面所有层与后面层的密集连接，实现了特征在通道维度上的复用，不但减缓了梯度消失的现象，也使其…
阅读更多...
【二分图】【二分图最大匹配】LCP 04. 覆盖

【二分图】【二分图最大匹配】LCP 04. 覆盖

作者推荐 视频算法专题 本文涉及知识点 二分图 二分图最大匹配 LeetCode LCP 04. 覆盖 你有一块棋盘，棋盘上有一些格子已经坏掉了。你还有无穷块大小为1 * 2的多米诺骨牌，你想把这些骨牌不重叠地覆盖在完好的格子上，请找出你最多能在棋盘…
阅读更多...
2024年京东云主机租用价格_京东云服务器优惠价格表

2024年京东云主机租用价格_京东云服务器优惠价格表

2024年京东云服务器优惠价格表，轻量云主机优惠价格5.8元1个月、轻量云主机2C2G3M价格50元一年、196元三年，2C4G5M轻量云主机165元一年，4核8G5M云主机880元一年，游戏联机服务器4C16G配置26元1个月、4C32G价格65元1个月、8核32G费用…
阅读更多...
新书速递——《可解释AI实战（PyTorch版）》

新书速递——《可解释AI实战（PyTorch版）》

本书旨在帮助你实施最新的可解释AI技术，以构建公平且可解释的AI系统。可解释AI是当今AI研究中的热门话题，但只有少数资源和指南涵盖了所有重要技术，这些技术对实践者来说非常有价值。本书旨在填补这一空白。 本书读者对象 本书既适合那些有兴…
阅读更多...
GIt的原理和使用（五）：模拟多人协作的两种情况

GIt的原理和使用（五）：模拟多人协作的两种情况

目录 多人协作 多人协作一 准备工作 协作开发 多人协作二 准备工作 额外场景 申请单合并分支 更推荐写法 远程分支删除后，本地git branch -a依然能看到的解决办法 多人协作 多人协作一 目标：在远程master分支下的file.txt文件新增代码“aaa”…
阅读更多...
鸿蒙OS开发实例：【窥探网络请求】

鸿蒙OS开发实例：【窥探网络请求】

HarmonyOS 平台中使用网络请求，需要引入 "ohos.net.http", 并且需要在 module.json5 文件中申请网络权限, 即 “ohos.permission.INTERNET” 本篇文章将尝试使用 ohos.net.http 来实现网络请求 场景设定 WeiBo UniDemo HuaWei : 请求顺序WeiBo1 UniDem…
阅读更多...
华为数通 HCIP-Datacom H12-831 题库补充（3/27）

华为数通 HCIP-Datacom H12-831 题库补充（3/27）

2024年 HCIP-Datacom（H12-831）最新题库，完整题库请扫描上方二维码，持续更新。 如图所示，关于R4路由器通过IS-IS计算出来的IPv6路由，哪一选项的描述是错误的？ A：R4通过IS—IS只学习到…
阅读更多...
基于SpringBoot的“校园台球厅人员与设备管理系统”的设计与实现（源码+数据库+文档+PPT)

基于SpringBoot的“校园台球厅人员与设备管理系统”的设计与实现（源码+数据库+文档+PPT)

基于SpringBoot的“校园台球厅人员与设备管理系统”的设计与实现（源码数据库文档PPT) 开发语言：Java 数据库：MySQL 技术：SpringBoot 工具：IDEA/Ecilpse、Navicat、Maven 系统展示 系统功能结构图 系统首页界面图…
阅读更多...
PCL拟合并绘制平面(二)

PCL拟合并绘制平面(二)

使用RANSAC拟合点云平面 1、C实现2、效果图 普通的点云平面拟合方式在一般情况下可以得到较好的平面拟合效果，但是容易出现平面拟合错误或是拟合的平面不是最优的情况。此时就需要根据自己的实际使用情况，调整平面拟合的迭代次数以及收敛条件。 使用RAN…
阅读更多...
亚马逊测评新策略：解决底层环境防关联，提升下单成功率

亚马逊测评新策略：解决底层环境防关联，提升下单成功率

对于做测评的环境系统，确保稳定性和成功率是非常重要的。市面上有各种环境方案，如虚拟机、模拟机、gcs、云手机、VPS等。然而，这些方案不仅成本高，而且成功率很低。因此，一个好的环境系统是成功的基础。 亚马逊平台的…
阅读更多...
记录rocketMQ5.+启动报错解决过程

记录rocketMQ5.+启动报错解决过程

1.根据官方文档指引下载对应的rocketMQ源码包，上传到服务器解压 2. 启动NameServer nohup sh bin/mqnamesrv & 验证namesrv是否启动成功 tail -f ~/logs/rocketmqlogs/namesrv.log The Name Server boot success… 3.启动BrokerProxy nohup sh bin/mqbroker -n …
阅读更多...
第3章.引导ChatGPT精准角色扮演：高效输出专业内容

第3章.引导ChatGPT精准角色扮演：高效输出专业内容

角色提示技术 角色提示技术（role prompting technique），是通过模型扮演特定角色来产出文本的一种方法。用户为模型设定一个明确的角色，它就能更精准地生成符合特定上下文或听众需求的内容。 比如，想生成客户服务的回复…
阅读更多...
Java作业3-字符串

Java作业3-字符串

题目一 代码 import java.util.*; public class Main {public static void main(String[] args) {Scanner input new Scanner( System.in );String str input.nextLine();int len str.length();StringBuilder s new StringBuilder(len);//StringBuilder类参考菜鸟教程for…
阅读更多...
深入理解HDFS工作原理：大数据存储和容错性机制解析

深入理解HDFS工作原理：大数据存储和容错性机制解析

** 引言： ** 在当今数据爆炸的时代，存储和管理大规模数据成为了许多组织面临的重要挑战。为了解决这一挑战，分布式文件系统应运而生。Hadoop分布式文件系统（HDFS）作为Apache Hadoop生态系统的核心组件之一&#xff…
阅读更多...
是德科技keysight N9000B 信号分析仪

是德科技keysight N9000B 信号分析仪

181/2461/8938产品概述： 工程的内涵就是将各种创意有机地联系起来，并解决遇到的问题。 CXA 信号分析仪具有出色的实际性能，它是一款出类拔萃、经济高效的基本信号表征工具。 它的功能十分强大，为一般用途和教育行业的用户执行测试…
阅读更多...
深入探讨分布式ID生成方案

深入探讨分布式ID生成方案

✨✨谢谢大家捧场，祝屏幕前的小伙伴们每天都有好运相伴左右，一定要天天开心哦！✨✨ 🎈🎈作者主页： 喔的嘛呀🎈🎈 ✨✨ 帅哥美女们，我们共同加油！一起进步&am…
阅读更多...
vue3数据库中存头像图片相对路径在前端用prop只能显示路径或无法显示图片只能显示alt中内容的问题的解决

vue3数据库中存头像图片相对路径在前端用prop只能显示路径或无法显示图片只能显示alt中内容的问题的解决

不想看前情可以直接跳到头像部分代码 前情： 首先我们是在数据库中存图片相对路径，这里我们是在vue的src下的assets专门建一个文件夹img存头像图片。 然后我们如果用prop"avatar" label"头像"是只能显示图片路径的，即lo…
阅读更多...
java数组与集合框架（一） -- 数据结构，数组

java数组与集合框架（一） -- 数据结构，数组

数据结构 概述 为什么要讲数据结构？ 任何一个有志于从事IT领域的人员来说，数据结构（Data Structure）是一门和计算机硬件与软件都密切相关的学科，它的研究重点是在计算机的程序设计领域中探讨如何在计算机中组织和存储…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023