当前位置: 首页 > 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 的使用;...

Cursor实现用excel数据填充word模版的方法

cursor主页&#xff1a;https://www.cursor.com/ 任务目标&#xff1a;把excel格式的数据里的单元格&#xff0c;按照某一个固定模版填充到word中 文章目录 注意事项逐步生成程序1. 确定格式2. 调试程序 注意事项 直接给一个excel文件和最终呈现的word文件的示例&#xff0c;…...

Go 语言接口详解

Go 语言接口详解 核心概念 接口定义 在 Go 语言中&#xff0c;接口是一种抽象类型&#xff0c;它定义了一组方法的集合&#xff1a; // 定义接口 type Shape interface {Area() float64Perimeter() float64 } 接口实现 Go 接口的实现是隐式的&#xff1a; // 矩形结构体…...

企业如何增强终端安全?

在数字化转型加速的今天&#xff0c;企业的业务运行越来越依赖于终端设备。从员工的笔记本电脑、智能手机&#xff0c;到工厂里的物联网设备、智能传感器&#xff0c;这些终端构成了企业与外部世界连接的 “神经末梢”。然而&#xff0c;随着远程办公的常态化和设备接入的爆炸式…...

云原生安全实战:API网关Kong的鉴权与限流详解

&#x1f525;「炎码工坊」技术弹药已装填&#xff01; 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、基础概念 1. API网关&#xff08;API Gateway&#xff09; API网关是微服务架构中的核心组件&#xff0c;负责统一管理所有API的流量入口。它像一座…...

华为OD机考-机房布局

import java.util.*;public class DemoTest5 {public static void main(String[] args) {Scanner in new Scanner(System.in);// 注意 hasNext 和 hasNextLine 的区别while (in.hasNextLine()) { // 注意 while 处理多个 caseSystem.out.println(solve(in.nextLine()));}}priv…...

6.9-QT模拟计算器

源码: 头文件: widget.h #ifndef WIDGET_H #define WIDGET_H#include <QWidget> #include <QMouseEvent>QT_BEGIN_NAMESPACE namespace Ui { class Widget; } QT_END_NAMESPACEclass Widget : public QWidget {Q_OBJECTpublic:Widget(QWidget *parent nullptr);…...

C++课设:实现本地留言板系统(支持留言、搜索、标签、加密等)

名人说&#xff1a;路漫漫其修远兮&#xff0c;吾将上下而求索。—— 屈原《离骚》 创作者&#xff1a;Code_流苏(CSDN)&#xff08;一个喜欢古诗词和编程的Coder&#x1f60a;&#xff09; 专栏介绍&#xff1a;《编程项目实战》 目录 一、项目功能概览与亮点分析1. 核心功能…...

Three.js进阶之粒子系统(一)

一些特定模糊现象&#xff0c;经常使用粒子系统模拟&#xff0c;如火焰、爆炸等。Three.js提供了多种粒子系统&#xff0c;下面介绍粒子系统 一、Sprite粒子系统 使用场景&#xff1a;下雨、下雪、烟花 ce使用代码&#xff1a; var materialnew THRESS.SpriteMaterial();//…...

Jmeter(四) - 如何在jmeter中创建网络测试计划

1.简介 如何创建基本的 测试计划来测试网站。您将创建五个用户&#xff0c;这些用户将请求发送到JMeter网站上的两个页面。另外&#xff0c;您将告诉用户两次运行测试。 因此&#xff0c;请求总数为&#xff08;5个用户&#xff09;x&#xff08;2个请求&#xff09;x&#xff…...

Go 语言中switch case条件分支语句

1. 基本语法 package main import "fmt" func main() {var extname ".css"switch extname {case ".html":fmt.Println("text/html")case ".css":fmt.Println("text/css") // text/csscase ".js":fmt.…...