How can I get messages or data from the Kafka cluster for a specified day. For example 13 September, can anyone provide me code for this. I have googled it and found only theory but I want the code
标签:
apache-kafka
相关问题
- Delete Messages from a Topic in Apache Kafka
- Serializing a serialized Thrift struct to Kafka in
- Kafka broker shutdown while cleaning up log files
- Getting : Error importing Spark Modules : No modul
- How to transform all timestamp fields when using K
相关文章
- Kafka doesn't delete old messages in topics
- Kafka + Spark Streaming: constant delay of 1 secon
- Spring Kafka Template implementaion example for se
- How to fetch recent messages from Kafka topic
- Determine the Kafka-Client compatibility with kafk
- Kafka to Google Cloud Platform Dataflow ingestion
- Kafka Producer Metrics
- Spark Structured Streaming + Kafka Integration: Mi
There is no access method for this. Also, before Kafka
v0.10
messages do not contain any timestamp information, thus, it is impossible to know when a message was written into a topic.As of Kafka
v0.10
each message contains a meta data timestamp attribute, that is either set by the producer on message creation time, or by the broker on message insertion time. A time-based index is planned, but not available yet. Thus, you need to consume the whole topic and check the timestamp field (and ignore all messaged you are not interested in). To find the beginning, you could also do a binary search with regard to offsets and timestamps to find the first message faster.Update:
Kakfa
0.10.1
adds a time-based index. It allows toseek
to the first record with a timestamp equals or larger of the given timestamp. You can use it viaKafkaConsumer#offsetsForTime()
. This will return the corresponding offsets and you can feed them intoKafkaConsumer#seek()
. You can just consume the data and check the records timestamp field viaConsumerRecord#timestamp()
to see when you can stop processing.Note, that data is strictly ordered by offsets but not by timestamp. Thus, during processing, you might get "late" records with smaller timestamp than your start timestamp (you could simple skip over those records though).
A more difficult problem is late arriving record at the end of your search interval though. After you got the first timestamp with a larger timestamp than your search interval, there might still be records with timestamp that are part of your search interval later on (if those records did got appended to the topic "late"). There is no way to know that though. Thus, you might want to keep reading "some more" records and check if there are "late" records. How much "some records" means, is a design decision you need to make by yourself.
There is not general guideline though -- if you have additional knowledge about your "write pattern" it can help to define a good strategy to how many records you want to consumer after your search interval "ended". Of course there are two default strategies: (1) stop at the very first record with larger timestamp than you search interval (and effectively ignore any late arriving records -- if you use "log append time" configuration this is of course a safe strategy); (2) you read to the end of the log -- this is the safest strategy with regard to completeness but might result in prohibitive overhead (also note, as record can be appended any time and if record "delay" could be arbitrary large, a late record might even be append after you reach end-of-log).
In practice, it might be a good idea to think about a "max expected delay" and read until you get a record with larger timestamp than this upper delay bound.
getting data for a specific day from kafka is NOT efficient, as the data is stored linearly inside kafka on each broker's storage system. therefore, even if you have timestamp inside each of your message or use kafka's message metadata which could contain timestamp in the later kafka message version(>=0.10), you still have to scan the entire topic on each partition to get the data. due to the fact that the data inside kafka is not indexed by date but only offset.
remember, kafka is a queue, NOT a database. if you want this date based retrieve pattern, you might want to consider storing kafka message inside another suitable databases system and use timestamp as your index.