Hi Kyle,
If i'm reading this correctly it is like an N way outer join? So an input
on any stream will always produce a new aggregated value - is that correct?
Effectively, each Aggregator just looks up the current value, aggregates
and forwards the result.
I need to look into it and think about it a bit more, but it seems like it
could be a useful optimization.
On Thu, 4 May 2017 at 23:21 Kyle Winkelman <winkelman.k...@gmail.com> wrote:
> I sure can. I have added the following description to my KIP. If this
> doesn't help let me know and I will take some more time to build a diagram
> and make more of a step by step description:
>
> Example with Current API:
>
> KTable<K, V1> table1 =
> builder.stream("topic1").groupByKey().aggregate(initializer1, aggregator1,
> aggValueSerde1, storeName1);
> KTable<K, V2> table2 =
> builder.stream("topic2").groupByKey().aggregate(initializer2, aggregator2,
> aggValueSerde2, storeName2);
> KTable<K, V3> table3 =
> builder.stream("topic3").groupByKey().aggregate(initializer3, aggregator3,
> aggValueSerde3, storeName3);
> KTable<K, CG> cogrouped = table1.outerJoin(table2,
> joinerOneAndTwo).outerJoin(table3, joinerOneTwoAndThree);
>
> As you can see this creates 3 StateStores, requires 3 initializers, and 3
> aggValueSerdes. This also adds the pressure to user to define what the
> intermediate values are going to be (V1, V2, V3). They are left with a
> couple choices, first to make V1, V2, and V3 all the same as CG and the two
> joiners are more like mergers, or second make them intermediate states such
> as Topic1Map, Topic2Map, and Topic3Map and the joiners use those to build
> the final aggregate CG value. This is something the user could avoid
> thinking about with this KIP.
>
> When a new input arrives lets say at "topic1" it will first go through a
> KStreamAggregate grabbing the current aggregate from storeName1. It will
> produce this in the form of the first intermediate value and get sent
> through a KTableKTableOuterJoin where it will look up the current value of
> the key in storeName2. It will use the first joiner to calculate the second
> intermediate value, which will go through an additional
> KTableKTableOuterJoin. Here it will look up the current value of the key in
> storeName3 and use the second joiner to build the final aggregate value.
>
> If you think through all possibilities for incoming topics you will see
> that no matter which topic it comes in through all three stores are queried
> and all of the joiners must get used.
>
> Topology wise for N incoming streams this creates N
> KStreamAggregates, 2*(N-1) KTableKTableOuterJoins, and N-1
> KTableKTableJoinMergers.
>
>
>
> Example with Proposed API:
>
> KGroupedStream<K, V1> grouped1 = builder.stream("topic1").groupByKey();
> KGroupedStream<K, V2> grouped2 = builder.stream("topic2").groupByKey();
> KGroupedStream<K, V3> grouped3 = builder.stream("topic3").groupByKey();
> KTable<K, CG> cogrouped = grouped1.cogroup(initializer1, aggregator1,
> aggValueSerde1, storeName1)
> .cogroup(grouped2, aggregator2)
> .cogroup(grouped3, aggregator3)
> .aggregate();
>
> As you can see this creates 1 StateStore, requires 1 initializer, and 1
> aggValueSerde. The user no longer has to worry about the intermediate
> values and the joiners. All they have to think about is how each stream
> impacts the creation of the final CG object.
>
> When a new input arrives lets say at "topic1" it will first go through a
> KStreamAggreagte and grab the current aggregate from storeName1. It will
> add its incoming object to the aggregate, update the store and pass the new
> aggregate on. This new aggregate goes through the KStreamCogroup which is
> pretty much just a pass through processor and you are done.
>
> Topology wise for N incoming streams the new api will only every create N
> KStreamAggregates and 1 KStreamCogroup.
>
> On Thu, May 4, 2017 at 4:42 PM, Matthias J. Sax <matth...@confluent.io>
> wrote:
>
> > Kyle,
> >
> > thanks a lot for the KIP. Maybe I am a little slow, but I could not
> > follow completely. Could you maybe add a more concrete example, like 3
> > streams with 3 records each (plus expected result), and show the
> > difference between current way to to implement it and the proposed API?
> > This could also cover the internal processing to see what store calls
> > would be required for both approaches etc.
> >
> > I think, it's pretty advanced stuff you propose, and it would help to
> > understand it better.
> >
> > Thanks a lot!
> >
> >
> > -Matthias
> >
> >
> >
> > On 5/4/17 11:39 AM, Kyle Winkelman wrote:
> > > I have made a pull request. It can be found here.
> > >
> > > https://github.com/apache/kafka/pull/2975
> > >
> > > I plan to write some more unit tests for my classes and get around to
> > > writing documentation for the public api additions.
> > >
> > > One thing I was curious about is during the
> > KCogroupedStreamImpl#aggregate
> > > method I pass null to the KGroupedStream#repartitionIfRequired method.
> I
> > > can't supply the store name because if more than one grouped stream
> > > repartitions an error is thrown. Is there some name that someone can
> > > recommend or should I leave the null and allow it to fall back to the
> > > KGroupedStream.name?
> > >
> > > Should this be expanded to handle grouped tables? This would be pretty
> > easy
> > > for a normal aggregate but one allowing session stores and windowed
> > stores
> > > would required KTableSessionWindowAggregate and KTableWindowAggregate
> > > implementations.
> > >
> > > Thanks,
> > > Kyle
> > >
> > > On May 4, 2017 1:24 PM, "Eno Thereska" <eno.there...@gmail.com> wrote:
> > >
> > >> I’ll look as well asap, sorry, been swamped.
> > >>
> > >> Eno
> > >>> On May 4, 2017, at 6:17 PM, Damian Guy <damian....@gmail.com> wrote:
> > >>>
> > >>> Hi Kyle,
> > >>>
> > >>> Thanks for the KIP. I apologize that i haven't had the chance to look
> > at
> > >>> the KIP yet, but will schedule some time to look into it tomorrow.
> For
> > >> the
> > >>> implementation, can you raise a PR against kafka trunk and mark it as
> > >> WIP?
> > >>> It will be easier to review what you have done.
> > >>>
> > >>> Thanks,
> > >>> Damian
> > >>>
> > >>> On Thu, 4 May 2017 at 11:50 Kyle Winkelman <winkelman.k...@gmail.com
> >
> > >> wrote:
> > >>>
> > >>>> I am replying to this in hopes it will draw some attention to my KIP
> > as
> > >> I
> > >>>> haven't heard from anyone in a couple days. This is my first KIP and
> > my
> > >>>> first large contribution to the project so I'm sure I did something
> > >> wrong.
> > >>>> ;)
> > >>>>
> > >>>> On May 1, 2017 4:18 PM, "Kyle Winkelman" <winkelman.k...@gmail.com>
> > >> wrote:
> > >>>>
> > >>>>> Hello all,
> > >>>>>
> > >>>>> I have created KIP-150 to facilitate discussion about adding
> cogroup
> > to
> > >>>>> the streams DSL.
> > >>>>>
> > >>>>> Please find the KIP here:
> > >>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > >>>>> 150+-+Kafka-Streams+Cogroup
> > >>>>>
> > >>>>> Please find my initial implementation here:
> > >>>>> https://github.com/KyleWinkelman/kafka
> > >>>>>
> > >>>>> Thanks,
> > >>>>> Kyle Winkelman
> > >>>>>
> > >>>>
> > >>
> > >>
> > >
> >
> >
>
