1 周后如何处理来自分布式日志代理（例如 Kafka）的日志？答案

【问题标题】：How to process logs from distributed log broker (Eg Kafka) exactly after 1 week?1 周后如何处理来自分布式日志代理（例如 Kafka）的日志？
【发布时间】：2020-10-22 05:13:17
【问题描述】：

如果我想处理来自 Kafka 的恰好 1 周前的日志，可以进行什么设置？

用例是我维护过去 1 周用户活动的累积统计信息。我对最终的一致性很好，不需要统计数据恰好是 1 周。

我有一个流设置，它处理来自 Kafka 的传入日志并更新统计信息。任何超过 1 周的活动都应从统计数据中删除。我可以实现的一种方法是使用批处理（例如 Spark）从统计数据中删除超过 1 周的活动。

有什么方法可以使用流处理从统计数据中删除超过 1 周的用户活动？各种方法的优缺点是什么？

如果我在 Kafka 中至少使用过一次并且统计数据偏离了基本事实，有什么方法可以定期更正统计数据？

【问题讨论】：

你可以 watermarking 来自 Spark 结构化流的概念！

标签： java python apache-spark apache-kafka apache-storm

【解决方案1】：

如果您的 Kafka 消息具有正确的时间戳，那么您可以获得前一周时间戳的偏移量。所以你可以使用..

Map<TopicPartition,OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition,Long> timestampsToSearch)

documentation 说：

按时间戳查找给定分区的偏移量。这每个分区的返回偏移量是其最早的偏移量时间戳大于或等于给定时间戳对应的分区。

要获取主题分区列表，您可以调用consumer.assignment()（在subscribe() 或assign() 之后）返回分配给消费者的Set<TopicPartition>。地图中的Long 值基本上就是时间戳。因此，对于您案例中的所有键，它将是相同的值（即 1 周大的时间戳）

现在，您有一个Map<TopicPartition, OffsetAndTimestamp>。您现在可以使用seek(TopicPartition partition, long offset) 查找每个偏移量。

consumer.subscribe(topics);
Set<TopicPartition> partitions = consumer.assignment();
Map<TopicPartition, Long> map = new LinkedHashMap<>();
partitions.forEach(partition -> map.put(partition, oneWeekOldTimestamp));
Map<TopicPartition, OffsetAndTimestamp> offsetsMap = consumer.offsetForTimes(map);
offsetsMap.forEach((partition, offsetTimestamp) -> consumer.seek(partition, offsetTimestamp.offset()));

现在，您的消费者将处于一周前的消息位置。所以，当你poll() 时，你从上周到现在进行投票。

您可以更改时间戳以满足您的要求，例如，任何超过 1 周的时间都意味着，从时间戳 0 到上周时间戳。

所有前一周数据均指2weekOldTimestamp - 1weekOldTimestamp。

因此，在这种情况下，您必须寻找2weekOldTimestamp，然后处理每个分区，直到遇到1weekOldTimestamp

【讨论】：