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
