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 >>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>> >>>>> >>> >>>
