No problem, I just wanted to make sure people still had more to say. I will wait another week.
Thanks, Kyle On May 16, 2017 4:25 AM, "Eno Thereska" <eno.there...@gmail.com> wrote: > Hi Kyle, > > Sorry for the delay in reviews, tomorrow is feature freeze deadline ( > https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.11.0.0 < > https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.11.0.0>) > so people are busier than usual. Stay tuned. > > Eno > > On 15 May 2017, at 13:25, Kyle Winkelman <winkelman.k...@gmail.com> > wrote: > > > > Damian Guy, could you let me know if you plan to review this further? > There > > is no rush, but if you dont have any additional comments I could start > the > > voting and finish my WIP PR. > > > > Thanks, > > Kyle > > > > On May 9, 2017 11:07 AM, "Kyle Winkelman" <winkelman.k...@gmail.com> > wrote: > > > >> Eno, is there anyone else that is an expert in the kafka streams realm > >> that I should reach out to for input? > >> > >> I believe Damian Guy is still planning on reviewing this more in depth > so > >> I will wait for his inputs before continuing. > >> > >> On May 9, 2017 7:30 AM, "Eno Thereska" <eno.there...@gmail.com> wrote: > >> > >>> Thanks Kyle, good arguments. > >>> > >>> Eno > >>> > >>>> On May 7, 2017, at 5:06 PM, Kyle Winkelman <winkelman.k...@gmail.com> > >>> wrote: > >>>> > >>>> *- minor: could you add an exact example (similar to what Jay’s > example > >>> is, > >>>> or like your Spark/Pig pointers had) to make this super concrete?* > >>>> I have added a more concrete example to the KIP. > >>>> > >>>> *- my main concern is that we’re exposing this optimization to the > DSL. > >>> In > >>>> an ideal world, an optimizer would take the existing DSL and do the > >>> right > >>>> thing under the covers (create just one state store, arrange the nodes > >>>> etc). The original DSL had a bunch of small, composable pieces (group, > >>>> aggregate, join) that this proposal groups together. I’d like to hear > >>> your > >>>> thoughts on whether it’s possible to do this optimization with the > >>> current > >>>> DSL, at the topology builder level.* > >>>> You would have to make a lot of checks to understand if it is even > >>> possible > >>>> to make this optimization: > >>>> 1. Make sure they are all KTableKTableOuterJoins > >>>> 2. None of the intermediate KTables are used for anything else. > >>>> 3. None of the intermediate stores are used. (This may be impossible > >>>> especially if they use KafkaStreams#store after the topology has > already > >>>> been built.) > >>>> You would then need to make decisions during the optimization: > >>>> 1. Your new initializer would the composite of all the individual > >>>> initializers and the valueJoiners. > >>>> 2. I am having a hard time thinking about how you would turn the > >>>> aggregators and valueJoiners into an aggregator that would work on the > >>>> final object, but this may be possible. > >>>> 3. Which state store would you use? The ones declared would be for the > >>>> aggregate values. None of the declared ones would be guaranteed to > hold > >>> the > >>>> final object. This would mean you must created a new state store and > not > >>>> created any of the declared ones. > >>>> > >>>> The main argument I have against it is even if it could be done I > don't > >>>> know that we would want to have this be an optimization in the > >>> background > >>>> because the user would still be required to think about all of the > >>>> intermediate values that they shouldn't need to worry about if they > only > >>>> care about the final object. > >>>> > >>>> In my opinion cogroup is a common enough case that it should be part > of > >>> the > >>>> composable pieces (group, aggregate, join) because we want to allow > >>> people > >>>> to join more than 2 or more streams in an easy way. Right now I don't > >>> think > >>>> we give them ways of handling this use case easily. > >>>> > >>>> *-I think there will be scope for several such optimizations in the > >>> future > >>>> and perhaps at some point we need to think about decoupling the 1:1 > >>> mapping > >>>> from the DSL into the physical topology.* > >>>> I would argue that cogroup is not just an optimization it is a new way > >>> for > >>>> the users to look at accomplishing a problem that requires multiple > >>>> streams. I may sound like a broken record but I don't think users > should > >>>> have to build the N-1 intermediate tables and deal with their > >>> initializers, > >>>> serdes and stores if all they care about is the final object. > >>>> Now if for example someone uses cogroup but doesn't supply additional > >>>> streams and aggregators this case is equivalent to a single grouped > >>> stream > >>>> making an aggregate call. This case is what I view an optimization as, > >>> we > >>>> could remove the KStreamCogroup and act as if there was just a call to > >>>> KGroupedStream#aggregate instead of calling KGroupedStream#cogroup. (I > >>>> would prefer to just write a warning saying that this is not how > >>> cogroup is > >>>> to be used.) > >>>> > >>>> Thanks, > >>>> Kyle > >>>> > >>>> On Sun, May 7, 2017 at 5:41 AM, Eno Thereska <eno.there...@gmail.com> > >>> wrote: > >>>> > >>>>> Hi Kyle, > >>>>> > >>>>> Thanks for the KIP again. A couple of comments: > >>>>> > >>>>> - minor: could you add an exact example (similar to what Jay’s > example > >>> is, > >>>>> or like your Spark/Pig pointers had) to make this super concrete? > >>>>> > >>>>> - my main concern is that we’re exposing this optimization to the > DSL. > >>> In > >>>>> an ideal world, an optimizer would take the existing DSL and do the > >>> right > >>>>> thing under the covers (create just one state store, arrange the > nodes > >>>>> etc). The original DSL had a bunch of small, composable pieces > (group, > >>>>> aggregate, join) that this proposal groups together. I’d like to hear > >>> your > >>>>> thoughts on whether it’s possible to do this optimization with the > >>> current > >>>>> DSL, at the topology builder level. > >>>>> > >>>>> I think there will be scope for several such optimizations in the > >>> future > >>>>> and perhaps at some point we need to think about decoupling the 1:1 > >>> mapping > >>>>> from the DSL into the physical topology. > >>>>> > >>>>> Thanks > >>>>> Eno > >>>>> > >>>>>> On May 5, 2017, at 4:39 PM, Jay Kreps <j...@confluent.io> wrote: > >>>>>> > >>>>>> I haven't digested the proposal but the use case is pretty common. > An > >>>>>> example would be the "customer 360" or "unified customer profile" > use > >>>>> case > >>>>>> we often use. In that use case you have a dozen systems each of > which > >>> has > >>>>>> some information about your customer (account details, settings, > >>> billing > >>>>>> info, customer service contacts, purchase history, etc). Your goal > is > >>> to > >>>>>> join/munge these into a single profile record for each customer that > >>> has > >>>>>> all the relevant info in a usable form and is up-to-date with all > the > >>>>>> source systems. If you implement that with kstreams as a sequence of > >>>>> joins > >>>>>> i think today we'd fully materialize N-1 intermediate tables. But > >>> clearly > >>>>>> you only need a single stage to group all these things that are > >>> already > >>>>>> co-partitioned. A distributed database would do this under the > covers > >>>>> which > >>>>>> is arguably better (at least when it does the right thing) and > >>> perhaps we > >>>>>> could do the same thing but I'm not sure we know the partitioning so > >>> we > >>>>> may > >>>>>> need an explicit cogroup command that impllies they are already > >>>>>> co-partitioned. > >>>>>> > >>>>>> -Jay > >>>>>> > >>>>>> On Fri, May 5, 2017 at 5:56 AM, Kyle Winkelman < > >>> winkelman.k...@gmail.com > >>>>>> > >>>>>> wrote: > >>>>>> > >>>>>>> Yea thats a good way to look at it. > >>>>>>> I have seen this type of functionality in a couple other platforms > >>> like > >>>>>>> spark and pig. > >>>>>>> https://spark.apache.org/docs/0.6.2/api/core/spark/ > >>>>> PairRDDFunctions.html > >>>>>>> https://www.tutorialspoint.com/apache_pig/apache_pig_ > >>>>> cogroup_operator.htm > >>>>>>> > >>>>>>> > >>>>>>> On May 5, 2017 7:43 AM, "Damian Guy" <damian....@gmail.com> wrote: > >>>>>>> > >>>>>>>> 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 > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>> > >>>>> > >>> > >>> > >
