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

Flink消费kafka出现空指针异常

文章目录

      • 出现场景:
      • 表现:
      • 问题:
      • 解决:

tombstone : Kafka中提供了一个墓碑消息(tombstone)的概念,如果一条消息的key不为null,但是其value为null,那么此消息就是墓碑消息.

出现场景:

双流join时,采用的是left join的方式,众所周知该方式会产生回撤流,下游kafka连接器使用的是upsert-kafka,在产生回撤流时,kafka会删除未join上的消息,填充join后的消息进去。

表现:

在这里插入图片描述

问题:

此时消费该topic的flink程序会出现,空指针异常

DataStream Api会出现,Table Api 未发现

解决:

自定义kafka反序列化器过滤Null值,flink1.14.4
代码:

public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("").setTopics("test").setGroupId("gid").setStartingOffsets(OffsetsInitializer.earliest()).setValueOnlyDeserializer(new MySimpleStringSchema()).setProperty("auto.offset.commit", "false").build();DataStreamSource<String> kfkDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kfk");kfkDs.print();env.execute();}// 自定义反序列化器static class MySimpleStringSchema implements DeserializationSchema<String>, SerializationSchema<String>{@Overridepublic String deserialize(byte[] message) {if (message != null) return new String(message, StandardCharsets.UTF_8);else{return deserialize(new byte[1]); // 返回空 不是Null}}@Overridepublic boolean isEndOfStream(String nextElement) {return false;}@Overridepublic byte[] serialize(String element) {return element.getBytes(StandardCharsets.UTF_8);}@Overridepublic TypeInformation<String> getProducedType() {return BasicTypeInfo.STRING_TYPE_INFO;}}

相关文章:

Flink消费kafka出现空指针异常

文章目录 出现场景&#xff1a;表现&#xff1a;问题&#xff1a;解决&#xff1a; tombstone : Kafka中提供了一个墓碑消息&#xff08;tombstone&#xff09;的概念&#xff0c;如果一条消息的key不为null&#xff0c;但是其value为null&#xff0c;那么此消息就是墓碑消息. …...

【探索 Kubernetes|作业管理篇 系列 9】Pod 的服务对象

前言 大家好&#xff0c;我是秋意零。 在上一篇中&#xff0c;我们介绍了 Pod 的生命周期以及区分 Pod 字段的层次级别&#xff0c;相信你对此有了充分的认识。 今天&#xff0c;我们还会接着以 Pod 展开&#xff0c;说说它的 “服务对象”&#xff0c;一听就知道是对 Pod 提…...

多种拖拽= =自用留档

<template> <div class"main-drag"> <div v-if"stencil 0" class"mapped-fields"> <el-form ref"mapped" :model"mapped" class"demo-fieldsForm"> <el-form-item label"切换数…...

贝叶斯与认知——读《贝叶斯的博弈》有感

关于对贝叶斯与认知问题的相关思考 一、贝叶斯定理二、贝叶斯与认知的本质三、经验的偏见四、总结 自古以来&#xff0c;人们就在思考知识来自何处&#xff0c;“冯翼惟象&#xff0c;何以识之&#xff1f;”&#xff0c;对此的思考逐渐发展成哲学的认识论分支。德国哲学家康德…...

MySQL安装失败starting the sever

MySQL安装失败starting the sever 如果电脑是第一次安装MySQL&#xff0c;一般不会出现这样的报错。starting the sever失败&#xff0c;通常是因为上次安装该软件没有清除干净。 第一种解决方法&#xff1a;完全卸载mysql&#xff0c;重新安装 完全卸载该软件的办法&#…...

合并文件夹中所有文件,并输出重复的条形码值

文章目录 一、需求二、处理方式三、代码实现 一、需求 每天会生成一个记录文件&#xff08;文件名按日期yyyyMMdd格式命名&#xff09;&#xff0c;记录文件中记录有条形码的内容&#xff0c;需要合并最近20次的数据&#xff0c;并提取出有重复的条形码。 也可以进行最近30天数…...

P3089 [USACO13NOV] Pogo-Cow S 弹簧踩高跷

P3089 [USACO13NOV] Pogo-Cow S 弹簧踩高跷 洛谷题目传送门 文章目录 P3089 [USACO13NOV] Pogo-Cow S 弹簧踩高跷题目描述输入格式输出格式样例 #1样例输入 #1样例输出 #1 提示题目大意方法一&#xff08;线段树维护dp&#xff09;code 方法二 &#xff08;单调队列维护dp&…...

计算机网络 - 第一章(下)

1.2_1 分层结构、协议、接口、服务_哔哩哔哩_bilibili1.2_1 分层结构、协议、接口、服务是王道计算机考研 计算机网络的第7集视频&#xff0c;该合集共计76集&#xff0c;视频收藏或关注UP主&#xff0c;及时了解更多相关视频内容。https://www.bilibili.com/video/BV19E411D78…...

【Uniapp】小程序携带Token请求接口+无感知登录方案2.0

本次改进原文《【Uniapp】小程序携带Token请求接口无感知登录方案》&#xff0c;在实际使用过程中我发现以下bug&#xff1a; 若token恰好在用户访问接口时到期&#xff0c;就会直接查询为空&#xff0c;不反映token过期问题&#xff08;例如&#xff1a;弹窗显示订单查询记录…...

Ubuntu常用命令

文章目录 1&#xff1a;文件管理2&#xff1a;文档编辑3&#xff1a;系统管理4&#xff1a;磁盘管理5&#xff1a;文件传输6&#xff1a;网络通讯7&#xff1a;设备管理8&#xff1a;备份压缩9&#xff1a;其他命令扩展&#xff1a;知识干货 1&#xff1a;文件管理 ls命令 –…...

ERP重构-SLA子分类账-分布式实现方案

背景 ERP中的GL总账模块&#xff0c;明细数据来源于各个业务模块如库存、成本、应收、应付、费控、资产等&#xff0c;统称为子模块&#xff0c;生成的账叫做子分类账。然而记账的业务逻辑各式各样&#xff0c;但是最终输出都是来源、类型、期间、科目、借贷金额等等关键信息。…...

IP路由协议(RIP、IGRP、OSPF、IS-IS、BGP)

文章目录 1、路由分类2、RIP协议1&#xff09;RIP的工作原理2&#xff09;RIP路由表的更新过程3&#xff09;RIP路由表的更新原则4&#xff09;RIP的特性5&#xff09;RIP协议的版本 4、IGRP协议1&#xff09;IGRP路由表的更新2&#xff09;IGRP的度量标准 5、OSPF协议1&#x…...

互斥锁、自旋锁、读写锁、悲观锁、乐观锁的应用场景

多线程访问共享资源的时候&#xff0c;避免不了资源竞争而导致数据错乱的问题&#xff0c;所以我们通常为了解决这一问题&#xff0c;都会在访问共享资源之前加锁。 最常用的就是互斥锁&#xff0c;当然还有很多种不同的锁&#xff0c;比如自旋锁、读写锁、乐观锁等&#xff0…...

Python WSGI 与 Web 开发框架

目录 文章目录 目录WSGIWSGI 的工作原理environ 参数start_resposne 参数 WSGI 的中间件 WSGI Web 开发框架OpenStack 中的应用案例进程入口WSGI Application 加载Paste/PasteDeployRoutesWebOb WSGI Server 启动 WSGI WSGI&#xff08;Web Server Gateway Interface&#xff…...

[洛谷]P6464 [传智杯 #2 决赛] 传送门

看到数据范围&#xff1a;n<100&#xff0c;嗯......脑子闪过&#xff1a;还在想什么呢&#xff01;Floyd啊。哈哈哈 思路&#xff1a; 详细注释&#xff1a; 话不多说&#xff0c;上ACcode&#xff01;: #include<bits/stdc.h> using namespace std; #define int lo…...

Http协议和RestTemplate协议有什么区别?

目录 一、功能不同 二、技术不同 三、使用场景不同 四、总结 RestTemplate 是一个 Spring 框架提供的用于发送 HTTP请求的客户端工具&#xff0c;它封装了 Java 原生的 HTTP 客户端库&#xff0c;并提供了一组简洁易用的 API 来发送 HTTP 请求和处理响应。而 HTTP&#xff…...

基于SpringBoot+微信小程序的医院预约叫号小程序

✌全网粉丝20W,csdn特邀作者、博客专家、CSDN新星计划导师、java领域优质创作者,博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取项目下载方式&#x1f345; 一、项目背景介绍&#xff1a; 该项目是基于uniappWe…...

springboot整合RabbitMQ 消费端处理数据

pom 依赖 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>写一个rabbitmq配置文件 import org.springframework.amqp.core.Binding; import org.springframewo…...

计算机中CPU、内存、缓存的关系

CPU&#xff08;Central Processing Unit&#xff0c;中央处理器&#xff09; 内存&#xff08;Random Access Memory&#xff0c;随机存取存储器&#xff09; 缓存&#xff08;Cache&#xff09; CPU、内存和缓存之间有着密切的关系&#xff0c;它们共同构成了计算机系统的核…...

【Linux实验】构造一个简单的 shell

一、实验目的 l 用 C/C++构造一个简单的 shell; l 理解 shell 程序的功能; l 学会 shell 的使用;...

调用支付宝接口响应40004 SYSTEM_ERROR问题排查

在对接支付宝API的时候&#xff0c;遇到了一些问题&#xff0c;记录一下排查过程。 Body:{"datadigital_fincloud_generalsaas_face_certify_initialize_response":{"msg":"Business Failed","code":"40004","sub_msg…...

黑马Mybatis

Mybatis 表现层&#xff1a;页面展示 业务层&#xff1a;逻辑处理 持久层&#xff1a;持久数据化保存 在这里插入图片描述 Mybatis快速入门 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/6501c2109c4442118ceb6014725e48e4.png //logback.xml <?xml ver…...

为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?

在建筑行业&#xff0c;项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升&#xff0c;传统的管理模式已经难以满足现代工程的需求。过去&#xff0c;许多企业依赖手工记录、口头沟通和分散的信息管理&#xff0c;导致效率低下、成本失控、风险频发。例如&#…...

质量体系的重要

质量体系是为确保产品、服务或过程质量满足规定要求&#xff0c;由相互关联的要素构成的有机整体。其核心内容可归纳为以下五个方面&#xff1a; &#x1f3db;️ 一、组织架构与职责 质量体系明确组织内各部门、岗位的职责与权限&#xff0c;形成层级清晰的管理网络&#xf…...

从零实现STL哈希容器:unordered_map/unordered_set封装详解

本篇文章是对C学习的STL哈希容器自主实现部分的学习分享 希望也能为你带来些帮助~ 那咱们废话不多说&#xff0c;直接开始吧&#xff01; 一、源码结构分析 1. SGISTL30实现剖析 // hash_set核心结构 template <class Value, class HashFcn, ...> class hash_set {ty…...

聊一聊接口测试的意义有哪些?

目录 一、隔离性 & 早期测试 二、保障系统集成质量 三、验证业务逻辑的核心层 四、提升测试效率与覆盖度 五、系统稳定性的守护者 六、驱动团队协作与契约管理 七、性能与扩展性的前置评估 八、持续交付的核心支撑 接口测试的意义可以从四个维度展开&#xff0c;首…...

Typeerror: cannot read properties of undefined (reading ‘XXX‘)

最近需要在离线机器上运行软件&#xff0c;所以得把软件用docker打包起来&#xff0c;大部分功能都没问题&#xff0c;出了一个奇怪的事情。同样的代码&#xff0c;在本机上用vscode可以运行起来&#xff0c;但是打包之后在docker里出现了问题。使用的是dialog组件&#xff0c;…...

如何在网页里填写 PDF 表格?

有时候&#xff0c;你可能希望用户能在你的网站上填写 PDF 表单。然而&#xff0c;这件事并不简单&#xff0c;因为 PDF 并不是一种原生的网页格式。虽然浏览器可以显示 PDF 文件&#xff0c;但原生并不支持编辑或填写它们。更糟的是&#xff0c;如果你想收集表单数据&#xff…...

Reasoning over Uncertain Text by Generative Large Language Models

https://ojs.aaai.org/index.php/AAAI/article/view/34674/36829https://ojs.aaai.org/index.php/AAAI/article/view/34674/36829 1. 概述 文本中的不确定性在许多语境中传达,从日常对话到特定领域的文档(例如医学文档)(Heritage 2013;Landmark、Gulbrandsen 和 Svenevei…...

招商蛇口 | 执笔CID,启幕低密生活新境

作为中国城市生长的力量&#xff0c;招商蛇口以“美好生活承载者”为使命&#xff0c;深耕全球111座城市&#xff0c;以央企担当匠造时代理想人居。从深圳湾的开拓基因到西安高新CID的战略落子&#xff0c;招商蛇口始终与城市发展同频共振&#xff0c;以建筑诠释对土地与生活的…...