代码之家 › 专栏 › 技术社区 › xmar

Kafka流:是否可以使用不同的时间戳进行删除,而不是流处理?

apache-kafka-streams apache-kafka

xmar · 技术社区 · 8 年前

在卡夫卡流2.0。

我的用例:能够(部分)使用事件创建时间(用户从原始数据定义并通过TimestampExtractor设置)从重新处理应用程序的历史开始处重新处理数据,并与长时间运行的不停止应用程序一起运行,向输出主题发送数据(两个应用程序都将读取并发送到相同的输出主题,这些主题用于生成状态)。

存储是根据这些主题构建的,包括按会话打开窗口。想象一下,我想给这些主题一个月的保留期(对于无序事件和消耗),当重新处理时,如果使用事件时间,我将处理(和生成)一个月以上的事件。

使用 message.timestamp.type=LogAppendTime 根据 KIP-32 为了避免删除,将在状态存储中生成错误的数据(因为时间戳将不正确,并且它们将用于例如会话)。

使用事件时间、保持完全保留和在重新处理完成并消耗完数据后应用清除数据是很乏味的,但这将有助于减少主题的大小—但是,从这些主题构建的存储又如何呢?E、 g.为了在重新处理时保存数据,我必须设置 until 伪无穷大的设置,但DSL创建的存储是(或应该是)只读的,不可操作。

所以,回到标题:

1 回复 | 直到 8 年前

Matthias J. Sax 8 年前

对于流,使用 LogAppendTime 对于重新分区,主题是未配置。还要注意,在重新分区主题中不会丢失任何数据,因为这些主题是用 Integer.MAX_VALUE (参见。 https://cwiki.apache.org/confluence/display/KAFKA/KIP-284%3A+Set+default+retention+ms+for+Streams+repartition+topics+to+Long.MAX_VALUE ). 流使用 purgeData 用于在数据被使用后从重新分区主题中删除数据的API(cf。 https://issues.apache.org/jira/browse/KAFKA-6150 )以避免无限生长。

因此,我建议通过 log.message.timestamp.type (即,主题级配置)。

推荐文章

Stella · Kafka流处理期间处理数据时出现异常

8 年前

Stella · kafka流中的处理器节点

8 年前

mukesh210 · 卡夫卡流:Lib Rocks数据库上的UnsatifiedLinkError

8 年前

el323 · Kafka流:通过Json日志中的键进行分组

8 年前

Suchita · 如何使用Avro反序列化器和Avro序列化器为KafkaStreams编写Junit测试用例

8 年前

Cemo · 如何获得有关GlobalKTable状态存储更新的通知?

8 年前

px5x2 · Kafka将GlobalKTable同步流传输到应用程序

8 年前

Dth · 加工机为Kafka流中的单个任务多次调用init()

8 年前

nattyddubbs · Kafka Streams多实例设计含义

8 年前

dvlcis · 如何使用卡夫卡流窗口为烛台图生成创建一条记录

8 年前