Flink与Pulsar集成实战：架构、连接器与生产实践-育师

1. 为什么把Flink和Pulsar放在一起？——端到端实时链路的最后一环

做实时数据处理的人，这几年应该都有一个明显的感受：消息队列和流计算引擎的关系，已经从"能用就行"变成了"深度绑定"。过去我们习惯Kafka搭配Flink，这套组合非常成熟，社区案例一堆，文档也齐全。但如果真正在线上跑过大规模实时链路，你会发现Kafka+Flink的黄金组合在运维层面有不少让人头疼的地方——分区扩容要手动做数据重分布、Broker挂掉以后副本恢复慢、长时间消息回溯会直接把磁盘IO打满，诸如此类。

我第一次认真考虑Pulsar，是在一个夜间值班的现场。那会儿我们负责的实时推荐系统每天处理上亿条用户行为日志，Kafka集群的主题分区数已经到了几千，业务方提了个新需求：希望数据能回溯到七天前重新灌一遍特征计算。在Kafka上做七天保留其实没问题，但你一旦真的去消费这么久远的数据，磁盘命中率骤降、消费者组重平衡频繁、broker负载飙高，整套链路摇摇欲坠。后来我们调研到Pulsar，发现它的存储层是独立于Broker的，消息写入BookKeeper的Ledger之后长期保留，读历史数据时不会像Kafka那样把所有IO压力砸在单个副本上，这才有了后面的集成方案。

先说结论：Flink与Pulsar集成，本质上解决的是"计算引擎和消息系统如何高效配合，构建真正可回放、可扩展、可运维的实时数据管道"的问题。Pulsar补了Kafka在存储弹性上的短板，Flink补了Pulsar在状态计算上的短板，两者一配合，端到端的实时链路上游可以做海量事件接入，中游可以任意时间回溯重算，下游可以精确一次语义落地到各种存储或触发下游业务。这篇文章我会从架构原理、连接器机制、代码实操、踩坑记录四个维度把它拆开讲清楚，适合已经在用Flink但想换一种更稳的消息底座、或者正在做消息系统选型的实时平台工程师。

2. 从Kafka迁移到Pulsar之前，必须想明白的架构差异

2.1 存储与计算分离：Pulsar真正的底气

很多人以为Pulsar只是"另一个Kafka"，直接拿Kafka的认知套上去，用起来当然别扭。Kafka的Broker既承担网络接入、分区管理，又把数据落在本地磁盘上，集群扩容要靠Rebalance把数据分片搬来搬去。Pulsar则把存储层抽离出来，用Apache BookKeeper作为底层分布式日志存储，Broker变成一个无状态的接入层，只负责处理生产消费请求、执行路由策略，数据落盘是BookKeeper的事。

这个架构差异带来的最直接好处是两个：

消息的写入和读取都在存储层完成，Broker重启或升级时不需要搬迁数据，扩缩容可以在分钟级完成。
数据的存续周期不再受副本迁移成本约束，Pulsar的保留策略能支持任意时长的消息回溯，只要磁盘空间够。

Flink这类流计算引擎，天然依赖消息系统具备两个能力：一是消息能够按序重放，二是消费位点可以持久化。Kafka的offset提交机制和分区架构支持了Flink的checkpoint，而Pulsar的MessageId天然单调递增，并且客户端可以按MessageId精确seek到任意位置，这意味着Flink把检查点数据保存在Pulsar的各个订阅游标上，可以做到非常干净的一致性恢复。

2.2 订阅模型的灵活度，直接影响下游计算拓扑

Pulsar在订阅模型上比Kafka的Consumer Group多了一个维度。Kafka只有一条"分区共享消费"的路子，消费者的数量不能超过分区数，否则多出来的消费者闲着。Pulsar则提供了四种订阅类型：

Exclusive：一个消费者独占整个主题，适合严格有序处理。
Shared：多个消费者以轮询方式共享消息流，适合吞吐优先、允许乱序处理。
Failover：一个主消费者处理流，主消费者故障后从多个备选消费者中挑选一个接管。
Key_Shared：按键路由，相同Key的消息永远发给同一个消费者，兼顾吞吐与局部有序。

这四种订阅落在Flink数据源层面，直接影响Source的并行度设计。Flink的Pulsar Connector默认按分区粒度建立消费者，一个并行子任务处理一个或多个分区的消息，如果上游主题使用了Key_Shared订阅，Flink可以按key的hash路由到特定的并行实例，从而在Source端保住局部有序性。这个特性在"按用户ID聚合特征"这类场景下非常有用，能显著降低下游状态存储的写入频率。

2.3 和Kafka关键能力的对比，别被迁移惯性带偏了

对比维度	Kafka	Pulsar
存储架构	Broker本地磁盘，分区数据与Broker绑定	BookKeeper分布式存储，Broker无状态
扩容方式	需要分区重分配，IO长时间占用	新增Broker即可，数据自动平衡
消息回溯	按offset消费，但历史数据读取性能衰减明显	按MessageId精确seek，存储层与计算隔离
消息保留	基于时间/大小删除，支持compact	同样支持TTL/大小限制，且支持无限流式重放
订阅模型	仅消费组模式	四种订阅模式灵活组合

在选型的时候，我的建议是：如果团队已经有成熟的Kafka运维体系和工具链，业务主要是Kafka架构下的标准流处理，平迁Pulsar的收益并不大。但如果你的场景是"实时数据湖"、长时间回溯分析、跨地域复制、多租户消息服务，Pulsar的分层架构优势会非常明显。Flink只是计算层，底层哪张"桌子"更稳，决定了你能在这张桌子上跑多久。

3. Flink Pulsar Connector的运作机制：Source、位点与Checkpoint

3.1 连接器的三个核心部件

Flink官方在1.15版本之后开始提供Pulsar连接器，之前的版本主要依赖StreamNative维护的connector。我建议直接用flink-connector-pulsar官方连接器，版本跟随Flink主版本，API设计和Kafka Connector风格一致，老Flink用户上手很快。

这个连接器的运作机制，概括起来是三个部分：

SourceReader：Flink的Source端基于Pulsar Consumer API实现。每个Flink子任务持有若干Pulsar分区读取权限，消息进入SourceReader后按Checkpoint机制上报位点。
Pending消息缓存：SourceReader内部维护一个待确认消息队列，消息先写入Flink的状态结构，确认处理完后才向Pulsar Broker发送ack。这一步是精确一次语义的根基。
Cursor维护：Pulsar的订阅游标负责记录每个消费者的消费位置。Flink在checkpoint触发时，把当前正在处理的消息ID固化到状态后端，同时向Pulsar服务端提交游标。恢复时从这个游标继续消费。

3.2 位点管理与Checkpoint的配合方式

Kafka集成中，Flink通过ConsumerSeekToEnd、CommittedOffset这些概念做位点对齐。Pulsar集成稍微不同——Pulsar的MessageId不是简单的数字offset，而是由LedgerId、EntryId、PartitionIndex组成的三元组，天然包含数据在存储层的位置信息。

Flink的Pulsar Connector里，SourceReader会把当前未消费的消息起始MessageId保存为检查点状态的一部分。每次checkpoint完成，连接器就调用Pulsar Consumer的acknowledgeCumulativeAsync方法，把游标推进到已处理的消息位置。这里有个容易忽略的细节：Pulsar的消费确认有两种，acknowledge是逐条确认，acknowledgeCumulative是累计确认到指定消息ID。Flink连接器用的是累计确认，好处是减少确认开销，坏处是实现要严谨——一旦累计确认之后又发生任务重试，这部分消息可能无法重新消费。所以Flink的pending消息机制会保证"先处理完并进入checkpoint，再累计确认"，顺序反了就会丢数据。

3.3 为什么不能把Kafka Connector直接指向Pulsar的Kafka协议卖点

Pulsar有一个KoP（Kafka-on-Pulsar）特性，可以兼容Kafka协议访问。有些同学图省事，直接把Flink的Kafka连接器指向Pulsar的Kafka地址。短期跑通没问题，但我不建议在生产环境这么干。原因很简单：

KoP协议兼容层的功能会滞后于Pulsar原生API，一些高级特性比如Key_Shared订阅、事务消息在KoP上要么不可用要么实现不完整。
Flink Kafka Connector提交的offset在KoP上映射为Pulsar游标，调试时两边看到的数据口径不对齐，排查问题容易踩空。
原生的Pulsar Connector能感知BookKeeper的分层存储优势，比如有界读流和数据无限回溯，KoP很难发挥这种能力。

既然要做集成，就用正规军。原生连接器写起来并不复杂，后面我给你完整代码示例。

4. 实操落地：从环境准备到Flink读写Pulsar的完整代码

4.1 环境准备与依赖引入

我这边使用的版本组合是：Flink 1.17.1、Pulsar 3.2.1、flink-connector-pulsar 1.17.1。需要注意Pulsar的C++客户端和服务端版本兼容性，如果用的是Pulsar 2.x版本，连接器版本也要跟着降级，否则会出现protobuf协议不匹配的奇怪报错。

在Maven项目里引入依赖：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-pulsar</artifactId> <version>1.17.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>1.17.1</version> </dependency> <dependency> <groupId>org.apache.pulsar</groupId> <artifactId>pulsar-client</artifactId> <version>3.2.1</version> </dependency>

注意依赖冲突问题，flink-connector-pulsar内部已经带了pulsar-client，如果你还用别的组件拉入了不同版本的pulsar-client，启动时会报NoSuchMethodError，这时候要检查依赖树，统一版本。

4.2 Pulsar Source接入的完整实现

Flink的Pulsar Source API遵循Flink 1.15之后的新数据源接口，用起来是三步走：创建PulsarClient、构造SourceBuilder、指定位点策略。

import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.connector.pulsar.source.PulsarSource; import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor; import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor; PulsarSource<String> pulsarSource = PulsarSource.builder() .setServiceUrl("pulsar://localhost:6650") .setAdminUrl("http://localhost:8080") .setTopics("persistent://public/default/input-topic") .setDeserializationSchema(new SimpleStringSchema()) .setSubscriptionName("flink-subscription") .setSubscriptionType(SubscriptionType.Shared) .setStartCursor(StartCursor.latest()) .setStopCursor(StopCursor.never()) .build(); DataStream<String> stream = env.fromSource( pulsarSource, WatermarkStrategy.noWatermarks(), "pulsar-source" );

逐行解释几个关键点：

setServiceUrl是客户端连Broker的地址，setAdminUrl是连接器用来读取主题分区元数据的地址，实际消息读写走6650，分区发现和控制面走8080，两边都要配通。
setTopics支持一个或多个完整Topic，也支持正则表达式，例如persistent://public/default/.*-log。注意Pulsar topic名必须带完整persistent://或non-persistent://前缀，缺了前置路径会解析失败。
setStartCursor(StartCursor.latest())表示从最新消息开始消费，如果要做历史回溯，改用StartCursor.earliest()或StartCursor.fromMessageId(MessageId)。
setSubscriptionType(SubscriptionType.Shared)和前面讲的订阅模型对应，Flink Source每个并行子任务会获得一个Pulsar Consumer，订阅名一致才能共享游标。

4.3 Pulsar Sink接入的完整实现

Sink端header是写入Pulsar主题，核心是配置序列化器、路由策略、语义级别。

import org.apache.flink.connector.pulsar.sink.PulsarSink; import org.apache.flink.connector.pulsar.sink.config.RoutingStrategy; import org.apache.flink.connector.pulsar.sink.writer.schema.PulsarSchema; import org.apache.pulsar.client.api.Schema; PulsarSink<String> pulsarSink = PulsarSink.builder() .setServiceUrl("pulsar://localhost:6650") .setAdminUrl("http://localhost:8080") .setTopics("persistent://public/default/output-topic") .setSerializationSchema(PulsarSchema.flinkSchema(new SimpleStringSchema())) .setRoutingStrategy(RoutingStrategy.keyHash()) .setDeliverySemantic(DeliverySemantic.EXACTLY_ONCE) .build(); stream.sinkTo(pulsarSink);

重点讲两个配置：

setDeliverySemantic支持AT_LEAST_ONCE、EXACTLY_ONCE两种模式，默认是EXACTLY_ONCE。如果Pulsar服务端没开启事务（默认事务是开启的，但需要Broker配置transactionCoordinatorEnabled=true），连接器会在运行时抛出PulsarTransactionNotEnabledException，这一点我在踩坑章节会详细说。
setRoutingStrategy控制消息写到哪个分区。RoutingStrategy.keyHash()表示根据消息key做哈希路由，保证同一个key的消息进到同一个分区。如果不用key，就用RoutingStrategy.roundRobin()做轮询。

把Source和Sink串起来，就是一条完整管道：

public class PulsarFlinkPipeline { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000); PulsarSource<String> source = ... // 上面代码 PulsarSink<String> sink = ... // 上面代码 DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "pulsar-source"); stream.map(new MapFunction<String, String>() { @Override public String map(String value) throws Exception { // 业务处理逻辑，比如清洗、过滤、特征提取 return "processed-" + value; } }).sinkTo(sink); env.execute("flink-pulsar-pipeline"); } }

4.4 用Admin API和命令行验证链路是否通

代码写完先别急着提交集群任务，我习惯先在Pulsar端造点数据验证链路：

# 创建主题 bin/pulsar-admin topics create persistent://public/default/output-topic # 生产测试消息 bin/pulsar-client produce persistent://public/default/input-topic --messages "hello-pulsar-1,hello-pulsar-2,hello-pulsar-3" # 消费确认 bin/pulsar-client consume persistent://public/default/output-topic -s test-sub -n 3

如果命令行能正常收到processed-*前缀的消息，说明Source、Sink、订阅、游标、序列化器这些环节全部打通了，再往Flink集群上部署。这个验证习惯帮我省了至少七八次"任务提交失败但不知道是代码问题还是环境问题"的排查时间。

5. 精确一次语义：Flink检查点如何与Pulsar事务协作

5.1 三种一致性级别的实现逻辑

实时管道里最怕数据重复或丢失，尤其在做金融风控、库存扣减这类场景。Flink的Pulsar连接器提供三个级别的一致性：

至少一次（At-Most-Once）：Source不保存位点，故障恢复后可能从更早的位置重新消费，也可能跳过部分消息，一般只用于吞吐优先的可容忍丢数据的场景。
至少一次（At-Least-Once）：Source把消息先推给下游，等Flink检查点完成后再确认，故障恢复时从头重放，保证不丢，但可能出现重复。
精确一次（Exactly-Once）：在"至少一次"的基础上，下游写入端配合Pulsar事务做原子提交，保证消息不重不漏。

Flink实现精确一次的链路是这样走的：Flink的作业检查点触发时，Source端把当前消费位置和所有Pending消息固化到状态后端，同时向JobManager汇报；检查点完成之后，Source端一次性确认这一批消息。下游Pulsar Sink在写出时开启事务，把一批消息写在事务里，等检查点完成信号到达后提交事务。这样设计的好处是，消息写入Pulsar和数据位点提交之间有了强一致关联——要么整批提交并推进游标，要么整批回滚重新消费。

5.2 Pulsar事务开启的Broker配置

Pulsar从2.7.0开始正式支持事务，但需要Broker显式开启事务协调器。在conf/standalone.conf或conf/broker.conf中确认配置项：

transactionCoordinatorEnabled=true transactionCoordinatorReplicas=1 transactionMetadataStoreProviderClassName=org.apache.bookkeeper.transaction.tools.provider.SimpleTransactionMetadataStoreProvider transactionBufferSchedulerInPartitionedTopic=1

Flink连接器做精确一次写入时，如果检测到Broker没开事务，会直接抛异常而不是降级到至少一次，这是好事——宁可失败也不要悄悄降级，否则你以为管道是精确一次，实际却可能重复消费。

这里多说一句关于生产环境的话题。Pulsar事务在3.x版本已经相当稳定，但事务在部分场景下仍有一定性能开销，尤其是吞吐量特别大的集群（GB/s级别），开启事务后Broker端CPU会增加大概5%到10%。如果你的业务场景可以接受少量重复数据，我建议用AT_LEAST_ONCE跑，把精确一次留给真正需要的位置，比如下游是MySQL、Redis、Elasticsearch这些需要幂等写入的系统。

5.3 幂等设计比一致性级别更重要

这个观点值得单列出来：一致性语义是系统层面的兜底，但不应该是唯一防线。我在项目里始终贯彻一套原则——下游写入尽量做幂等设计。比如往Redis写用户行为计数，可以用时间戳字段做最后的写覆盖；往HBase写特征值，用RowKey带业务唯一ID并做版本冲突检查；往MySQL写订单状态，用INSERT ON DUPLICATE KEY UPDATE。这样即使上游某次故障重放导致重复数据，下游也不会被重复处理污染。

Flink和Pulsar的集成再完美，也架不住下游存储不支持事务或者写入了修改数据的记录，所以思想要同时放到两个层面：计算引擎层的语义保障，以及数据应用层的幂等兜底。

6. 生产踩坑实录：从连接器版本到JDBC异常，排查链路完整还原

6.1 坑一：连接器版本与服务端版本错配，报错藏在序列化基础上

我们第一次在测试环境联调时，遇到一个非常诡异的报错：任务能提交，Source能拉消息，但每处理几条就抛org.apache.pulsar.shaded...开头的异常，堆栈指向protobuf反射代码。当时第一反应是Pulsar客户端和服务端的protobuf协议冲突，逐条检查了依赖树，发现项目中还引了一个老版本的pulsar-client-admin，是之前做监控面板时残留的，它把com.google.protobuf拉到旧版本，导致连接器内部使用的shaded class在运行时发生NoSuchMethodError。

排查链路是这样的：

看到异常堆栈，先定位是客户端侧还是服务端侧——抓Broker日志，Broker并没有报错，说明问题出在客户端本地。
在Flink作业的pom.xml执行mvn dependency:tree -Dincludes=org.apache.pulsar，发现存在三个不同版本的pulsar-client。
用mvn dependency:exclusions排除多余版本，统一到连接器声明版本。
重新打包提交测试，问题消失。

同类问题在引入flink-connector-jdbc时也会出现，所以团队后来定了一条规矩：凡是跟Flink Connector打交道的组件，一律先做依赖树净检，再上线。这事不费时间，却省下大量夜间排查成本。

6.2 坑二：长时间Checkpoint超时导致消息堆积和游标不前进

上线后的第二个星期，监控告警说消费延迟持续上涨，我的第一反应是处理逻辑太慢导致背压。点开Flink UI看Metrics，发现Source端Current Fetch Event Time Lag不高，处理算子也没有明显Backpressure，但Checkpoint连续失败，最长一次超过10分钟。

顺着Checkpoint的失败原因去看，发现根因出在Pulsar Sink的事务提交上。当时的任务设置了10分钟的checkpoint间隔，但Sink的Pulsar生产者事务在Pulsar Broker端处于Pending状态，等Flink发出commit信号后，事务协调器因为超时阈值配置太小无法按时确认提交，导致整个checkpoint超时，然后Flink回滚重新弄，消息就在外部堆积了。

解决方式分两步：

调大Pulsar事务超时时间，对应配置是transactionCoordinatorTransactionTtlMs，我改成了60分钟，跟Flink的checkpoint超时上限拉开安全距离。
优化Flink侧checkpoint配置，原来是固定间隔触发，改成CheckpointConfig.enableUnalignedCheckpoints(true)，并设置setCheckpointTimeout(15min)。非对齐检查点在消息堆积时不会强制Source端停止消费，吞吐曲线明显平滑多了。

这个坑给我的经验是：Pulsar事务协调器、Flink检查点间隔、Sink事务提交这三个指标必须联动观察，单一指标看着都正常，串起来却是链路的隐蔽瓶颈。

6.3 坑三：消费位点回溯失败，真相是SubscriptionName冲突

另外一次比较迷惑的坑，是我想用一个Flink SQL任务重新消费某个主题的历史数据来做补算，直接改了下setStartCursor(StartCursor.earliest())，重启后却发现还是从之前的位置消费，新数据也不断刷进来，但历史消息一条都没有。

排查之后发现问题出在订阅名上。Flink Pulsar Source的游标是按(topic, subscriptionName)维度保存的，同一个订阅名下，无论你重启多少次Flink作业，Pulsar服务端记录的Cursor位置都保留在上次checkpoint提交的地方，StartCursor只在"订阅不存在"时才生效。要重新从最早消费，必须换一个全新的订阅名，或者在Pulsar端先把旧订阅删除。

bin/pulsar-admin subscriptions delete persistent://public/default/input-topic -s flink-subscription

删除后重新启动Flink作业，换一个新的订阅名，历史数据顺利重放。这次踩坑让我牢牢记住一个原则：订阅是状态的载体，改消费位点要认清谁才是状态的持有者。

6.4 和JDBC连接器联动时的经典异常：事务提交与外部存储的不一致

热词里有人搜"flink的jdbc连接器异常"，我猜大概率碰到了这么一种场景——Flink从Pulsar消费数据，经过处理后通过JDBC Sink写入MySQL，写一半任务故障，恢复后Pulsar重放数据，MySQL出现重复主键插入异常。这是典型的分布式事务边界问题，Pulsar这端的消息可以精确一次重放，但MySQL那端的写入没有幂等保障。

我的做法是这样：先启用MySQL连接器的事务提交回退功能，配置setTransactionTimeoutMillis和setIsolationLevel，确保Flink在两次检查点之间把DB写入操作放在同一事务；同时给目标表加上业务幂等键，插入语句改成ON DUPLICATE KEY UPDATE。如果目标库不支持这种语法，就在Flink侧维护一个状态去重Layer，消费时先查一条Key，存在就跳过。这本质上是把Pulsar端的消息幂等性延伸到下游存储边界，连接器的异常才能被解耦掉。

6.5 排查链路的方法论沉淀

踩过这些坑之后，组里沉淀了一套排查SOP，写在这里供你参考：

第一步：看Flink UI的Checkpoint指标和Backpressure指标，锁定故障层在Source、算子还是Sink。
第二步：抓Pulsar Broker日志的TransactionTimeout、AckTimeout、SubscriptionCursorReset相关关键词，判断是不是游标和事务侧的问题。
第三步：核对连接器版本与服务端版本、依赖树净检。
第四步：用最小复现任务（同时读写同一个测试主题）替换业务算子，快速确认是连接器问题还是业务代码问题。
第五步：落库排查加入幂等保障，确保非连接器因素不污染后续分析。

这套SOP现在直接写进了团队的Flink任务排障文档里，任何新同学接手都能照着走。

7. 性能调优与上线建议：并行度、批大小、资源配比

7.1 并行度怎么定：分区数不是唯一标准

Flink Pulsar Source的并行度由两个因素决定：主题分区数和连接器内部的PulsarSourceEnumerator分配策略。默认情况下，一个Flink子任务可以处理多个Pulsar分区，分区数越少，单个子任务的吞吐压力越大。建议的设法是：

如果主题分区数远大于Flink可用slot数，让连接器自动分配即可，它会做均匀分摊。
如果分区数比slot数少，建议把Pulsar主题分区数至少扩到slot数的2倍，避免节点故障时单点热点过重。
在机器性能异构明显的集群，可以开启连接器的setPollTimeoutMs和setMaxFetchRecords推进弹性吞吐，这两个参数控制单次拉取消息的数量和超时时间。

我这边数据量峰值大约120万条每秒，8个Broker，3台Flink TaskManager各有16个slot，Source并行度设在24，Sink并行度设在16，整体跑下来没有明显背压。

7.2 写在最后的调优细节

有几个小实践不一定是标准答案，但在我这个场景都被验证有效：

给Pulsar生产端开启setBatchingMaxMessages(1000)和setBatchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)，批量发送可以显著降低客户端到Broker的RPC次数。Flink Sink端会自动继承这些参数，只需在连接器构建时同步设置。
开启Pulsar端到端压缩，用CompressionType.LZ4或ZSTD（CPU换网络带宽），大消息场景收益明显。
给Flink作业分配独立内存管理，taskmanager.memory.process.size不要低于2GB，因为Pulsar连接器内部会持有较大的发送和接收缓冲区。
实时链路尽量别在Flink SQL里直接写Pulsar的DDL，DDL只适合做调试，生产上用DataStream API构建管道能更精细控制状态TTL、Checkpoint配置和分隔路由规则。

8. 关于集成方案的最终体验与选择倾向

我个人的真实体会是，Flink和Pulsar的集成在1.15连接器成熟之后，已经可以放心作为生产链路的底座来用。相比Kafka，它的优势集中体现在数据回放弹性、Broker无状态扩容和订阅模型的灵活性上；代价是运维多一个BookKeeper集群需要照料，团队的体系化能力要求更高。用Kafka还是用Pulsar，本质上没有绝对答案，取决于你对"可扩展、可回放、多租户"的需求优先级。

如果让我给一个最直接的落地建议：小团队、标准消费组场景、已有Kafka运维经验，继续待在Kafka体系里没有错；如果业务增长迅速、需要频繁重算历史数据、或者未来要做跨地域多活，Pulsar+Flink值得认真投入。最后再分享一个实用策略——上线新管道之前，先搭一个影子管道，把线上10%的流量切进去跑一周，观察消费延迟、检查点稳定性、事务提交成功率这三个核心指标再全量切换。这套流程帮我们避开了很多上线后才暴露的隐性坑，在实时系统领域，速度诚然重要，稳，才是真正的长期主义。

Flink与Pulsar集成实战：架构、连接器与生产实践