如何基于时间戳获取kafka消息

mbjcgjjk 于 2021-06-04 发布在 Kafka

关注(0)|答案(2)|浏览(646)

我正在开发一个应用程序，其中我使用Kafka和技术是scala。我的Kafka消费代码如下：

val props = new Properties()
        props.put("group.id", "test")
        props.put("bootstrap.servers", "localhost:9092")
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        props.put("auto.offset.reset", "earliest")
        props.put("group.id", "consumer-group")
    val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)
    consumer.subscribe(util.Collections.singletonList(topic))
    val record = consumer.poll(Duration.ofMillis(500)).asScala.toList

它给我所有的记录，但问题是我已经在Kafka消费者的数据，这可能会导致重复的数据意味着相同的关键数据可以在主题中已经存在。有没有什么方法可以让我从某个特定的时间检索数据。是指在轮询之前，我是否可以计算当前时间并仅检索在该时间之后出现的记录。我有什么办法可以做到吗？

scala apache-kafka kafka-consumer-api

来源：https://stackoverflow.com/questions/63140301/how-to-get-kafka-messages-based-on-timestamp

2条答案

按热度按时间

0yg35tkg1#

您可以在kafkaconsumerapi中使用offsetsfortimes方法。

代码

import java.time.Duration
import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import collection.JavaConverters._
object OffsetsForTime extends App {
  implicit def toJavaOffsetQuery(offsetQuery: Map[TopicPartition, scala.Long]): java.util.Map[TopicPartition, java.lang.Long] =
    offsetQuery
      .map { case (tp, time) => tp -> new java.lang.Long(time) }
      .asJava
  val topic = "myInTopic"
  val timestamp: Long = 1595971151000L
  val props = new Properties()
  props.put("group.id", "group-id1337")
  props.put("bootstrap.servers", "localhost:9092")
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  props.put("auto.offset.reset", "earliest")
  val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)
  val topicPartition = new TopicPartition(topic, 0)
  consumer.assign(java.util.Collections.singletonList(topicPartition))
  // dummy poll before calling seek
  consumer.poll(Duration.ofMillis(500))
  // get next available offset after given timestamp
  val (_, offsetAndTimestamp) = consumer.offsetsForTimes(Map(topicPartition -> timestamp)).asScala.head
  // seek to offset
  consumer.seek(topicPartition, offsetAndTimestamp.offset())
  // poll data
  val record = consumer.poll(Duration.ofMillis(500)).asScala.toList
  for (data <- record) {
    println(s"Timestamp: ${data.timestamp()}, Key: ${data.key()}, Value: ${data.value()}")
  }
}

试验

./kafka/current/bin/kafconsole-consumer.sh --bootstrap-server localhost:9092 --topic myInTopic --from-beginning --property print.value=true --property print.timestamp=true
CreateTime:1595971142560    1_old
CreateTime:1595971147697    2_old
CreateTime:1595971150136    3_old
CreateTime:1595971192649    1_new
CreateTime:1595971194489    2_new
CreateTime:1595971196416    3_new

将时间戳选择为 3_old 以及 1_new 只使用“新”消息。

输出

Timestamp: 1595971192649, Key: null, Value: 1_new
Timestamp: 1595971194489, Key: null, Value: 2_new
Timestamp: 1595971196416, Key: null, Value: 3_new

展开查看全部

赞(0）回复(0）举报 2021-06-04

dzjeubhm2#

使用任何给定时间戳的唯一方法是
查找
offsetsForTimes seek 至和 commitSync 这个结果
开始轮询
但是，您需要意识到数据流是连续的，以后可能还会有重复的键。
如果您在数据中有相同的键，您只希望看到最新的，那么最好使用ktable

赞(0）回复(0）举报 2021-06-04

我来回答

如何基于时间戳获取kafka消息

2条答案

代码

试验

输出

相关问题

热门标签

最新问答