flink连接kafka

Kafka Source

示例

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("clicks")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

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

以下属性在构建 KafkaSource 时是必须指定的：

• Bootstrap server，通过 setBootstrapServers(String) 方法配置
• 消费者组 ID，通过setGroupId(String) 配置
• 要订阅的 Topic / Partition， 用于解析 Kafka消息的反序列化器（Deserializer）

起始消费位点

Kafka source 能够通过位点初始化器（OffsetsInitializer）来指定从不同的偏移量开始消费 。内置的位点初始化器包括：

KafkaSource.builder()
    // 从消费组提交的位点开始消费，不指定位点重置策略
    .setStartingOffsets(OffsetsInitializer.committedOffsets())
    // 从消费组提交的位点开始消费，如果提交位点不存在，使用最早位点
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    // 从时间戳大于等于指定时间戳（毫秒）的数据开始消费
    .setStartingOffsets(OffsetsInitializer.timestamp(1657256176000L))
    // 从最早位点开始消费
    .setStartingOffsets(OffsetsInitializer.earliest())
    // 从最末尾位点开始消费
    .setStartingOffsets(OffsetsInitializer.latest());

有界 / 无界模式

Kafka Source 支持流式和批式两种运行模式。默认情况下，KafkaSource 设置为以流模式运行，因此作业永远不会停止，直到 Flink 作业失败或被取消。 可以使用 setBounded(OffsetsInitializer) 指定停止偏移量使 Kafka Source 以批处理模式运行。当所有分区都达到其停止偏移量时，Kafka Source 会退出运行。

流模式下运行通过使用 setUnbounded(OffsetsInitializer) 也可以指定停止消费位点，当所有分区达到其指定的停止偏移量时，Kafka Source 会退出运行。

其他属性

除了上述属性之外，您还可以使用 setProperties(Properties) 和 setProperty(String, String) 为 Kafka Source 和 Kafka Consumer 设置任意属性。KafkaSource 有以下配置项：

client.id.prefix，指定用于 Kafka Consumer 的客户端 ID 前缀
partition.discovery.interval.ms，定义 Kafka Source 检查新分区的时间间隔。 请参阅下面的动态分区检查一节
register.consumer.metrics 指定是否在 Flink 中注册 Kafka Consumer 的指标
commit.offsets.on.checkpoint 指定是否在进行 checkpoint 时将消费位点提交至 Kafka broker
Kafka consumer 的配置可以参考 Apache Kafka 文档。

请注意，即使指定了以下配置项，构建器也会将其覆盖：

key.deserializer 始终设置为 ByteArrayDeserializer
value.deserializer 始终设置为 ByteArrayDeserializer
auto.offset.reset.strategy 被 OffsetsInitializer#getAutoOffsetResetStrategy() 覆盖
partition.discovery.interval.ms 会在批模式下被覆盖为 -1

动态分区检查

为了在不重启 Flink 作业的情况下处理 Topic 扩容或新建 Topic 等场景，可以将 Kafka Source 配置为在提供的 Topic / Partition 订阅模式下定期检查新分区。要启用动态分区检查，请将 partition.discovery.interval.ms 设置为非负值：

KafkaSource.builder()
    .setProperty("partition.discovery.interval.ms", "10000"); // 每 10 秒检查一次新分区

env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy");

空闲

如果并行度高于分区数，Kafka Source 不会自动进入空闲状态。您将需要降低并行度或向水印策略添加空闲超时。如果在这段时间内没有记录在流的分区中流动，则该分区被视为“空闲”并且不会阻止下游操作符中水印的进度。 这篇文档 描述了有关如何定义 WatermarkStrategy#withIdleness 的详细信息.

消费位点提交

Kafka source 在 checkpoint 完成时提交当前的消费位点 ，以保证 Flink 的 checkpoint 状态和 Kafka broker 上的提交位点一致。如果未开启 checkpoint，Kafka source 依赖于 Kafka consumer 内部的位点定时自动提交逻辑，自动提交功能由 enable.auto.commit 和 auto.commit.interval.ms 两个 Kafka consumer 配置项进行配置。

注意：Kafka source 不依赖于 broker 上提交的位点来恢复失败的作业。提交位点只是为了上报 Kafka consumer 和消费组的消费进度，以在 broker 端进行监控。