flink1.16使用消费/生产kafka之DataStream-Toy模板网

这篇具有很好参考价值的文章主要介绍了flink1.16使用消费/生产kafka之DataStream。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

flink高级版本后，消费kafka数据一种是Datastream 一种之tableApi。

上官网 Kafka | Apache Flink

Kafka Source

引入依赖 flink和kafka的连接器，里面内置了kafka-client

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.16.2</version>
</dependency>

使用方法

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

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

很简单一目了然。

topic和partition

多个topic
KafkaSource.builder().setTopics("topic-a", "topic-b");
正则匹配多个topic
KafkaSource.builder().setTopicPattern("topic.*");
指定分区
final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList(
        new TopicPartition("topic-a", 0),    // Partition 0 of topic "topic-a"
        new TopicPartition("topic-b", 5)));  // Partition 5 of topic "topic-b"
KafkaSource.builder().setPartitions(partitionSet);

反序列化

import org.apache.kafka.common.serialization.StringDeserializer;

KafkaSource.<String>builder()
        .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class));

其实就是实现接口 DeserializationSchema 的deserialize()方法把byte转为你想要的类型。

起始消费位点

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 会退出运行。

估计99的人都用不到这个。也就是设置一个结束的offset的点

其他属性

除了上述属性之外，您还可以使用 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 秒检查一次新分区

分区检查功能默认 不开启。需要显式地设置分区检查间隔才能启用此功能。

这个是为了扩容的，因为kafka的消费能力和分区有关，消费能力不够的时候需要动态增加分区

事件时间和水印

默认情况下，Kafka Source 使用 Kafka 消息中的时间戳作为事件时间。您可以定义自己的水印策略（Watermark Strategy）以从消息中提取事件时间，并向下游发送水印：

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

消费位点提交

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 端进行监控。

这里是说明消费的offset是由kafka管理还是flink管理。

举个例子

.setProperty("enable.auto.commit","true")
.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"100")

并且env没有 env.enableCheckpoint()

此时只要消息进入到flink，过了100ms就会被认为已经消费过了offset会+1 不管你这个消息是否处理完，是否处理失败了。

但是如果你env.enableCheckpoint()，那么此时就是由checkpoint被提交的之前提交offset了。和checkpoint息息相关。checkpoint如果失败了那么这个offset就不会被提交了。

kafka安全认证

KafkaSource.builder()
    .setProperty("security.protocol", "SASL_PLAINTEXT")
    .setProperty("sasl.mechanism", "PLAIN")
    .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");

上面这个比较常用

另一个更复杂的例子，使用 SASL_SSL 作为安全协议并使用 SCRAM-SHA-256 作为 SASL 机制：

KafkaSource.builder()
    .setProperty("security.protocol", "SASL_SSL")
    // SSL 配置
    // 配置服务端提供的 truststore (CA 证书) 的路径
    .setProperty("ssl.truststore.location", "/path/to/kafka.client.truststore.jks")
    .setProperty("ssl.truststore.password", "test1234")
    // 如果要求客户端认证，则需要配置 keystore (私钥) 的路径
    .setProperty("ssl.keystore.location", "/path/to/kafka.client.keystore.jks")
    .setProperty("ssl.keystore.password", "test1234")
    // SASL 配置
    // 将 SASL 机制配置为 as SCRAM-SHA-256
    .setProperty("sasl.mechanism", "SCRAM-SHA-256")
    // 配置 JAAS
    .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"password\";");

如果在作业 JAR 中 Kafka 客户端依赖的类路径被重置了（relocate class），登录模块（login module）的类路径可能会不同，因此请根据登录模块在 JAR 中实际的类路径来改写以上配置。

Kafka Sink

KafkaSink 可将数据流写入一个或多个 Kafka topic。

使用方法

DataStream<String> stream = ...;
        
KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers(brokers)
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
            .setTopic("topic-name")
            .setValueSerializationSchema(new SimpleStringSchema())
            .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();
        
stream.sinkTo(sink);

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

Bootstrap servers, setBootstrapServers(String)
消息序列化器（Serializer）, setRecordSerializer(KafkaRecordSerializationSchema)
如果使用DeliveryGuarantee.EXACTLY_ONCE 的语义保证，则需要使用 setTransactionalIdPrefix(String)

序列化器

KafkaRecordSerializationSchema.builder()
    .setTopicSelector((element) -> {<your-topic-selection-logic>})
    .setValueSerializationSchema(new SimpleStringSchema())
    .setKeySerializationSchema(new SimpleStringSchema())
    .setPartitioner(new FlinkFixedPartitioner())
    .build();

其中消息体（value）序列化方法和 topic 的选择方法是必须指定的，此外也可以通过 setKafkaKeySerializer(Serializer) 或 setKafkaValueSerializer(Serializer) 来使用 Kafka 提供而非 Flink 提供的序列化器。

容错

KafkaSink 总共支持三种不同的语义保证（DeliveryGuarantee）。对于 DeliveryGuarantee.AT_LEAST_ONCE 和 DeliveryGuarantee.EXACTLY_ONCE，Flink checkpoint 必须启用。默认情况下 KafkaSink 使用 DeliveryGuarantee.NONE。以下是对不同语义保证的解释：

一旦启用了基于 Kerberos 的 Flink 安全性后，只需在提供的属性配置中包含以下两个设置（通过传递给内部 Kafka 客户端），即可使用 Flink Kafka Consumer 或 Producer 向 Kafk a进行身份验证：

DeliveryGuarantee.NONE 不提供任何保证：消息有可能会因 Kafka broker 的原因发生丢失或因 Flink 的故障发生重复。
DeliveryGuarantee.AT_LEAST_ONCE: sink 在 checkpoint 时会等待 Kafka 缓冲区中的数据全部被 Kafka producer 确认。消息不会因 Kafka broker 端发生的事件而丢失，但可能会在 Flink 重启时重复，因为 Flink 会重新处理旧数据。
DeliveryGuarantee.EXACTLY_ONCE: 该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），在 Flink 发生重启时不会发生数据重复。然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务不会互相影响！此外，强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，否则 Kafka 对未提交事务的过期处理会导致数据丢失。
启用 Kerberos 身份验证
Flink 通过 Kafka 连接器提供了一流的支持，可以对 Kerberos 配置的 Kafka 安装进行身份验证。只需在 flink-conf.yaml 中配置 Flink。像这样为 Kafka 启用 Kerberos 身份验证：
1.通过设置以下内容配置 Kerberos 票据
security.kerberos.login.use-ticket-cache：默认情况下，这个值是 true，Flink 将尝试在 kinit 管理的票据缓存中使用 Kerberos 票据。注意！在 YARN 上部署的 Flink jobs 中使用 Kafka 连接器时，使用票据缓存的 Kerberos 授权将不起作用。
security.kerberos.login.keytab 和 security.kerberos.login.principal：要使用 Kerberos keytabs，需为这两个属性设置值。
2。将 KafkaClient 追加到 security.kerberos.login.contexts：这告诉 Flink 将配置的 Kerberos 票据提供给 Kafka 登录上下文以用于 Kafka 身份验证。
将 security.protocol 设置为 SASL_PLAINTEXT（默认为 NONE）：用于与 Kafka broker 进行通信的协议。使用独立 Flink 部署时，也可以使用 SASL_SSL；请在此处查看如何为 SSL 配置 Kafka 客户端。
将 sasl.kerberos.service.name 设置为 kafka（默认为 kafka）：此值应与用于 Kafka broker 配置的 sasl.kerberos.service.name 相匹配。客户端和服务器配置之间的服务名称不匹配将导致身份验证失败。

问题排查

数据丢失

根据你的 Kafka 配置，即使在 Kafka 确认写入后，你仍然可能会遇到数据丢失。特别要记住在 Kafka 的配置中设置以下属性：

acks
log.flush.interval.messages
log.flush.interval.ms
log.flush.*

UnknownTopicOrPartitionException

导致此错误的一个可能原因是正在进行新的 leader 选举，例如在重新启动 Kafka broker 之后或期间。这是一个可重试的异常，因此 Flink job 应该能够重启并恢复正常运行。也可以通过更改 producer 设置中的 retries 属性来规避。但是，这可能会导致重新排序消息，反过来可以通过将 max.in.flight.requests.per.connection 设置为 1 来避免不需要的消息。

ProducerFencedException

这个错误是由于 FlinkKafkaProducer 所生成的 transactional.id 与其他应用所使用的的产生了冲突。多数情况下，由于 FlinkKafkaProducer 产生的 ID 都是以 taskName + "-" + operatorUid 为前缀的，这些产生冲突的应用也是使用了相同 Job Graph 的 Flink Job。我们可以使用 setTransactionalIdPrefix() 方法来覆盖默认的行为，为每个不同的 Job 分配不同的 transactional.id 前缀来解决这个问题。文章来源地址https://www.toymoban.com/news/detail-605321.html

到了这里，关于flink1.16使用消费/生产kafka之DataStream的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！