flink1.12 kafka升级记录

作者：机器笔尖舞者 | 2024-01-29 18:11:55

踩

flink1.12 kafka升级记录

在1.12之前的做法

在1.12之前 flinksql是不支持kafka 多个topic的设置, 所以之前使用的时候简单对官方kafka连接器中kafkaDynamicSource进行封装,让其支持多个topic的设置

对应文章是1.11.0 flinksql自定义kafka源(支持多个topic)

现在1.12版本中, kafka做了一些改动,如下

1.12多topic的支持(源码调试查看)

比如KafkaOptions中TOPIC的类型是List<String>

分隔符是什么呢?因为tablesql中的配置是字符串类型,所以支持多topic肯定是会有一定的规定.

查看KafkaDynamicTableFactory的createDynamicTableSource中创建kafka源,构造方法中需要传入topic

进入 KafkaOptions.getSourceTopics(tableOptions)

进入 getOptional(TOPIC)

调试可以发现进入 ConfigurationUtils.convertToList(v, clazz),再进入就能看到他是用分号切分的

1.12 多topic的支持(官网查看)

kafka官网配置

跟我们查看源码时对应上了

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/blog/article/detail/44115

推荐阅读

article
SpringBoot——集成Kafka详解_springboot kafka
文章浏览阅读1w次，点赞32次，收藏100次。我们知道，kafka中每个topic被划分为多个分区，那么生产者将消息发送到topic时，具体追加到哪个分区呢？这就是所谓的分区策略，Kafka为我们提供了默认的分区策略，同时它也支持自定义分区... [详细]
赞
踩
article
Kafka（一）使用Docker Compose安装单机Kafka以及Kafka UI
这里我们的配置是选择的是KRaft，因为Kafka官方已经计划在Kafak中移除Zookeeper。对于UI配置项没什么特别要说的，这里只是提一下，注意这里的docker-compose.yml中environment的写法，和上面的Kaf... [详细]
赞
踩
article
kafka 分布式的情况下，如何保证消息的顺序消费?_kafka顺序消费
kafka分布式的情况下，如何保证消息的顺序消费?_kafka顺序消费kafka顺序消费目录一、什么是分布式二、kafka介绍三、消息的顺序消费四、如何保证消息的顺序消费一、什么是分布式分布式是指将计算任务分散到多个计算节点上进... [详细]
赞
踩
article
阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来
在讲Kafka日志源码之前，我们要先对Kafka日志有一个大体的认识这也是阅读源码的关键，一步一步来前面我们聊到了Kafka的生产端的整体架构可以看到，我们每一个Topic都可以分为多个Partition，而每一个Partition对应着一... [详细]
赞
踩
article
Flink系列之：Upsert Kafka SQL 连接器
更准确地说，数据记录中的value被解释为同一key的最后一个value的UPDATE，如果有这个key（如果不存在相应的key，则该更新被视为INSERT）。总之，这段代码的作用是通过Kafka连接器创建两个表，并将"pageviews"... [详细]
赞
踩
article
Flink Kafka[输入/输出] Connector_flink kafka数据流和控制流配置关联赋值在输出
本章重点介绍生产环境中最常用到的。使用Flink的同学，一定会很熟悉kafka，它是一个分布式的、分区的、多副本的、支持高吞吐的、发布订阅消息系统。生产环境环境中也经常会跟kafka进行一些数据的交换，比如利用读取数据，然后进行一系列的处理... [详细]
赞
踩
article
Flink（十一）Flink连接Kafka输出到HDFS_flink kafka hdfs
一、配置项目的依赖其中flink-connector-filesystem_2.11是将Hadoop作为Flink的BucketingSink接入，hadoop-hdfs、hadoop-common、hadoop-client解决Jar包依... [详细]
赞
踩
article
Flink Kafka source神操作之Flink Kafka connector
引言Flink提供了专门的Kafka连接器，向Kafkatopic中读取或者写入数据。FlinkKafkaConsumer集成了Flink的Checkpoint机制，可提供exactly-once的处理语义。为此，Flink并不完全依赖于跟... [详细]
赞
踩
article
【Flink实战系列】Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/consumer/ConsumerRecord；)Ljava/_serialize(ljava/lang/object;ljava/lang/long;)lorg/
本文主要是分享了一个jar包冲突异常，以及怎样一步步排查问题到最后的解决问题，其实这两个异常在平时是非常常见的，给大家提供一个排查的思路，如果遇到类似的问题，知道怎么去解决。httpshttpshttps。......_serialize(... [详细]
赞
踩
article
Flink Kafka Connector介绍和使用（DataStream and Table）_flink keyby kafka分区 connectedstream
前言Flink提供了一个ApacheKafka连接器，我们可以很方便的实现从Kafka主题读取数据和向其写入数据。Flink附带了提供了多个Kafka连接器：universal通用版本，0.10，0.11官方文档解释说universal(通... [详细]
赞
踩
article
flink kafka connector_flink1.17安装kafka需要什么版本
1.kafkaconnector版本选取Flink有多个Kafkaconnector：universal，0.10和0.11。Flink1.7开始就有这个universal的Kafkaconnector通用版本，跟Kafkaclient端的... [详细]
赞
踩
article
Flink流处理操作、并对Kafka和mysql进行数据流处理_flink kafka mysql
Flink流处理1.输入数据集DataSource1.基于集合的source2.基于文件的source3.基于网络套接字的source4.自定义source5.使用Kafka作为数据源6.使用MySQL作为数据源2.DataStream的T... [详细]
赞
踩
article
(Ljava/lang/Object；Ljava/lang/Long；)Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/producer
问题原因是kafka版本过低再项目中导入kafka-client包。_lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/producer/producerreclorg/apac... [详细]
赞
踩
article
Flink Kafka connector详解
在Flink中提供了特殊的Connectors从kafka中读写数据，它基于KafkaConsumerAPI以及ProducerAPI封装了自己的一套API来连接kafka，即可读取kafka中的数据流，也可以对接数据流到kafka中，基于... [详细]
赞
踩
article
python flink kafka_Flink-Kafka-Connector Flink结合Kafka实战
戳更多文章：简介Flink-kafka-connector用来做什么？Kafka中的partition机制和Flink的并行度机制结合，实现数据恢复Kafka可以作为Flink的source和sink任务失败，通过设置kafka的offse... [详细]
赞
踩
article
Flink Kafka Connector 实现机制简析
说完了preCommit和commit，在两次Checkpoint发生的间隔当中，会持续的执行invoke方法将数据的写到Kafka，直到snapshotState方法被调用，旧的数据被进行预提交，同时生成新的事务，数据继续写入，直到所有任... [详细]
赞
踩
article
flink 启动job时提示Lorg/apache/flink/metrics/groups/OperatorMetricGroup_flink kafka getmetricgroup()lorg/apache/flink/metr
INITIALIZINGtoFAILEDwithfailurecause:java.lang.NoSuchMethodError:org.apache.flink.api.common.functions.RuntimeContext.ge... [详细]
赞
踩
article
Flink Kafka Connector源码核心_kafkadynamictablesource
入口依然是通过SPI机制找到kafka的factory（KafkaDynamicTableFactory），Flink中大量使用了SPI机制，有时间再整理一篇SPI在Flink中的应用。话不多说，进入正题。Source通过createDyn... [详细]
赞
踩
article
使Flink SQL Kafka Source支持独立设置并行度
前言社区在Flink1.12版本通过FLIP-146提出了增强FlinkSQLDynamicTableSource/Sink接口的动议，其中的一个主要工作就是让它们支持独立设置并行度。很多Sink都已经可以配置sink.parallelis... [详细]
赞
踩
article
一文搞懂 FlinkSQL 的 KafkaSource_flink sql source kafka
背景前面我们了解了写给大忙人看的Flink消费Kafka，今天我们一起来看一下FlinkSQLKafka是如何与FlinkStreamingKafka结合起来的正文创建kafkasourceCREATETABLEorders(statusi... [详细]
赞
踩

flink1.12 kafka升级记录

在1.12之前的做法

1.12多topic的支持(源码调试查看)

1.12 多topic的支持(官网查看)

SpringBoot——集成Kafka详解_springboot kafka

Kafka（一）使用Docker Compose安装单机Kafka以及Kafka UI

kafka 分布式的情况下，如何保证消息的顺序消费?_kafka顺序消费

阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来

Flink系列之：Upsert Kafka SQL 连接器

Flink Kafka[输入/输出] Connector_flink kafka数据流和控制流配置关联赋值在输出

Flink（十一）Flink连接Kafka输出到HDFS_flink kafka hdfs

Flink Kafka source神操作之Flink Kafka connector

【Flink实战系列】Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/consumer/ConsumerRecord；)Ljava/_serialize(ljava/lang/object;ljava/lang/long;)lorg/

Flink Kafka Connector介绍和使用（DataStream and Table）_flink keyby kafka分区 connectedstream

flink kafka connector_flink1.17安装kafka需要什么版本

Flink流处理操作、并对Kafka和mysql进行数据流处理_flink kafka mysql

(Ljava/lang/Object；Ljava/lang/Long；)Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/producer

Flink Kafka connector详解

python flink kafka_Flink-Kafka-Connector Flink结合Kafka实战

Flink Kafka Connector 实现机制简析

flink 启动job时提示Lorg/apache/flink/metrics/groups/OperatorMetricGroup_flink kafka getmetricgroup()lorg/apache/flink/metr

Flink Kafka Connector源码核心_kafkadynamictablesource

使Flink SQL Kafka Source支持独立设置并行度

一文搞懂 FlinkSQL 的 KafkaSource_flink sql source kafka