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.rhkb.cn/news/292382.html

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

相关文章

（南京观海微电子）——GOA介绍

（南京观海微电子）——GOA介绍

GOA是Gate on Array的简写，简单可以理解为gate IC集成在玻璃上了，面板就可以不用gate ic了，是一种低成本的设计，窄边框面板大多数都用了GOA技术。还有一些公司叫GIP（Gate in Panel），GDM等等。 …
阅读更多...
软考数据库

软考数据库

目录 分值分布1. 事务管理1.1 事物的基本概念1.2 数据库的并发控制1.2.1 事务调度概念1.2.2 并发操作带来的问题1.2.3 并发控制技术1.2.4 隔离级别： 1.3 数据库的备份和恢复1.3.1 故障种类1.3.2 备份方法1.3.3 日志文件1.3.4 恢复 SQL语言发权限收权限视图触发器创建…
阅读更多...
AJAX（二）：axios 和 fetch函数发送AJAX请求、同源策略、 jsonp、CORS

AJAX（二）：axios 和 fetch函数发送AJAX请求、同源策略、 jsonp、CORS

一、各种发送AJAX请求 jquery基于回调函数，axios基于promise 1.axios发送AJAX请求!!! axios (v1.5.0) - Axios 是一个基于 promise 的 HTTP 库,可以用在浏览器和 Node.js 中。 | BootCDN - Bootstrap 中文网开源项目免费 CDN 加速服务 服务器： app.…
阅读更多...
python-pytorch获取FashionMNIST实际图片标签数据集

python-pytorch获取FashionMNIST实际图片标签数据集

在查看pytorch官方文档的时候，在这里链接中https://pytorch.org/tutorials/beginner/basics/data_tutorial.html的Creating a Custom Dataset for your files章节，有提到要自定义数据集，需要用到实际的图片和标签。 在网上找了半天没找到&a…
阅读更多...
K8S命令行可视化实验

K8S命令行可视化实验

以下为K8s命令行可视化工具的实验内容，相比于直接使用命令行，可视化工具可能更直观、更易于操作。 Lens Lens是用于监控和调试的K8S IDE。可以在Windows、Linux以及Mac桌面上完美运行。在 Kubernetes 上： 托管地址：github/lensa…
阅读更多...
物联网学习1、什么是 MQTT？

物联网学习1、什么是 MQTT？

MQTT（Message Queuing Telemetry Transport）是一种轻量级、基于发布-订阅模式的消息传输协议，适用于资源受限的设备和低带宽、高延迟或不稳定的网络环境。它在物联网应用中广受欢迎，能够实现传感器、执行器和其它设备之间的高效通…
阅读更多...
3D数据格式导出工具HOOPS Publish如何生成高质量3D PDF？

3D数据格式导出工具HOOPS Publish如何生成高质量3D PDF？

在当今数字化时代，从建筑设计到制造业，从医学领域到电子游戏开发，3D技术已经成为了不可或缺的一部分。在这个进程中，将3D模型导出为3D PDF格式具有重要的意义。同时，HOOPS Publish作为一个领先的解决方案，为…
阅读更多...
【搜索引擎2】实现API方式调用ElasticSearch8接口

【搜索引擎2】实现API方式调用ElasticSearch8接口

1、理解ElasticSearch各名词含义 ElasticSearch对比Mysql Mysql数据库Elastic SearchDatabase7.X版本前有Type，对比数据库中的表，新版取消了TableIndexRowDocumentColumnmapping Elasticsearch是使用Java开发的，8.1版本的ES需要JDK17及以上…
阅读更多...
从0开始搭建基于VUE的前端项目(一) 项目创建和配置

从0开始搭建基于VUE的前端项目(一) 项目创建和配置

准备与版本 安装nodejs(v20.11.1)安装vue脚手架(@vue/cli 5.0.8) ,参考（https://cli.vuejs.org/zh/）vue版本（2.7.16），vue2的最后一个版本vue.config.js的配置详解(https://cli.vuejs.org/zh/config/)element-ui(2.15.14)(https://element.eleme.io/)vuex(3.6.2) (https://…
阅读更多...
Python 基于 OpenCV 视觉图像处理实战 之 OpenCV 简单实战案例 之八 简单水彩画效果

Python 基于 OpenCV 视觉图像处理实战 之 OpenCV 简单实战案例 之八 简单水彩画效果

Python 基于 OpenCV 视觉图像处理实战 之 OpenCV 简单实战案例 之八 简单水彩画效果 目录 Python 基于 OpenCV 视觉图像处理实战 之 OpenCV 简单实战案例 之八 简单水彩画效果 一、简单介绍 二、简单图像浮雕效果实现原理 三、简单水彩画效果案例实现简单步骤 四、注意事项…
阅读更多...
Spring IoCDI(3)

Spring IoCDI(3)

DI详解 接下来学习一下依赖注入DI的细节. 依赖注入是一个过程, 是指IoC容器在创建Bean时, 去提供运行时所依赖的资源, 而资源指的就是对象. 在之前的案例中, 使用了Autowired这个注解, 完成了依赖注入这个操作. 简单来说, 就是把对象取出来放到某个类的属性中. 在一些文章中…
阅读更多...
Matlab将日尺度数据转化为月尺度数据

Matlab将日尺度数据转化为月尺度数据

日尺度转化为月尺度 clcclear all% load datadata xlread(data.xlsx) % 例如该数据为1961-01-01至2022-12-31，共计22645天data data(:,1:3) % 该数据有22645行，数据分别为降水，气温，湿度等三列dt datetime(1961-01-01):datatim…
阅读更多...
从MVC 到DDD 架构

从MVC 到DDD 架构

目录 一、前言 二、MVC架构 三、DDD架构 四、我为什么会使用DDD？ 五、DDD架构分层 一、前言 最近在做一个项目，使用的是DDD架构思，觉得很不错，在此记录下。 二、MVC架构 MVC是一种经典的软件架构模式，主要用于…
阅读更多...
关于未来自我的发展和一些学习方法（嵌入式方向）

关于未来自我的发展和一些学习方法（嵌入式方向）

我是一名大二的学生，考研还是就业，到底是重视专业课还是重视数学英语，这些问题一直困扰了我很久，但如今已经有了一些浅显的认识，所以才会想写这样一篇文章来记录一下自己的状态和未来的规划 下面的看法都是个人的看法&…
阅读更多...
开源一款剪贴板跨设备共享工具

开源一款剪贴板跨设备共享工具

shigen坚持更新文章的博客写手，擅长Java、python、vue、shell等编程语言和各种应用程序、脚本的开发。记录成长，分享认知，留住感动。 个人IP：shigen 昨天遇到了一个很棘手的问题：我的手机剪贴板需要同步到另外的两台设…
阅读更多...
基于Arduino IDE 野火ESP8266模块 一键配网 的开发

基于Arduino IDE 野火ESP8266模块 一键配网 的开发

一、配网介绍 ESP8266 一键配网（也称为 SmartConfig 或 FastConfig）是一种允许用户通过智能手机上的应用程序快速配置 ESP8266 Wi-Fi 模块的方法，而无需手动输入 SSID 和密码。为了实现这一功能，则需要一个支持 SmartConfig 的智能…
阅读更多...
基于java实现学科竞赛管理系统【Springboot+mybatis+layui】

基于java实现学科竞赛管理系统【Springboot+mybatis+layui】

基于java实现学科竞赛管理系统【Springbootmybatislayui】 博主介绍：多年java开发经验，专注Java开发、定制、远程、文档编写指导等,csdn特邀作者、专注于Java技术领域 作者主页 央顺技术团队 Java毕设项目精品实战案例《1000套》 欢迎点赞 收藏 ⭐留言 文…
阅读更多...
Ribbon简介

Ribbon简介

目录 一 、概念介绍 1、Ribbon是什么 2、认识负载均衡 2.1 服务器端的负载均衡 2.2 客户端的负载均衡 3、Ribbon工作原理 4、Ribbon的主要组件 IClientConfig ServerList ServerListFilter IRule Iping ILoadBalancer ServerListUpdater 5、Ribbon支持…
阅读更多...
AndroidStudio出现类似 Could not create task ‘:app:ToolOperatorDemo.main()‘. 错误

AndroidStudio出现类似 Could not create task ‘:app:ToolOperatorDemo.main()‘. 错误

先看我们的报错 翻译过来大概意思是:无法创建任务:app:ToolOperatorDemo.main()。 没有找到名称为“main”的源集。 解决方法&#xff1a; 在.idea文件夹下的gradle.xml文件中 <GradleProjectSettings>标签下添加<option name"delegatedBuild" value"f…
阅读更多...
华为OD机试 - 芯片资源限制（Java 2024 C卷 100分）

华为OD机试 - 芯片资源限制（Java 2024 C卷 100分）

华为OD机试 2024C卷题库疯狂收录中&#xff0c;刷题点这里 专栏导读 本专栏收录于《华为OD机试&#xff08;JAVA&#xff09;真题&#xff08;A卷B卷C卷&#xff09;》。 刷的越多&#xff0c;抽中的概率越大&#xff0c;每一题都有详细的答题思路、详细的代码注释、样例测试…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023