java

7.1 连接器：Kafka Source/Sink

Flink Kafka 连接器深度配置：KafkaSource 与 KafkaSink 的依赖版本匹配、topic/partition 订阅、offset 初始化策略 earliest/latest/committed、反序列化 schema 自定义、exactly-once 两阶段提交语义，以及 offset 提交与分区分配的真实行为。

Leeting Yan 2026-11-01 3 分钟阅读 1229 字

Flink Kafka Connector 版本

Flink 的 Kafka Connector 独立于 Flink 核心版本：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>3.3.0-1.20</version>  <!-- connector 3.3.0 for Flink 1.20 -->
</dependency>

版本号规则：{connector-version}-{flink-version}

Kafka Source

import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

KafkaSource<Event> source = KafkaSource.<Event>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("eventd-cleaned")
    .setGroupId("eventd-analytics")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new EventDeserializationSchema())
    .build();

DataStream<Event> stream = env.fromSource(
    source,
    WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withTimestampAssigner((event, ts) -> event.getEventTime()),
    "Kafka Source"
);

不写 withTimestampAssigner 也能编译：此时用 Kafka 消息自带的 record timestamp 作为事件时间。只有当消息时间戳就是业务事件时间时才该省略；否则窗口结果会随生产者写入时刻漂移。

Offset 初始化策略

策略	行为
`OffsetsInitializer.earliest()`	从 Topic 最早消息开始
`OffsetsInitializer.latest()`	从最新消息开始（不消费历史）
`OffsetsInitializer.committedOffsets()`	从 Consumer Group 已提交的 offset 开始
`OffsetsInitializer.offsets(Map)`	从指定 offset 开始

多个 Topic

Topic 在 builder 阶段指定。KafkaSource 是构建完成后的不可变对象，没有 setTopics 方法：

KafkaSource<Event> source = KafkaSource.<Event>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("topic1", "topic2", "topic3")
    // 或正则匹配：.setTopicPattern(Pattern.compile("eventd-.*"))
    .setValueOnlyDeserializer(new EventDeserializationSchema())
    .build();

setTopics 与 setTopicPattern 互斥，同时设置会在 build() 时抛异常。

自定义 DeserializationSchema

public class EventDeserializationSchema implements DeserializationSchema<Event> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public Event deserialize(byte[] message) throws IOException {
        return mapper.readValue(message, Event.class);
    }

    @Override
    public boolean isEndOfStream(Event nextElement) {
        return false;  // 无界流永不到达 end
    }

    @Override
    public TypeInformation<Event> getProducedType() {
        return TypeInformation.of(Event.class);
    }
}

Kafka Sink

import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;

// SimpleStringSchema 的泛型是 String，Sink 也只能是 KafkaSink<String>；
// 业务对象 Result 要先序列化成 JSON 再进 Sink
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("eventd-metrics")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("eventd-analytics-tx")
    .build();

stream.map(JsonMapper::toJson).sinkTo(sink);

投递语义

语义	配置	说明
`NONE`	无保证	最快，可能丢失
`AT_LEAST_ONCE`	重试到成功	不丢，可能重复
`EXACTLY_ONCE`	事务	不丢不重复，但吞吐下降

Exactly-Once Sink

Flink Kafka Sink 的 Exactly-Once 基于 Kafka 事务：

Checkpoint Barrier 到达 Sink
    → Sink 开始 Kafka 事务
    → 写入消息到 Topic（预提交）
    → Checkpoint 完成
    → Sink commit Kafka 事务
    → 消息变为可读

如果 Checkpoint 失败 → 事务 abort → 消息不可见

注意三点：必须调用 setTransactionalIdPrefix(...)（否则 build() 报错），Kafka 版本要 >= 0.11；下游消费者要以 isolation.level=read_committed 读取，否则仍能读到未提交事务里的消息。

分区映射

Kafka Source 的并行度与 Kafka 分区的映射：

Kafka Topic 有 8 个分区
Flink Source 并行度 = 4
    → 每个 Flink 实例消费 2 个 Kafka 分区

Kafka Topic 有 4 个分区
Flink Source 并行度 = 8
    → 4 个实例空闲（浪费）

最佳实践：Flink Source 并行度 = Kafka 分区数。

offset 提交与分区分配

这里最容易照搬普通 Kafka 消费者的默认值。KafkaSource 的 builder 在你没有显式设置时会把 enable.auto.commit 写成 false（3.3.0 起如果显式设成 true 会尊重这个设置，并要求配 group.id，但那等于退回客户端自动提交），所以：

默认没有每 5 秒自动提交。offset 随 Checkpoint 提交（commit.offsets.on.checkpoint，默认 true）；Checkpoint 完成前，Kafka 里的 committed offset 不会前进。
分区分配由 Flink 的 Source 枚举器下发，通过 setGroupId 让多个作业共享消费进度，而不是依赖客户端 rebalance 回调。
没开 Checkpoint 就没有提交。作业重启时从 setStartingOffsets 指定的位置重新开始，与上次跑到哪里无关。

// 想让 offset 与状态快照解耦（例如只做无状态转发），才考虑关掉它；
// 关掉后必须自己保证提交策略，否则重启会从头/尾重读
.setProperty("commit.offsets.on.checkpoint", "false")

分区数变化或并行度调整后，未完成的 Checkpoint 不能直接复用；从 Savepoint 恢复时 Flink 会按新并行度重新分配 Source 状态。

小结

概念	说明
`OffsetsInitializer`	起始 offset 策略
`DeserializationSchema`	自定义反序列化
`DeliveryGuarantee`	Sink 投递语义
`setTransactionalIdPrefix()`	EOS 事务标识前缀
并行度匹配	Source 并行度 ≈ Kafka 分区数

下一节 7.2 连接器：JDBC / Redis / Elasticsearch 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页