This suggestion lgtm. I would vote for the first alternative than adding it to the `KStreamBuilder` though.
On Thu, Jun 8, 2017 at 10:58 AM, Xavier Léauté <xav...@confluent.io> wrote: > I have a minor suggestion to make the API a little bit more symmetric. > I feel it would make more sense to move the initializer and serde to the > final aggregate statement, since the serde only applies to the state store, > and the initializer doesn't bear any relation to the first group in > particular. It would end up looking like this: > > KTable<K, CG> cogrouped = > grouped1.cogroup(aggregator1) > .cogroup(grouped2, aggregator2) > .cogroup(grouped3, aggregator3) > .aggregate(initializer1, aggValueSerde, storeName1); > > Alternatively, we could move the the first cogroup() method to > KStreamBuilder, similar to how we have .merge() > and end up with an api that would be even more symmetric. > > KStreamBuilder.cogroup(grouped1, aggregator1) > .cogroup(grouped2, aggregator2) > .cogroup(grouped3, aggregator3) > .aggregate(initializer1, aggValueSerde, storeName1); > > This doesn't have to be a blocker, but I thought it would make the API just > a tad cleaner. > > On Tue, Jun 6, 2017 at 3:59 PM Guozhang Wang <wangg...@gmail.com> wrote: > > > Kyle, > > > > Thanks a lot for the updated KIP. It looks good to me. > > > > > > Guozhang > > > > > > On Fri, Jun 2, 2017 at 5:37 AM, Jim Jagielski <j...@jagunet.com> wrote: > > > > > This makes much more sense to me. +1 > > > > > > > On Jun 1, 2017, at 10:33 AM, Kyle Winkelman < > winkelman.k...@gmail.com> > > > wrote: > > > > > > > > I have updated the KIP and my PR. Let me know what you think. > > > > To created a cogrouped stream just call cogroup on a KgroupedStream > and > > > > supply the initializer, aggValueSerde, and an aggregator. Then > continue > > > > adding kgroupedstreams and aggregators. Then call one of the many > > > aggregate > > > > calls to create a KTable. > > > > > > > > Thanks, > > > > Kyle > > > > > > > > On Jun 1, 2017 4:03 AM, "Damian Guy" <damian....@gmail.com> wrote: > > > > > > > >> Hi Kyle, > > > >> > > > >> Thanks for the update. I think just one initializer makes sense as > it > > > >> should only be called once per key and generally it is just going to > > > create > > > >> a new instance of whatever the Aggregate class is. > > > >> > > > >> Cheers, > > > >> Damian > > > >> > > > >> On Wed, 31 May 2017 at 20:09 Kyle Winkelman < > winkelman.k...@gmail.com > > > > > > >> wrote: > > > >> > > > >>> Hello all, > > > >>> > > > >>> I have spent some more time on this and the best alternative I have > > > come > > > >> up > > > >>> with is: > > > >>> KGroupedStream has a single cogroup call that takes an initializer > > and > > > an > > > >>> aggregator. > > > >>> CogroupedKStream has a cogroup call that takes additional > > groupedStream > > > >>> aggregator pairs. > > > >>> CogroupedKStream has multiple aggregate methods that create the > > > different > > > >>> stores. > > > >>> > > > >>> I plan on updating the kip but I want people's input on if we > should > > > have > > > >>> the initializer be passed in once at the beginning or if we should > > > >> instead > > > >>> have the initializer be required for each call to one of the > > aggregate > > > >>> calls. The first makes more sense to me but doesnt allow the user > to > > > >>> specify different initializers for different tables. > > > >>> > > > >>> Thanks, > > > >>> Kyle > > > >>> > > > >>> On May 24, 2017 7:46 PM, "Kyle Winkelman" < > winkelman.k...@gmail.com> > > > >>> wrote: > > > >>> > > > >>>> Yea I really like that idea I'll see what I can do to update the > kip > > > >> and > > > >>>> my pr when I have some time. I'm not sure how well creating the > > > >>>> kstreamaggregates will go though because at that point I will have > > > >> thrown > > > >>>> away the type of the values. It will be type safe I just may need > to > > > >> do a > > > >>>> little forcing. > > > >>>> > > > >>>> Thanks, > > > >>>> Kyle > > > >>>> > > > >>>> On May 24, 2017 3:28 PM, "Guozhang Wang" <wangg...@gmail.com> > > wrote: > > > >>>> > > > >>>>> Kyle, > > > >>>>> > > > >>>>> Thanks for the explanations, my previous read on the wiki > examples > > > was > > > >>>>> wrong. > > > >>>>> > > > >>>>> So I guess my motivation should be "reduced" to: can we move the > > > >> window > > > >>>>> specs param from "KGroupedStream#cogroup(..)" to > > > >>>>> "CogroupedKStream#aggregate(..)", and my motivations are: > > > >>>>> > > > >>>>> 1. minor: we can reduce the #.generics in CogroupedKStream from 3 > > to > > > >> 2. > > > >>>>> 2. major: this is for extensibility of the APIs, and since we are > > > >>> removing > > > >>>>> the "Evolving" annotations on Streams it may be harder to change > it > > > >>> again > > > >>>>> in the future. The extended use cases are that people wanted to > > have > > > >>>>> windowed running aggregates on different granularities, e.g. > "give > > me > > > >>> the > > > >>>>> counts per-minute, per-hour, per-day and per-week", and today in > > DSL > > > >> we > > > >>>>> need to specify that case in multiple aggregate operators, which > > gets > > > >> a > > > >>>>> state store / changelog, etc. And it is possible to optimize it > as > > > >> well > > > >>> to > > > >>>>> a single state store. Its implementation would be tricky as you > > need > > > >> to > > > >>>>> contain different lengthed windows within your window store but > > just > > > >>> from > > > >>>>> the public API point of view, it could be specified as: > > > >>>>> > > > >>>>> CogroupedKStream stream = stream1.cogroup(stream2, ... > > > >>>>> "state-store-name"); > > > >>>>> > > > >>>>> table1 = stream.aggregate(/*per-minute window*/) > > > >>>>> table2 = stream.aggregate(/*per-hour window*/) > > > >>>>> table3 = stream.aggregate(/*per-day window*/) > > > >>>>> > > > >>>>> while underlying we are only using a single store > > "state-store-name" > > > >> for > > > >>>>> it. > > > >>>>> > > > >>>>> > > > >>>>> Although this feature is out of the scope of this KIP, I'd like > to > > > >>> discuss > > > >>>>> if we can "leave the door open" to make such changes without > > > modifying > > > >>> the > > > >>>>> public APIs . > > > >>>>> > > > >>>>> Guozhang > > > >>>>> > > > >>>>> > > > >>>>> On Wed, May 24, 2017 at 3:57 AM, Kyle Winkelman < > > > >>> winkelman.k...@gmail.com > > > >>>>>> > > > >>>>> wrote: > > > >>>>> > > > >>>>>> I allow defining a single window/sessionwindow one time when you > > > >> make > > > >>>>> the > > > >>>>>> cogroup call from a KGroupedStream. From then on you are using > the > > > >>>>> cogroup > > > >>>>>> call from with in CogroupedKStream which doesnt accept any > > > >> additional > > > >>>>>> windows/sessionwindows. > > > >>>>>> > > > >>>>>> Is this what you meant by your question or did I misunderstand? > > > >>>>>> > > > >>>>>> On May 23, 2017 9:33 PM, "Guozhang Wang" <wangg...@gmail.com> > > > >> wrote: > > > >>>>>> > > > >>>>>> Another question that came to me is on "window alignment": from > > the > > > >>> KIP > > > >>>>> it > > > >>>>>> seems you are allowing users to specify a (potentially > different) > > > >>> window > > > >>>>>> spec in each co-grouped input stream. So if these window specs > are > > > >>>>>> different how should we "align" them with different input > > streams? I > > > >>>>> think > > > >>>>>> it is more natural to only specify on window spec in the > > > >>>>>> > > > >>>>>> KTable<RK, V> CogroupedKStream#aggregate(Windows); > > > >>>>>> > > > >>>>>> > > > >>>>>> And remove it from the cogroup() functions. WDYT? > > > >>>>>> > > > >>>>>> > > > >>>>>> Guozhang > > > >>>>>> > > > >>>>>> On Tue, May 23, 2017 at 6:22 PM, Guozhang Wang < > > wangg...@gmail.com> > > > >>>>> wrote: > > > >>>>>> > > > >>>>>>> Thanks for the proposal Kyle, this is a quite common use case > to > > > >>>>> support > > > >>>>>>> such multi-way table join (i.e. N source tables with N > aggregate > > > >>> func) > > > >>>>>> with > > > >>>>>>> a single store and N+1 serdes, I have seen lots of people using > > > >> the > > > >>>>>>> low-level PAPI to achieve this goal. > > > >>>>>>> > > > >>>>>>> > > > >>>>>>> On Fri, May 19, 2017 at 10:04 AM, Kyle Winkelman < > > > >>>>>> winkelman.k...@gmail.com > > > >>>>>>>> wrote: > > > >>>>>>> > > > >>>>>>>> I like your point about not handling other cases such as count > > > >> and > > > >>>>>> reduce. > > > >>>>>>>> > > > >>>>>>>> I think that reduce may not make sense because reduce assumes > > > >> that > > > >>>>> the > > > >>>>>>>> input values are the same as the output values. With cogroup > > > >> there > > > >>>>> may > > > >>>>>> be > > > >>>>>>>> multiple different input types and then your output type cant > be > > > >>>>>> multiple > > > >>>>>>>> different things. In the case where you have all matching > value > > > >>> types > > > >>>>>> you > > > >>>>>>>> can do KStreamBuilder#merge followed by the reduce. > > > >>>>>>>> > > > >>>>>>>> As for count I think it is possible to call count on all the > > > >>>>> individual > > > >>>>>>>> grouped streams and then do joins. Otherwise we could maybe > make > > > >> a > > > >>>>>> special > > > >>>>>>>> call in groupedstream for this case. Because in this case we > > dont > > > >>>>> need > > > >>>>>> to > > > >>>>>>>> do type checking on the values. It could be similar to the > > > >> current > > > >>>>> count > > > >>>>>>>> methods but accept a var args of additonal grouped streams as > > > >> well > > > >>>>> and > > > >>>>>>>> make > > > >>>>>>>> sure they have a key type of K. > > > >>>>>>>> > > > >>>>>>>> The way I have put the kip together is to ensure that we do > type > > > >>>>>> checking. > > > >>>>>>>> I don't see a way we could group them all first and then make > a > > > >>> call > > > >>>>> to > > > >>>>>>>> count, reduce, or aggregate because with aggregate they would > > > >> need > > > >>> to > > > >>>>>> pass > > > >>>>>>>> a list of aggregators and we would have no way of type > checking > > > >>> that > > > >>>>>> they > > > >>>>>>>> match the grouped streams. > > > >>>>>>>> > > > >>>>>>>> Thanks, > > > >>>>>>>> Kyle > > > >>>>>>>> > > > >>>>>>>> On May 19, 2017 11:42 AM, "Xavier Léauté" < > xav...@confluent.io> > > > >>>>> wrote: > > > >>>>>>>> > > > >>>>>>>>> Sorry to jump on this thread so late. I agree this is a very > > > >>> useful > > > >>>>>>>>> addition and wanted to provide an additional use-case and > some > > > >>> more > > > >>>>>>>>> comments. > > > >>>>>>>>> > > > >>>>>>>>> This is actually a very common analytics use-case in the > > > >> ad-tech > > > >>>>>>>> industry. > > > >>>>>>>>> The typical setup will have an auction stream, an impression > > > >>>>> stream, > > > >>>>>>>> and a > > > >>>>>>>>> click stream. Those three streams need to be combined to > > > >> compute > > > >>>>>>>> aggregate > > > >>>>>>>>> statistics (e.g. impression statistics, and click-through > > > >> rates), > > > >>>>>> since > > > >>>>>>>>> most of the attributes of interest are only present the > auction > > > >>>>>> stream. > > > >>>>>>>>> > > > >>>>>>>>> A simple way to do this is to co-group all the streams by the > > > >>>>> auction > > > >>>>>>>> key, > > > >>>>>>>>> and process updates to the co-group as events for each stream > > > >>> come > > > >>>>> in, > > > >>>>>>>>> keeping only one value from each stream before sending > > > >> downstream > > > >>>>> for > > > >>>>>>>>> further processing / aggregation. > > > >>>>>>>>> > > > >>>>>>>>> One could view the result of that co-group operation as a > > > >>> "KTable" > > > >>>>>> with > > > >>>>>>>>> multiple values per key. The key being the grouping key, and > > > >> the > > > >>>>>> values > > > >>>>>>>>> consisting of one value per stream. > > > >>>>>>>>> > > > >>>>>>>>> What I like about Kyle's approach is that allows elegant > > > >>>>> co-grouping > > > >>>>>> of > > > >>>>>>>>> multiple streams without having to worry about the number of > > > >>>>> streams, > > > >>>>>>>> and > > > >>>>>>>>> avoids dealing with Tuple types or other generic interfaces > > > >> that > > > >>>>> could > > > >>>>>>>> get > > > >>>>>>>>> messy if we wanted to preserve all the value types in the > > > >>> resulting > > > >>>>>>>>> co-grouped stream. > > > >>>>>>>>> > > > >>>>>>>>> My only concern is that we only allow the cogroup + aggregate > > > >>>>> combined > > > >>>>>>>>> operation. This forces the user to build their own tuple > > > >>>>> serialization > > > >>>>>>>>> format if they want to preserve the individual input stream > > > >>> values > > > >>>>> as > > > >>>>>> a > > > >>>>>>>>> group. It also deviates quite a bit from our approach in > > > >>>>>> KGroupedStream > > > >>>>>>>>> which offers other operations, such as count and reduce, > which > > > >>>>> should > > > >>>>>>>> also > > > >>>>>>>>> be applicable to a co-grouped stream. > > > >>>>>>>>> > > > >>>>>>>>> Overall I still think this is a really useful addition, but I > > > >>> feel > > > >>>>> we > > > >>>>>>>>> haven't spend much time trying to explore alternative DSLs > that > > > >>>>> could > > > >>>>>>>> maybe > > > >>>>>>>>> generalize better or match our existing syntax more closely. > > > >>>>>>>>> > > > >>>>>>>>> On Tue, May 9, 2017 at 8:08 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 > > > >>>>>>>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>>> > > > >>>>>>>> > > > >>>>>>> > > > >>>>>>> > > > >>>>>>> > > > >>>>>>> -- > > > >>>>>>> -- Guozhang > > > >>>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> -- > > > >>>>>> -- Guozhang > > > >>>>>> > > > >>>>> > > > >>>>> > > > >>>>> > > > >>>>> -- > > > >>>>> -- Guozhang > > > >>>>> > > > >>>> > > > >>> > > > >> > > > > > > > > > > > > -- > > -- Guozhang > > > -- -- Guozhang
