Hi Michael, got it. I understand that it would be less error-prone to generate the final "altered" timestamp on the Producer side, instead of trying to compute it each time the record is consumed.
Thanks. Nicolas. 2017-01-16 10:03 GMT+01:00 Michael Noll <mich...@confluent.io>: > Nicolas, > > quick feedback on timestamps: > > > In our system, clients send data to an HTTP API. This API produces the > > records in Kafka. I can't rely on the clock of the clients sending the > > original data, (so the records' timestamps are set by the servers > ingesting > > the records in Kafka), but I can rely on a time difference. The client > only > > gives information about the time spent since the first version of the > > record was sent. Via a custom timestamp extractor, I just need to > subtract > > the time spent to the record's timestamp to ensure that it will fall in > > same window. > > Alternatively, you can also let the HTTP API handle the timestamp > calculations, and then embed the "final" timestamp in the message payload > (like the messave value). Then, in your downstream application, you'd > define a custom timestamp extractor that returns this embedded timestamp. > > One advantage of the approach I outlined above is that other consumers of > the same data (who may or may not be aware of how you need to compute a > timestamp diff to get the "real" timestamp) can simply re-use the timestamp > embedded in the payload without having to know/worry about the custom > calculation. It might also be easier for Ops personnel to have access to a > ready-to-use timestamp in case they need to debug or troubleshoot. > > -Michael > > > > > On Sun, Jan 15, 2017 at 11:10 PM, Nicolas Fouché <nfou...@onfocus.io> > wrote: > > > Hi Eno, > > > > 2. Well, records could arrive out of order. But it should happen rarely, > > and it's no big deal anyway. So let's forget about the version number if > it > > makes things easier ! > > > > 3. I completely missed out on KTable aggregations. Thanks a lot for the > > pointer, that opens new perspectives. > > > > ... a few hours pass ... > > > > Ok, in my case, since my input is an infinite stream of new records, I > > would have to "window" my KTables, right ? > > With `KStream.groupBy().reduce()`, I can generate a windowed KTable of > > records, and even use the reducer function to compare the version > numbers. > > Next, I use `KTable.groupBy().aggregate()` to benefit from the `adder` > and > > `substractor` mechanisms [1]. > > > > The last problem is about the record timestamp. If I work on a one-hour > > window, and records are sent between let's say 00:59 and 01:01, they > would > > live in two different KTables and this would create duplicates. > > To deal with this, I could mess with the records timestamps, so each new > > record version is considered by Kafka Streams having the same timestamp > > than the first version seen by the producer. > > Here is my idea: > > In our system, clients send data to an HTTP API. This API produces the > > records in Kafka. I can't rely on the clock of the clients sending the > > original data, (so the records' timestamps are set by the servers > ingesting > > the records in Kafka), but I can rely on a time difference. The client > only > > gives information about the time spent since the first version of the > > record was sent. Via a custom timestamp extractor, I just need to > subtract > > the time spent to the record's timestamp to ensure that it will fall in > > same window. > > Long text, small code: > > https://gist.github.com/nfo/6df4d1076af9da5fd1c29b0ad4564f2a .What do > you > > think ? > > > > About the windowed KTables in the first step, I guess I should avoid > making > > them too long, since they store the whole records. We usually aggregate > > with windows size from 1 hour to 1 month. I should compute all the > > aggregates covering more than 1 hour from the 1-hour aggregates, right ? > > > > [1] > > http://docs.confluent.io/3.1.1/streams/javadocs/org/apache/ > > kafka/streams/kstream/KGroupedTable.html#aggregate( > > org.apache.kafka.streams.kstream.Initializer,%20org. > > apache.kafka.streams.kstream.Aggregator,%20org.apache. > > kafka.streams.kstream.Aggregator,%20org.apache. > kafka.common.serialization. > > Serde,%20java.lang.String) > > > > Thanks (a lot). > > Nicolas > > > > > > 2017-01-13 17:32 GMT+01:00 Eno Thereska <eno.there...@gmail.com>: > > > > > Hi Nicolas, > > > > > > There is a lot here, so let's try to split the concerns around some > > themes: > > > > > > 1. The Processor API is flexible and can definitely do what you want, > but > > > as you mentioned, at the cost of you having to manually craft the code. > > > 2. Why are the versions used? I sense there is concern about records > > > arriving out of order so the versions give each record with the same ID > > an > > > order. Is that correct? > > > 3. If you didn't have the version and the count requirement I'd say > using > > > a KTable to interpret the stream and then aggregating on that would be > > > sufficient. There might be a way to do that with a mixture of the DSL > and > > > the processor API. > > > > > > Another alternative might be to use the Interactive Query APIs ( > > > https://www.confluent.io/blog/unifying-stream-processing- > > and-interactive- > > > queries-in-apache-kafka/ <https://www.confluent.io/ > blog/unifying-stream- > > > processing-and-interactive-queries-in-apache-kafka/>) to first get all > > > your data in KTables and then query it periodically (you can decide on > > the > > > frequency manually). > > > > > > Thanks > > > Eno > > > > > > > > > > On 12 Jan 2017, at 22:19, Nicolas Fouché <nfou...@onfocus.io> wrote: > > > > > > > > Hi, > > > > > > > > long long technical story, sorry for that. > > > > > > > > I'm dealing with a special case. My input topic receives records > > > containing > > > > an id in the key (and another field for partitioning), and a version > > > number > > > > in the value, amongst other metrics. Records with the same id are > sent > > > > every 5 seconds, and the version number increments. > > > > > > > > These metrics in the record value are used in aggregations to compute > > > > `sums` and `counts` (then stored in a DB to compute averages), and to > > > > compute a few other data structures like cumulative time buckets. If > > the > > > > aggregation receives the same record with updated metrics, I have to > > > > decrement `sum` by the metric value of the previous record, and > > increment > > > > `sum` by the new metric value. Also, the `count` would be incremented > > by > > > 1 > > > > only if the record is seen for the first time (which is not the same > as > > > > "version number = 1"). > > > > > > > > To implement this, we would write a processor which would compute the > > > diff > > > > of metrics by storing the last version of each record in its state. > > This > > > > diff is sent to the aggregation, this diff also tells if the record > was > > > the > > > > first (so `count` is incremented). I think this can only written with > > the > > > > low level API. > > > > That could work well, except we have a dozen type of records, with a > > few > > > > metrics each, and quite a few fields to compute in aggregations. Each > > > time > > > > we deal with this type of "duplicate" records, we would have to write > > all > > > > the code to compute the diffs again, and the aggregation algorithm > > > becomes > > > > way less trivial (we deal with cumulative time buckets, if one knows > > > what I > > > > mean). > > > > > > > > So we got another idea, which does not seem to feel right in a > > > *streaming* > > > > environment, and quite inefficient: > > > > > > > > ==== > > > > The goal is to "buffer" records until we're quite sure no new version > > > will > > > > be received. And if a new version is actually received, it's ignored. > > > > A generic low level processor would be used in topologies which > receive > > > the > > > > same records with updated metrics and an incremented version. > > > > > > > > One state store: contains the records, used to know if a record was > > > already > > > > received and when, and if the record was already transferred. > > > > > > > > Algorithm: > > > > > > > > On each new record: > > > > - GET the previous record in the store by Key > > > > - ignore the new record if: > > > > -- the record version is lower than the one in the store > > > > -- the record timestamp is at least 5 minutes newer than the one in > > store > > > > - PUT (and thus replace) the record in the store > > > > > > > > Every 1 minute: > > > > - for each record in the store > > > > -- if the record has the field "forwarded == true" > > > > --- DELETE it from the store if the record is one hour old > > > > -- else > > > > --- if the timestamp is more that 5 minutes old > > > > ---- PUT the record in the store with the field "forwarded" set to > true > > > > ---- forward the record > > > > === > > > > > > > > Caveats: > > > > - low-level processors don't have access to the record's ingestion > > > > timestamp. So we would have to add it to the record value before > > > producing > > > > the record. > > > > - no secondary indexes, so we do complete iterations on each > > `ponctuate` > > > > - it feels so wrong > > > > > > > > Any suggestions ? It feels like a KStream of KTable records... > > > > > > > > Thanks. > > > > > > > > >
