Flink 1.20 Kafka Connector：新旧 API 深度解析与迁移指南

news2025/7/14 0:29:07

Flink Kafka Connector 新旧 API 深度解析与迁移指南

一、Flink Kafka Connector 演进背景

Apache Flink 作为实时计算领域的标杆框架，其 Kafka 连接器的迭代始终围绕性能优化、语义增强和API 统一展开。Flink 1.20 版本将彻底弃用基于 FlinkKafkaConsumer/FlinkKafkaProducer 的旧 API（标记为 @Deprecated），全面转向基于新 Source/Sink API（FLIP-27/FLIP-143）的 KafkaSource/KafkaSink。这一变革不仅带来了架构上的革新，更通过流批统一、精确一次语义和动态分区管理等特性，显著提升了用户体验。

二、新旧 API 核心差异对比

Apache Flink 在 1.13+ 版本逐步引入了全新的 Source/Sink API（也称为 FLIP-27 架构），取代了旧的 SourceFunction/SinkFunction 架构。这一变化旨在解决旧架构在扩展性、批流统一、状态管理等方面的局限性。

1. 新架构实现

这里我们从新 Source 举例，来了解新架构及实现原理：

开始了解 - 新 Source 原理 ：

Flink 原先数据源一直使用的是 SourceFunction。实现它的 run 方法，使用 SourceContextcollect 数据或者发送 watermark 就实现了一个数据源。但是它有如下问题（来源于FLIP-27: Refactor Source Interface - Apache Flink - Apache Software Foundation翻译）：

  • 同一类型数据源的批和流模式需要两套不同实现。

  • “work发现”（分片、分区等）和实际 “读取” 数据的逻辑混杂在 SourceFunction 接口和 DataStream API 中，导致实现非常复杂，如 Kafka 和 Kinesis 源等。

  • 分区/分片/拆分在接口中不是明确的。这使得以与 source 无关的方式实现某些功能变得困难，例如 event time 对齐、每个分区水印、动态拆分分配、工作窃取。例如，Kafka 和 Kinesis consumer 支持每个分区的 watermark，但从 Flink 1.8.1 开始，只有 Kinesis 消费者支持 event time 对齐（选择性地从拆分中读取以确保我们在事件时间上均匀推进）。

  • Checkpoint 锁由 source function “拥有”。实现必须确保进行元素发送和 state 更新时加锁。 Flink 无法优化它处理该锁的方式。
    锁不是公平锁。在锁竞争下，一些线程可能无法获得锁（checkpoint线程）。这也妨碍使用 actor/mailbox 无锁线程模型。

  • 没有通用的构建块，这意味着每个源都自己实现了一个复杂的线程模型。这使得实施和测试新 source 变得困难，并增加了对现有 source 的开发贡献的标准。

为了解决这些问题，Flink 引入了新的 Source 架构。

一个数据 source 包括三个核心组件：分片（Splits）、分片枚举器（SplitEnumerator） 以及 源阅读器（SourceReader）。

  • 分片（Split） 是对一部分 source 数据的包装，如一个文件或者日志分区。分片是 source 进行任务分配和数据并行读取的基本粒度。

  • 源阅读器（SourceReader） 会请求分片并进行处理，例如读取分片所表示的文件或日志分区。SourceReader 在 TaskManagers 上的 SourceOperators 并行运行，并产生并行的事件流/记录流。

  • 分片枚举器（SplitEnumerator） 会生成分片并将它们分配给 SourceReader。该组件在 JobManager 上以单并行度运行，负责对未分配的分片进行维护，并以均衡的方式将其分配给 reader。

Source 类作为API入口，将上述三个组件结合在了一起。

参考原文内容：https://blog.csdn.net/bigdatakenan/article/details/141064429

总的来说，Flink 新 Source/Sink 架构的本质是通过组件解耦和动态分片机制，以实现更加灵活、精细化的资源管理。

2. 核心类与依赖

特性旧 API（1.12 之前）新 API（1.13+）
Source 实现类FlinkKafkaConsumerKafkaSource
Sink 实现类FlinkKafkaProducerKafkaSink
依赖坐标 示例flink-connector-kafka_2.11:1.12flink-connector-kafka:1.20
批流统一需要不同 API同一 API 支持流批
资源效率静态并行度动态分片分配
并发与锁管理全局锁导致高竞争分片级并发

3. 关键功能介绍

（1）动态分区发现

旧 API 通过 flink.partition-discovery.interval-millis 配置分区发现间隔。

新 API 通过 partition.discovery.interval.ms 配置分区发现间隔。

// 设置动态分区发现，间隔为 10 秒
// 旧 API
Properties props = new Properties();
...
props.setProperty("flink.partition-discovery.interval-millis", "10000"); 
FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
    "topic-name",
    new SimpleStringSchema(),
    props
);

// 新 API 
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(Config.KAFKA_SERVER)
    .setValueOnlyDeserializer(new SimpleStringSchema())
    ...
    .setProperty("partition.discovery.interval.ms", "10000") 
    .build();
（2）起始偏移量控制

旧 API 通过 auto.offset.reset 参数 或 flinkKafkaConsumer.setStartFromEarliest()的方法配置 offset 。

新 API 提供更细粒度的 offset 控制：

// 旧 API
Properties props = new Properties();
props.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
    "topic-name",
    new SimpleStringSchema(),
    props
);
// 或
flinkKafkaConsumer.setStartFromEarliest();

// 新 API 
// 从最早可用偏移量（earliest offset）开始消费，忽略消费者组已提交的偏移量。
KafkaSource.builder()
    .setStartingOffsets(OffsetsInitializer.earliest())
    .build();

// 优先使用消费者组已提交的偏移量（若存在），如果无提交的偏移量（如首次启动消费者组），则回退到 EARLIEST 偏移量
KafkaSource.builder()
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .build();

// 从大于或等于指定时间戳（Unix 毫秒时间戳）的 Kafka 消息开始消费
KafkaSource.builder()
    .setStartingOffsets(OffsetsInitializer.timestamp(1657256257000L))
    .build();
（3）事务性写入

相较于旧 API ，新 API 配置事务方式更加简洁和易读。

Properties props = new Properties();
...
props.setProperty("enable.idempotence", true); 
props.setProperty("transaction.timeout.ms", "900000"); 

// 旧 API Exactly-Once Sink
FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer<>(
        "topic",
        new SimpleStringSchema(),
        props,
        FlinkKafkaProducer.Semantic.EXACTLY_ONCE // 旧 API 语义配置
    )
flinkKafkaProducer.setTransactionalIdPrefix("flink-transactional-id-");

// 新 API Exactly-Once Sink
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers(Config.KAFKA_SERVER)
    .setRecordSerializer(...)
    .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 新 API 语义配置
    .setTransactionalIdPrefix("flink-transactional-id-")
    .setKafkaProducerConfig(props)
    .build();

三、迁移实战指南

1. 依赖升级

<!-- 旧 API 依赖 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.12.x</version>
</dependency>

<!-- 新 API 依赖 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.20.x</version>
</dependency>

2. Source 迁移示例（ConsumerRecord 版）

（1）旧 API 实现
Properties properties = new Properties();
properties.setProperty("consumer.topic", "test_123");
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "group_id_001");
properties.setProperty("auto.offset.reset", "earliest");

FlinkKafkaConsumer<ConsumerRecord<byte[], byte[]>> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
    properties.getProperty("consumer.topic"),
    new KafkaDeserializationSchema<ConsumerRecord<byte[], byte[]>>(){
        @Override
        public TypeInformation getProducedType() {
            return TypeInformation.of(new TypeHint<ConsumerRecord<byte[], byte[]>>() {});
        }
        @Override
        public ConsumerRecord<byte[], byte[]> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
            return new ConsumerRecord<>(
                record.topic(),
                record.partition(),
                record.offset(),
                record.timestamp(),
                record.timestampType(),
                record.checksum(),
                record.serializedKeySize(),
                record.serializedValueSize(),
                record.key(),
                record.value()
            );
        }
        @Override
        public boolean isEndOfStream(ConsumerRecord<byte[], byte[]> nextElement) {
            return false;
        }
    },
    properties
);

env.addSource(flinkKafkaConsumer);
（2）新 API 实现
KafkaSource<ConsumerRecord<String, String>> source = KafkaSource.<ConsumerRecord<String, String>>builder()
    .setBootstrapServers(Config.KAFKA_SERVER)
    .setTopics(Config.KAFKA_TOPIC)
    .setGroupId(Config.KAFKA_GROUP_ID)
    .setStartingOffsets(OffsetsInitializer.earliest()) 
    .setDeserializer(KafkaRecordDeserializationSchema.of(new KafkaDeserializationSchema<ConsumerRecord<String, String>>() {
        @Override
        public TypeInformation<ConsumerRecord<String, String>> getProducedType() {
            return TypeInformation.of(new TypeHint<ConsumerRecord<String, String>>() {
            });
        }

        @Override
        public boolean isEndOfStream(ConsumerRecord<String, String> nextElement) {
            return false;
        }

        @Override
        public ConsumerRecord<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
            return new ConsumerRecord<String, String>(
                record.topic(),
                record.partition(),
                record.offset(),
                record.timestamp(),
                record.timestampType(),
                record.checksum(),
                record.serializedKeySize(),
                record.serializedValueSize(),
                record.key() == null ? "" : new String(record.key(), StandardCharsets.UTF_8),
                record.value() == null ? "" : new String(record.value(), StandardCharsets.UTF_8)
            );
        }
    }))
    .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "New Flink Kafka Source");

3. Sink 迁移示例（ProducerRecord 版）

（1）旧 API 实现
Properties props = new Properties();
properties.setProperty("producer.topic", "test_123");
props.setProperty("bootstrap.servers", "localhost:9092");

FlinkKafkaProducer<byte[]> flinkKafkaProducer = new FlinkKafkaProducer<>(
    properties.getProperty("producer.topic"),
    new KafkaSerializationSchema<byte[]>() {
        @Override
        public void open(SerializationSchema.InitializationContext context) throws Exception {
            KafkaSerializationSchema.super.open(context);
        }

        @Override
        public ProducerRecord<byte[], byte[]> serialize(byte[] element, @Nullable Long timestamp) {
            return new ProducerRecord<>(
                properties.getProperty("producer.topic"),
                "my_key_id".getBytes(StandardCharsets.UTF_8),
                element);
        }
    },
    properties,
    FlinkKafkaProducer.Semantic.AT_LEAST_ONCE
);

env.fromElements("apple", "banana", "orange")
    .map(i -> i.getBytes(StandardCharsets.UTF_8))
    .addSink(flinkKafkaProducer);
（2）新 API 实现
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers(Config.KAFKA_SERVER)
    .setRecordSerializer((KafkaRecordSerializationSchema<String>) (element, context, timestamp) -> {
        String keyId = "my_key_id";
        byte[] key = keyId.getBytes(StandardCharsets.UTF_8); // 指定 key
        byte[] value = element.getBytes(StandardCharsets.UTF_8); // 指定 value
        return new ProducerRecord<>(Config.KAFKA_TOPIC, key, value);
    })
    .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .build();

env.fromElements("apple", "banana", "orange")
    .sinkTo(sink);

四、总结与展望

通过Flink KafkaSource/KafkaSink 的连接器 API，用户在使用时不仅获得了更简洁的编程模型，还享受到更便捷的动态分区管理、精确一次语义和性能优化等高级特性。建议用户尽早迁移至新 API，以充分利用 Flink 1.20 的增强功能。

Flink 1.20 版本的发布标志着 Flink 在流批一体和云原生架构方面迈出了重要的一步，进一步巩固了其作为实时流数据处理平台的地位。这一版本在功能上进行了诸多增强，还在性能和易用性方面做出了显著改进，为用户提供了更强大的工具来处理复杂的流批任务。

Flink 1.20 核心改进总结【官宣｜Apache Flink 1.20 发布公告】：

  1. Flink SQL 物化表（Materialized Table）

    • 功能特性
      • 通过声明式 SQL 定义动态表结构与数据新鲜度（如 FRESHNESS = INTERVAL '3' MINUTE），引擎自动构建流批统一的数据加工链路。
      • 支持流式持续刷新、批式全量刷新、增量刷新三种模式，用户可根据成本灵活切换（如大促时秒级实时，日常天级批处理）。
      • 简化运维操作：暂停/恢复数据刷新（SUSPEND/RESUME）、手动回刷历史分区（REFRESH PARTITION）。
    • 价值
      • 降低 ETL 开发复杂度，无需分别维护流/批作业，提升实时数仓构建效率。

  2. 状态与检查点优化

    • 统一检查点文件合并机制
      • 将零散小文件合并为大文件，减少元数据压力，需配置 execution.checkpointing.file-merging.enabled=true。
      • 支持跨检查点合并（execution.checkpointing.file-merging.across-checkpoint-boundary=true）及文件池模式选择（阻塞/非阻塞）。
    • RocksDB 优化
      • 后台自动合并小 SST 文件，避免因文件数量膨胀导致检查点失败。

  3. 批处理容错能力增强

    • JobMaster 故障恢复机制
      • 通过 JobEventStore 持久化执行状态（如任务进度、算子协调器状态），故障后从外部存储恢复进度，避免重跑已完成任务。
      • 需启用集群高可用（HA）并配置 execution.batch.job-recovery.enabled=true。

  4. API 演进

    • DataSet API 弃用
      • 推荐迁移至 DataStream API 或 Table API/SQL，实现流批统一编程模型。
    • DataStream API 增强
      • 支持全量分区数据处理（fullWindowPartition），补齐批处理能力。

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

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

相关文章

2025年渗透测试面试题总结- 某四字大厂面试复盘扩展 一面（题目+回答）

2025年渗透测试面试题总结- 某四字大厂面试复盘扩展 一面（题目+回答）

网络安全领域各种资源&#xff0c;学习文档&#xff0c;以及工具分享、前沿信息分享、POC、EXP分享。不定期分享各种好玩的项目及好用的工具&#xff0c;欢迎关注。 目录 某四字大厂面试复盘扩展 一面 一、Java内存马原理与查杀 二、冰蝎与哥斯拉原理对比&#xff08;技术演…
阅读更多...
批量压缩 jpg/png 等格式照片|批量调整图片的宽高尺寸

批量压缩 jpg/png 等格式照片|批量调整图片的宽高尺寸

图片格式种类非常的多&#xff0c;并且不同的图片由于像素、尺寸不一样&#xff0c;可能占用的空间也会不一样。文件太大会占用较多的磁盘空间&#xff0c;传输及上传系统都非常不方便&#xff0c;可能会收到限制&#xff0c;因此我们经常会碰到需要对图片进行压缩的需求。如何…
阅读更多...
【动手学深度学习】卷积神经网络（CNN）入门

【动手学深度学习】卷积神经网络（CNN）入门

【动手学深度学习】卷积神经网络&#xff08;CNN&#xff09;入门 1&#xff0c;卷积神经网络简介2&#xff0c;卷积层2.1&#xff0c;互相关运算原理2.2&#xff0c;互相关运算实现2.3&#xff0c;实现卷积层 3&#xff0c;卷积层的简单应用&#xff1a;边缘检测3.1&#xff0…
阅读更多...
在huggingface上制作小demo

在huggingface上制作小demo

在huggingface上制作小demo 今天好兄弟让我帮他搞一个模型&#xff0c;他有小样本的化学数据&#xff0c;想让我根据这些数据训练一个小模型&#xff0c;他想用这个模型预测一些值 最终我简单训练了一个小模型&#xff0c;起初想把这个模型和GUI界面打包成exe发给他&#xff0…
阅读更多...
51.评论日记

51.评论日记

千万不能再挖了&#xff0c;否则整个华夏文明将被改写。_哔哩哔哩_bilibili 2025年4月7日22:13:42
阅读更多...
SpringCloud第二篇：注册中心Eureka

SpringCloud第二篇：注册中心Eureka

注册中心的意义 注册中心 管理各种服务功能包括服务的注册、发现、熔断、负载、降级等&#xff0c;比如dubbo admin后台的各种功能。 有了注册中心&#xff0c;调用关系的变化&#xff0c;画几个简图来看一下。(了解源码可求求: 1791743380) 服务A调用服务B 有了注册中心之后&a…
阅读更多...
ES 参数调优

ES 参数调优

1、refresh_interval 控制索引刷新的时间间隔。增大这个值可以减少I/O操作&#xff0c;从而提升写入性能&#xff0c;但会延迟新文档的可见性 查看 GET /content_erp_nlp_help_202503191453/_settings?include_defaultstrue 动态修改&#xff1a;refresh_interval 是一个动态…
阅读更多...
用claude3.7，不到1天写了一个工具小程序（11个工具6个游戏）

用claude3.7，不到1天写了一个工具小程序（11个工具6个游戏）

一、功能概览和本文核心 本次开发&#xff0c;不是1天干撸&#xff0c;而是在下班后或早起搞的&#xff0c;总体加和计算了一下&#xff0c;大概1天的时间&#xff08;12个小时&#xff09;&#xff0c;平常下班都是9点的衰仔&#xff0c;好在还有双休&#xff0c;谢天谢地。 …
阅读更多...
【GeoDa使用】空间自相关分析操作

【GeoDa使用】空间自相关分析操作

使用 GeoDa 软件进行空间自相关分析 双击打开 GeoDa 软件 选择 .shp 文件 导入文件 空间权重矩阵&#xff08;*.gal / *.gwt&#xff09;是进行任何空间分析的前提 构建空间权重矩阵 空间权重矩阵&#xff08;Spatial Weights Matrix&#xff09; 是一个用来描述空间对象之间…
阅读更多...
C++基于rapidjson的Json与结构体互相转换

C++基于rapidjson的Json与结构体互相转换

简介 使用rapidjson库进行封装&#xff0c;实现了使用C对结构体数据和json字符串进行互相转换的功能。最短只需要使用两行代码即可无痛完成结构体数据转换为Json字符串。 支持std::string、数组、POD数据&#xff08;int,float,double等&#xff09;、std::vector、嵌套结构体…
阅读更多...
OpenStack Yoga版安装笔记（十七）安全组笔记

OpenStack Yoga版安装笔记（十七）安全组笔记

一、安全组与iptables的关系 OpenStack的安全组&#xff08;Security Group&#xff09;默认是通过Linux的iptables实现的。以下是其主要实现原理和机制&#xff1a; 安全组与iptables的关系 OpenStack的安全组规则通过iptables的规则链实现。每条安全组规则会被转换为相应的i…
阅读更多...
通义万相2.1 图生视频：为AI绘梦插上翅膀，开启ALGC算力领域新纪元

通义万相2.1 图生视频：为AI绘梦插上翅膀，开启ALGC算力领域新纪元

通义万相2.1图生视频大模型 通义万相2.1图生视频技术架构万相2.1的功能特点性能优势与其他工具的集成方案 蓝耘平台部署万相2.1核心目标典型应用场景未来发展方向 通义万相2.1ALGC实战应用操作说明功能测试 为什么选择蓝耘智算蓝耘智算平台的优势如何通过API调用万相2.1 写在最…
阅读更多...
52.个人健康管理系统小程序(基于springbootvue)

52.个人健康管理系统小程序(基于springbootvue)

目录 1.系统的受众说明 2.开发环境与技术 2.1 MYSQL数据库 2.2 Java语言 2.3 微信小程序技术 2.4 SpringBoot框架 2.5 B/S架构 2.6 Tomcat 介绍 2.7 HTML简介 2.8 MyEclipse开发工具 3.系统分析 3.1 可行性分析 3.1.1 技术可行性 3.1.2 经济可行性 3.1.3 操作…
阅读更多...
学习比较JVM篇（六）：解读GC日志

学习比较JVM篇（六）：解读GC日志

一、前言 在之前的文章中&#xff0c;我们对JVM的结构、垃圾回收算法、垃圾回收器做了一些列的讲解&#xff0c;同时也使用了JVM自带的命令行工具进行了实际操作。今天我们继续讲解JVM。 我们学习JVM的目的是为了了解JVM&#xff0c;然后优化对应的参数。那么如何了解JVM运行…
阅读更多...
I²S协议概述与信号线说明

I²S协议概述与信号线说明

IIS协议概述 ​ IS&#xff08;Inter-IC Sound&#xff09;协议&#xff0c;又称 IIS&#xff08;Inter-IC Sound&#xff09;&#xff0c;是一种专门用于数字音频数据传输的串行总线标准&#xff0c;由飞利浦&#xff08;Philips&#xff09;公司提出。该协议通常用于微控制器…
阅读更多...
免费Deepseek-v3接口实现Browser-Use Web UI：浏览器自动化本地模拟抓取数据实录

免费Deepseek-v3接口实现Browser-Use Web UI：浏览器自动化本地模拟抓取数据实录

源码 https://github.com/browser-use/web-ui 我们按照官方教程&#xff0c;修订几个环节&#xff0c;更快地部署 步骤 1&#xff1a;克隆存储库 git clone https://github.com/browser-use/web-ui.git cd web-ui Step 2: Set Up Python Environment 第 2 步&#xff1a;设置…
阅读更多...
[蓝桥杯] 求和

[蓝桥杯] 求和

题目链接 P8772 [蓝桥杯 2022 省 A] 求和 - 洛谷 题目理解 这道题就是公式题&#xff0c;我们模拟出公式后&#xff0c;输出最终结果即可。 本题不难&#xff0c;相信很多同学第一次见到这道题都是直接暴力解题。 两个for循环&#xff0c;测试样例&#xff0c;直接拿下。 #in…
阅读更多...
通过Ollama本地部署DeepSeek R1模型（Windows版）

通过Ollama本地部署DeepSeek R1模型（Windows版）

嗨&#xff0c;大家好&#xff0c;我是心海 以下是一份详细的Windows系统下通过Ollama本地部署DeepSeek R1模型的教程&#xff0c;内容简洁易懂&#xff0c;适合新手用户参考 本地部署大模型&#xff0c;就有点像在你自己的电脑或者服务器上&#xff0c;安装并运行这样一个“私…
阅读更多...
【C++】vector的底层封装和实现

【C++】vector的底层封装和实现

目录 目录前言基本框架迭代器容量第一个测试&#xff0c;野指针异常第二轮测试&#xff0c;浅拷贝的问题 元素访问修改操作push_backinsert迭代器失效问题 erase 默认成员函数构造函数双重构造引发调用歧义 拷贝构造赋值重载析构函数 源码end 目录 前言 废话不多说&#xff0…
阅读更多...
Open CASCADE学习|读取点集拟合样条曲线（续）

Open CASCADE学习|读取点集拟合样条曲线（续）

问题 上一篇文章已经实现了样条曲线拟合&#xff0c;但是仍存在问题&#xff0c;Tolerance过大拟合成直线了&#xff0c;Tolerance过大头尾波浪形。 正确改进方案 1️⃣ 核心参数优化 通过调整以下参数控制曲线平滑度&#xff1a; Standard_Integer DegMin 3; // 最低阶…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023