Hi Mathieu, If you are only interested in the aggregate result "snapshot" but not its change stream (note that KTable itself is not actually a "table" as in RDBMS, but still a stream), you can try to use the queryable state feature that is available in trunk, which will be available in 0.10.1.0 release.
In sum, it allows you to query any states "snapshot" which is used in aggregation operators in real time with state store provided APIs such as get-by-key, range queries on windows, etc. Details can be found in thie KIP (we are working on more docs / blog posts at the time): https://cwiki.apache.org/confluence/display/KAFKA/KIP-67%3A+Queryable+state+for+Kafka+Streams Guozhang On Thu, Aug 18, 2016 at 6:40 AM, Mathieu Fenniak < mathieu.fenn...@replicon.com> wrote: > Hi Guozhang, > > Hm... I hadn't thought of the repartitioning involvement. > > I'm not confident I'm understanding completely, but I believe you're > saying the decision to process data in this way is made before the > data being processed is available, because the partition *may* change, > because the groupBy key *may* change. > > I'm still feeling that I'm stuck getting corrupted output in the > middle of an aggregation. > > It's especially problematic for me if the updates to the source KTable > don't actually affect the results of the aggregation. In the > word-count example in my original e-mail, this might be similar to > editing an unrelated field "author" in any article; doesn't actually > affect the groupBy, doesn't affect the aggregation, but still results > in the wrong output occurring temporarily. (and inefficient > processing) > > Are there any tools in Kafka Streams that might help me prevent > downstream calculations if the relevant inputs haven't changed? I was > thinking I'd be able to use mapValues to pluck only relevant fields > out of a KTable, materialize a new KTable (.through) from that, and > then there'd be some state from which KS would be able to only invoke > downstream nodes if data has changed... but it doesn't seem to work > like that. > > Thanks so much for your responses Guozhang, I really appreciate your > time to help me out. > > Mathieu > > > On Wed, Aug 17, 2016 at 5:51 PM, Guozhang Wang <wangg...@gmail.com> wrote: > > The problem is that Kafka Streams need to repartition the streams based > on > > the groupBy keys when doing aggregations. For your case, the original > > stream may be read from a topic that is partitioned on "K", and you need > to > > first repartition on "category" on an intermediate topic before the > > aggregation can be executed. > > > > Hence the old and new value may be sent to two different partitions of > the > > intermediate topic, and hence be processed by two different process (it > > won't be the case in your application, since you mentioned the "category" > > will never change). Since the library cannot tell if the groupBy key will > > never change, it has to be conservative and do this subtract / add > process > > while receiving the old / new value. > > > > > > Guozhang > > > > > > On Wed, Aug 17, 2016 at 1:45 PM, Mathieu Fenniak < > > mathieu.fenn...@replicon.com> wrote: > > > >> Hi Guozhang, > >> > >> Thanks for responding. Ah, I see what you're saying... in the case of > >> an update to the KTable, the aggregator's subtractor result would be > >> necessary if the group-by key changes in the update. > >> > >> It makes sense, but unfortunately the behavior leaves me feeling a > >> little sketchy... when the group-by key doesn't change (which is > >> guaranteed in my case), I'm outputting results that don't correspond > >> at all to the inputs, temporarily. It's immediately followed by a > >> corrected result. > >> > >> Would it be a feasible optimization to not send the subtractor's > >> result out of the aggregate, only in the case where the groupBy key > >> does not change between the old record and the new record? > >> > >> Mathieu > >> > >> On Wed, Aug 17, 2016 at 2:12 PM, Guozhang Wang <wangg...@gmail.com> > wrote: > >> > Hello Mathieu, > >> > > >> > Note that semantics of KTable aggregations (i.e. > >> "KTable.groupBy.aggregate" > >> > as in 0.10.0) and KStream aggregations (i.e. "KStream.aggregateByKey" > as > >> in > >> > 0.10.0) are different, in the sense that when the table is updated > (i.e. > >> a > >> > new record with the same key "K1" is received), the old record's > effect > >> on > >> > the aggregation need to first be subtracted before the new record's > >> effect > >> > on the aggregation can be added; whereas in the latter case there is > no > >> > "old values" that are not overridden, hence only "adder" aggregator is > >> > needed. > >> > > >> > So suppose your updated record on K1 is on a different "category", > say: > >> > > >> > K1, {"category": "kafka2", "text": "word1, word2, word3, word4"} > >> > > >> > > >> > Then the aggregated result should be: > >> > > >> > {key: "kafka", value: 2} > >> > {key: "kafka2", value: 4} > >> > > >> > > >> > Does this make sense now? > >> > > >> > Guozhang > >> > > >> > > >> > On Wed, Aug 17, 2016 at 7:59 AM, Mathieu Fenniak < > >> > mathieu.fenn...@replicon.com> wrote: > >> > > >> >> Hello again, kafka-users, > >> >> > >> >> When I aggregate a KTable, a future input that updates a KTable's > >> >> value for a specific key causes the aggregate's subtractor to be > >> >> invoked, and then its adder. This part is great, completely > >> >> as-expected. > >> >> > >> >> But what I didn't expect is that the intermediate result of the > >> >> subtractor would be sent downstream. This value doesn't reflect the > >> >> reality of the inputs to the aggregator, so sending it downstream is > >> >> effectively sending "corrupt" data to the next processing node. Is > >> >> this the expected behavior, or is this a bug? > >> >> > >> >> Take for example, a table of blog articles and an aggregator that > >> >> counts the number of words in each category of the blog: > >> >> > >> >> topic: articles > >> >> K1, {"category": "kafka", "text": "word1, word2, word3"} > >> >> K2, {"category": "kafka", "text": "word1, word2"} > >> >> > >> >> articles.groupBy((k,v) -> v.category) > >> >> .aggregate(() -> 0, > >> >> (k,v,t) -> t + v.text.split(" ").length, > >> >> (k,v,t) -> t - v.text.split(" ").length > >> >> ) > >> >> > >> >> This aggregator will produce {key: "kafka", value: 3}, then {key: > >> >> "kafka", value: 5}. If I update one of the blog articles and send a > >> >> new message to the articles topic: > >> >> > >> >> K1, {"category": "kafka", "text": "word1, word2, word3, word4"} > >> >> > >> >> The aggregator will first produce {key: "kafka", value: 2} when the > >> >> subtractor is called, then will produce {key: "kafka", value: 6} when > >> >> the adder is called. The subtractor's calculation does not actually > >> >> match the reality; K1 was never deleted, it was just updated. > >> >> > >> >> Mathieu > >> >> > >> > > >> > > >> > > >> > -- > >> > -- Guozhang > >> > > > > > > > > -- > > -- Guozhang > -- -- Guozhang
