廊坊网站建设策划,全球速卖通官网入口,顺企网官网电话,网站美工的重要性版本
flink 1.16.0kafka 2.3
流程描述#xff1a; flink利用KafkaSource#xff0c;读取kafka的数据#xff0c;然后经过一系列的处理#xff0c;通过KafkaSink#xff0c;采用 EXACTLY_ONCE 的模式#xff0c;将处理后的数据再写入到新的topic中。 问题描述#xff1…版本
flink 1.16.0kafka 2.3
流程描述 flink利用KafkaSource读取kafka的数据然后经过一系列的处理通过KafkaSink采用 EXACTLY_ONCE 的模式将处理后的数据再写入到新的topic中。 问题描述 数据写入到新的topic后过上几分钟的时间利用工具offset explorer观察对应topic的数据量显示为0。 刚写入没多久的数据消失了 大写的懵 定位问题
首先查看kafka的日志 阅读flink 官方文档 kafkaSink的介绍 DeliveryGuarantee.EXACTLY_ONCE: In this mode, the KafkaSink will write all messages in a Kafka transaction that will be committed to Kafka on a checkpoint. Thus, if the consumer reads only committed data (see Kafka consumer config isolation.level), no duplicates will be seen in case of a Flink restart. However, this delays record visibility effectively until a checkpoint is written, so adjust the checkpoint duration accordingly. Please ensure that you use unique transactionalIdPrefix across your applications running on the same Kafka cluster such that multiple running jobs do not interfere in their transactions! Additionally, it is highly recommended to tweak Kafka transaction timeout (see Kafka producer transaction.timeout.ms)» maximum checkpoint duration maximum restart duration or data loss may happen when Kafka expires an uncommitted transaction. 翻译过来的意思大概就是
在EXACTLY_ONCE这种模式下KafkaSink在事务中写入所有的消息这些消息在checkpoint上提交给kafka。因此在flink重启的情况下如果消费者值读取提交的数据不会看到重复的数据。缺点就是延迟记录可见性知道写入检查点为止。强烈建议调整kafka的事务超时时间(见Kafka producer transaction.timeout.ms)超时时间要大于【最大检查点持续时间最大重启持续时间】否则当Kafka过期未提交的事务时可能会发生数据丢失。
阅读kafka的官网介绍
Producer Configs transaction.timeout.ms60000默认值 参数描述 The maximum amount of time in ms that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction.If this value is larger than the transaction.max.timeout.ms setting in the broker, the request will fail with a InvalidTransactionTimeout error. Broker Configs transaction.max.timeout.ms900000默认值 参数描述 The maximum allowed timeout for transactions. If a client’s requested transaction time exceed this, then the broker will return an error in InitProducerIdRequest. This prevents a client from too large of a timeout, which can stall consumers reading from topics included in the transaction. 最后排查 在flink中设置的超时时间违反了kafka producer对应的参数规定。
解决问题
在kafkaSink的配置中加入
Properties properties new Properties();
// 根据上面的介绍自己计算这边的超时时间满足条件即可
properties.setProperty(transaction.timeout.ms,900000);KafkaSinkString sink KafkaSink.Stringbuilder().setBootstrapServers(bootstrapServers).setRecordSerializer(KafkaRecordSerializationSchema.Stringbuilder().setTopic(sinkTopic).setValueSerializationSchema(new SimpleStringSchema()).build()).setKafkaProducerConfig(properties).setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE).setTransactionalIdPrefix(flink-xhaodream-).build();总结
在使用现有框架和工具的时候往往只是懂得怎么用具体底层的逻辑、原理了解的很少。往往只有真正理解了原理遇到了问题才会更快、更准确的定位问题、解决问题。
