Hi Alessandro, I had a look at your code. Regarding your question whether you use the SpecificAvroSerde correctly, take a look at the following documentation:
https://docs.confluent.io/current/streams/developer-guide/datatypes.html I haven't had the time yet to take a closer look at your problems with the aggregation. I will have a look next week. Have a nice weekend, Bruno On Wed, Apr 17, 2019 at 4:43 PM Alessandro Tagliapietra < tagliapietra.alessan...@gmail.com> wrote: > So I've started with a new app with the archetype:generate as in > https://kafka.apache.org/22/documentation/streams/tutorial > > I've pushed a sample repo here: https://github.com/alex88/kafka-test > The avro schemas are a Metric with 2 fields: timestamp and production and a > MetricList with a list of records (Metric) to be able to manually do the > aggregation. > Right now the aggregation is simple just for the purpose of the sample repo > and to easily see if we're getting wrong values. > > What I wanted to achieve is: > - have a custom generator that generates 1 message per second with > production = 1 with 1 ore more separate message keys which in my case are > the sensor IDs generating the data > - a filter that removes out of order messages by having a state that > stores key (sensorID) -> last timestamp > - a window operation that for this example just sums the values in each 10 > seconds windows > > To show where I'm having issues I've setup multiple branches for the repo: > - *issue-01 <https://github.com/alex88/kafka-test/tree/issue-01>* is the > one I had initially "Failed to flush state store > KSTREAM-AGGREGATE-STATE-STORE-0000000003" that I tried to solve using > > https://stackoverflow.com/questions/55186727/kafka-streams-2-1-1-class-cast-while-flushing-timed-aggregation-to-store > - *issue-02 <https://github.com/alex88/kafka-test/tree/issue-02>* is the > one after I've tried to solve above problem with the materializer (maybe > the SpecificAvroSerde is wrong?) > - *issue-03 <https://github.com/alex88/kafka-test/tree/issue-03>* after > fixing issue-02 (by using groupByKey(Grouped.with(Serdes.String(), new > SpecificAvroSerde<>()))) everything seems to be working, if you let both > the producer and stream running, you'll see that the stream receives 10 > messages (with the timestamp incrementing 1 second for each message) like > this: > > S1 with filtered metric{"timestamp": 160000, "production": 1} > S1 with filtered metric{"timestamp": 161000, "production": 1} > S1 with filtered metric{"timestamp": 162000, "production": 1} > S1 with filtered metric{"timestamp": 163000, "production": 1} > S1 with filtered metric{"timestamp": 164000, "production": 1} > S1 with filtered metric{"timestamp": 165000, "production": 1} > S1 with filtered metric{"timestamp": 166000, "production": 1} > S1 with filtered metric{"timestamp": 167000, "production": 1} > S1 with filtered metric{"timestamp": 168000, "production": 1} > S1 with filtered metric{"timestamp": 169000, "production": 1} > > and at the 10 seconds interval something like: > > S1 with computed metric {"timestamp": 160000, "production": 10} > S1 with computed metric {"timestamp": 170000, "production": 10} > S1 with computed metric {"timestamp": 180000, "production": 10} > > and so on... > Now there are two problems, after stopping and restarting the stream > processor (by sending SIGINT via IntelliJ since I start the class main with > it) it happens: > - sometimes the aggregated count is wrong, if I have it start windowing > for 7 seconds (e.g. seconds 11-17), restart the stream, after restart it > might just emit a value for the new 3 missing seconds (seconds 18-20) and > the aggregated value is 3 not 10 > - sometimes the window outputs twice, in the example where I restart the > stream processor I might get as output > > S1 with filtered metric{"timestamp": 154000, "production": 1} > S1 with computed metric {"timestamp": 160000, "production": 5} > S1 with filtered metric{"timestamp": 155000, "production": 1} > S1 with filtered metric{"timestamp": 156000, "production": 1} > S1 with filtered metric{"timestamp": 157000, "production": 1} > S1 with filtered metric{"timestamp": 158000, "production": 1} > S1 with filtered metric{"timestamp": 159000, "production": 1} > S1 with filtered metric{"timestamp": 160000, "production": 1} > S1 with filtered metric{"timestamp": 161000, "production": 1} > S1 with computed metric {"timestamp": 160000, "production": 10} > S1 with filtered metric{"timestamp": 162000, "production": 1} > > as you can see, window for timestamp 160000 is duplicated > > Is this because the window state isn't persisted across restarts? > My ultimate goal is to have the window part emit only once and resume > processing across restarts, while avoiding processing out of order data > (that's the purpose of the TimestampIncrementalFilter) > > Thank you in advance > > -- > Alessandro Tagliapietra > > > On Tue, Apr 16, 2019 at 9:48 PM Alessandro Tagliapietra < > tagliapietra.alessan...@gmail.com> wrote: > > > Hi Bruno, > > > > I'm using the confluent docker images 5.2.1, so kafka 2.2. > > Anyway I'll try to make a small reproduction repo with all the different > > cases soon. > > > > Thank you > > > > -- > > Alessandro Tagliapietra > > > > > > On Tue, Apr 16, 2019 at 1:02 PM Bruno Cadonna <br...@confluent.io> > wrote: > > > >> Hi Alessandro, > >> > >> What version of Kafka do you use? > >> > >> Could you please give a more detailed example for the issues with the > two > >> keys you see? > >> > >> Could the following bug be related to the duplicates you see? > >> > >> > >> > https://issues.apache.org/jira/browse/KAFKA-7895?jql=project%20%3D%20KAFKA%20AND%20issuetype%20%3D%20Bug%20AND%20status%20in%20(Open%2C%20%22In%20Progress%22%2C%20Reopened)%20AND%20component%20%3D%20streams%20AND%20text%20~%20%22duplicate%22 > >> > >> How do you restart the processor? > >> > >> Best, > >> Bruno > >> > >> On Mon, Apr 15, 2019 at 11:02 PM Alessandro Tagliapietra < > >> tagliapietra.alessan...@gmail.com> wrote: > >> > >> > Thank you Bruno, > >> > > >> > I'll look into those, however average is just a simple thing I'm > trying > >> > right now just to get an initial windowing flow working. > >> > In the future I'll probably still need the actual values for other > >> > calculations. We won't have more than 60 elements per window for sure. > >> > > >> > So far to not manually serialize/deserialize the array list I've > >> created an > >> > Avro model with an array field containing the values. > >> > I had issues with suppress as explained here > >> > > >> > > >> > > >> > https://stackoverflow.com/questions/55699096/kafka-aggregate-with-materialized-with-specific-avro-serve-gives-nullpointerexce/55699198#55699198 > >> > > >> > but I got that working. > >> > So far everything seems to be working, except a couple things: > >> > - if I generate data with 1 key, I correctly get a value each 10 > >> seconds, > >> > if I later start generating data with another key (while key 1 is > still > >> > generating) the windowing emits a value only after the timestamp of > key > >> 2 > >> > reaches the last generated window > >> > - while generating data, if I restart the processor as soon as it > >> starts > >> > it sometimes generates 2 aggregates for the same window even if I'm > >> using > >> > the suppress > >> > > >> > Anyway, I'll look into your link and try to find out the cause of > these > >> > issues, probably starting from scratch with a simpler example > >> > > >> > Thank you for your help! > >> > > >> > -- > >> > Alessandro Tagliapietra > >> > > >> > On Mon, Apr 15, 2019 at 10:08 PM Bruno Cadonna <br...@confluent.io> > >> wrote: > >> > > >> > > Hi Alessandro, > >> > > > >> > > Have a look at this Kafka Usage Pattern for computing averages > without > >> > > using an ArrayList. > >> > > > >> > > > >> > > > >> > > >> > https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Stream+Usage+Patterns#KafkaStreamUsagePatterns-Howtocomputean(windowed)average > >> > > ? > >> > > > >> > > The advantages of this pattern over the ArrayList approach is the > >> reduced > >> > > space needed to compute the aggregate. Note that you will still need > >> to > >> > > implement a SerDe. However, the SerDe should be a bit easier to > >> implement > >> > > than a SerDe for an ArrayList. > >> > > > >> > > Hope that helps. > >> > > > >> > > Best, > >> > > Bruno > >> > > > >> > > On Mon, Apr 15, 2019 at 4:57 PM Alessandro Tagliapietra < > >> > > tagliapietra.alessan...@gmail.com> wrote: > >> > > > >> > > > Sorry but it seemed harder than I thought, > >> > > > > >> > > > to have the custom aggregation working I need to get an ArrayList > of > >> > all > >> > > > the values in the window, so far my aggregate DSL method creates > an > >> > > > ArrayList on the initializer and adds each value to the list in > the > >> > > > aggregator. > >> > > > Then I think I'll have to provide a serder to change the output > >> type of > >> > > > that method. > >> > > > I was looking at > >> > > > > >> > > > > >> > > > >> > > >> > https://stackoverflow.com/questions/46365884/issue-with-arraylist-serde-in-kafka-streams-api > >> > > > but > >> > > > that seems more towards a list of longs and already uses > longSerde. > >> > > > I'm currently trying to implement another avro model that has a > >> field > >> > of > >> > > > type array so I can use the regular avro serializer to implement > >> this. > >> > > > Should I create my own serdes instead or is this the right way? > >> > > > > >> > > > Thank you in advance > >> > > > > >> > > > -- > >> > > > Alessandro Tagliapietra > >> > > > > >> > > > On Mon, Apr 15, 2019 at 3:42 PM Alessandro Tagliapietra < > >> > > > tagliapietra.alessan...@gmail.com> wrote: > >> > > > > >> > > > > Thank you Bruno and Matthias, > >> > > > > > >> > > > > I've modified the transformer to implement the > >> > ValueTransformerWithKey > >> > > > > interface and everything is working fine. > >> > > > > I've now to window the data and manually aggregate each window > >> data > >> > > since > >> > > > > I've to do some averages and sum of differences. > >> > > > > So far I've just having some issues with message types since I'm > >> > > changing > >> > > > > the data type when aggregating the window but I think it's an > easy > >> > > > problem. > >> > > > > > >> > > > > Thank you again > >> > > > > Best > >> > > > > > >> > > > > -- > >> > > > > Alessandro Tagliapietra > >> > > > > > >> > > > > On Sun, Apr 14, 2019 at 11:26 AM Bruno Cadonna < > >> br...@confluent.io> > >> > > > wrote: > >> > > > > > >> > > > >> Hi Alessandro, > >> > > > >> > >> > > > >> the `TransformSupplier` is internally wrapped with a > >> > > > `ProcessorSupplier`, > >> > > > >> so the statement > >> > > > >> > >> > > > >> `transform` is essentially equivalent to adding the Transformer > >> via > >> > > > >> Topology#addProcessor() to your processor topology > >> > > > >> > >> > > > >> is correct. > >> > > > >> > >> > > > >> If you do not change the key, you should definitely use one of > >> the > >> > > > >> overloads of `transformValues` to avoid internal data > >> > redistribution. > >> > > In > >> > > > >> your case the overload with `ValueTransformerWithKeySupplier` > as > >> > > > suggested > >> > > > >> by Matthias would fit. > >> > > > >> > >> > > > >> Best, > >> > > > >> Bruno > >> > > > >> > >> > > > >> On Sat, Apr 13, 2019 at 12:51 PM Matthias J. Sax < > >> > > matth...@confluent.io > >> > > > > > >> > > > >> wrote: > >> > > > >> > >> > > > >> > There is also `ValueTransformerWithKey` that gives you > >> read-only > >> > > acess > >> > > > >> > to the key. > >> > > > >> > > >> > > > >> > -Matthias > >> > > > >> > > >> > > > >> > On 4/12/19 5:34 PM, Alessandro Tagliapietra wrote: > >> > > > >> > > Hi Bruno, > >> > > > >> > > > >> > > > >> > > Thank you for the quick answer. > >> > > > >> > > > >> > > > >> > > I'm actually trying to do that since it seems there is > >> really no > >> > > way > >> > > > >> to > >> > > > >> > > have it use `Processor<K, V>`. > >> > > > >> > > I just wanted (if that would've made any sense) to use the > >> > > Processor > >> > > > >> in > >> > > > >> > > both DSL and non-DSL pipelines. > >> > > > >> > > > >> > > > >> > > Anyway, regarding `transformValues()` I don't think I can > >> use it > >> > > as > >> > > > I > >> > > > >> > need > >> > > > >> > > the message key since that is the discriminating value for > >> the > >> > > > filter > >> > > > >> (I > >> > > > >> > > want to exclude old values per sensor ID so per message > key) > >> > > > >> > > > >> > > > >> > > Right now I've this > >> > > > >> > > > >> > > > >> > > >> > > > >> > >> > > > > >> > > > >> > > >> > https://gist.github.com/alex88/7d229698546971452c3efc862fb4d3fd#file-timestampfiltertransformer-java > >> > > > >> > > and > >> > > > >> > > i'm using it with `transform()` . > >> > > > >> > > > >> > > > >> > > One thing I've found confusing is this > >> > > > >> > > > >> > > > >> > > >> > > > >> > >> > > > > >> > > > >> > > >> > https://docs.confluent.io/current/streams/developer-guide/dsl-api.html#streams-developer-guide-dsl-process > >> > > > >> > > > >> > > > >> > > transform is essentially equivalent to adding the > Transformer > >> > via > >> > > > >> > >> Topology#addProcessor() to yourprocessor topology > >> > > > >> > >> < > >> > > > >> > > >> > > > >> > >> > > > > >> > > > >> > > >> > https://docs.confluent.io/current/streams/concepts.html#streams-concepts-processor-topology > >> > > > >> > > > >> > > > >> > >> . > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > is it? Doesn't `transform` need a TransformSupplier while > >> > > > >> `addProcessor` > >> > > > >> > > uses a ProcessorSupplier? > >> > > > >> > > > >> > > > >> > > Thank you again for your help > >> > > > >> > > > >> > > > >> > > -- > >> > > > >> > > Alessandro Tagliapietra > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > On Fri, Apr 12, 2019 at 5:04 PM Bruno Cadonna < > >> > br...@confluent.io > >> > > > > >> > > > >> > wrote: > >> > > > >> > > > >> > > > >> > >> Hi Alessandro, > >> > > > >> > >> > >> > > > >> > >> Have you considered using `transform()` (actually in your > >> case > >> > > you > >> > > > >> > should > >> > > > >> > >> use `transformValues()`) instead of `.process()`? > >> `transform()` > >> > > and > >> > > > >> > >> `transformValues()` are stateful operations similar to > >> > `.process` > >> > > > but > >> > > > >> > they > >> > > > >> > >> return a `KStream`. On a `KStream` you can then apply a > >> > windowed > >> > > > >> > >> aggregation. > >> > > > >> > >> > >> > > > >> > >> Hope that helps. > >> > > > >> > >> > >> > > > >> > >> Best, > >> > > > >> > >> Bruno > >> > > > >> > >> > >> > > > >> > >> > >> > > > >> > >> On Fri, Apr 12, 2019 at 4:31 PM Alessandro Tagliapietra < > >> > > > >> > >> tagliapietra.alessan...@gmail.com> wrote: > >> > > > >> > >> > >> > > > >> > >>> Hi there, > >> > > > >> > >>> > >> > > > >> > >>> I'm just starting with Kafka and I'm trying to create a > >> stream > >> > > > >> > processor > >> > > > >> > >>> that in multiple stages: > >> > > > >> > >>> - filters messages using a kv store so that only > messages > >> > with > >> > > > >> higher > >> > > > >> > >>> timestamp gets processed > >> > > > >> > >>> - aggregates the message metrics by minute giving e.g. > the > >> > avg > >> > > of > >> > > > >> > those > >> > > > >> > >>> metrics in that minute > >> > > > >> > >>> > >> > > > >> > >>> The message is simple, the key is the sensor ID and the > >> value > >> > is > >> > > > >> e.g. { > >> > > > >> > >>> timestamp: UNIX_TIMESTAMP, speed: INT }. > >> > > > >> > >>> > >> > > > >> > >>> I've started by creating a processor to use the kv store > >> and > >> > > > filter > >> > > > >> old > >> > > > >> > >>> messages: > >> > > > >> > >>> > >> > > > >> > >>> > >> > > > >> > >>> > >> > > > >> > >> > >> > > > >> > > >> > > > >> > >> > > > > >> > > > >> > > >> > https://gist.github.com/alex88/7d229698546971452c3efc862fb4d3fd#file-timestampfilter-java > >> > > > >> > >>> > >> > > > >> > >>> Then I was trying to implement windowing, I saw very nice > >> > > > windowing > >> > > > >> > >>> examples for the DSL but none for the Processor API > (only a > >> > > small > >> > > > >> > >> reference > >> > > > >> > >>> to the windowed store), can someone point me in the right > >> > > > direction? > >> > > > >> > >>> > >> > > > >> > >>> Now, since I wasn't able to find any example I tried to > use > >> > the > >> > > > DSL > >> > > > >> but > >> > > > >> > >>> haven't found a way to use my processor with it, I saw > this > >> > > > >> > >>> > >> > > > >> > >>> > >> > > > >> > >> > >> > > > >> > > >> > > > >> > >> > > > > >> > > > >> > > >> > https://kafka.apache.org/21/documentation/streams/developer-guide/dsl-api.html#applying-processors-and-transformers-processor-api-integration > >> > > > >> > >>> but > >> > > > >> > >>> it explains mostly transformers not processors. I also > saw > >> > after > >> > > > >> that > >> > > > >> > the > >> > > > >> > >>> example usage of the processor but `.process(...)` > returns > >> > void, > >> > > > so > >> > > > >> I > >> > > > >> > >>> cannot have a KStream from a processor? > >> > > > >> > >>> > >> > > > >> > >>> Thank you all in advance > >> > > > >> > >>> > >> > > > >> > >>> -- > >> > > > >> > >>> Alessandro Tagliapietra > >> > > > >> > >>> > >> > > > >> > >> > >> > > > >> > > > >> > > > >> > > >> > > > >> > > >> > > > >> > >> > > > > > >> > > > > >> > > > >> > > >> > > >
