Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I just sent out a call for a vote. I think everyone has had a good discussion :). If there are any more thoughts I would love to hear them. Walker On Thu, Oct 31, 2019 at 10:36 AM Guozhang Wang wrote: > Hello Walker / Matthias, > > Thanks for your explanation. I can tell you've put a lot of

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hello Walker / Matthias, Thanks for your explanation. I can tell you've put a lot of thoughts into this already and it seems we cannot avoid adding new interfaces in any ways, so I will rest my arguments trying to reduce the number of first-class citizens in the Streams DSL :) Guozhang On

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Interesting discussion. My personal take is as follows: (1) Co-group is not a special case of a multi-way KTable join, because multiple record wit the same key from a single input stream should be aggregated together and there are not update semantics. A co-group is rather a muti-stream

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Gouzhang, I am not sure what you mean by "Fields from different streams are never aggregated together", this certainly can be the case but not the general rule. If we want to take care of the special cases where the key-sets are disjoint for each stream then they can be given no-op operators.

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, This is a good point about compatibility breakage while overloading the existing classes; while reading John and your exchanges, I think I still need to clarify the motivations a bit more: 1) Multiple streams need to be aggregated together, inputs are always *KStreams* and end result

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi John, Thank you for the background information. I think I understand your point. I believe that this could be fixed by making the motivation a little clearer in the KIP. I think that the motivation is when you have multiple streams that need to aggregate together to form a single object the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, Sorry for the delay in responding. Thanks for your response earlier. I think there might be a subtlety getting overlooked in considering whether we're talking about streams versus tables in cogroup. As I'm sure you know, Kafka Streams treats "stream" records as independent, immutable,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Gouzhang, Matthias and I did talk about overloading different a type of aggregate methods in the cogroup that would take in the windows and returns a windowed KTable. We decided that it would break too much with the current pattern that was established in the normal KStream. We can revisit

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Thank you Bill, I can get behind keeping the correct pattern and adding the Named object into the aggregate. And adding something like Cogrouped Object is a bit beyond the scope of this KIP. I updated the KIP to include the changes to include the Named parameter and some minor text fixes. If

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, On Fri, Oct 25, 2019 at 1:34 PM Walker Carlson wrote: > Hi Guozhang, > > 1. I am familiar with the cogroup of spark, it is very similar to > their join operator but instead it makes the values iterable. I think that > the use cases are different enough that it makes sense to specify

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, Thanks for taking on KIP-150, the co-group will be very useful. Regarding the naming, IMHO, we should stick to the current pattern, and that is, we provide overloads with a "Named" operator for the "aggregate" methods (I believe those are the only ones that create a processor).

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Guozhang, 1. I am familiar with the cogroup of spark, it is very similar to their join operator but instead it makes the values iterable. I think that the use cases are different enough that it makes sense to specify the aggregator when we do. I like the idea of "absorb" and I think it could

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, thanks for the KIP! I made a pass on the writeup and have some comments below: Meta: 1. Syntax-wise, I'm wondering if we have compared our current proposal with Spark's co-group syntax (I know they are targeting for different use cases, but wondering if their syntax is closer to the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Matthias, I am not familiar with KIP-307. After looking into it a bit I think I understand better what Named is for and it answers some of my concerns. I think that we should consider how many more methods it would make, there would be 12 aggregate functions including timeWindowed and

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Walker, I am not sure if I can follow your argument. What do you exactly mean by > I also >> think that in this case it would be better to separate the 2 option out >> into separate overloads. Maybe you can give an example what method signature you have in mind? >> We could take a named

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

While I like the idea Sophie I don't think that it is necessary. I also think that in this case it would be better to separate the 2 option out into separate overloads. We could take a named parameter from upstream or add an extra naming option however I don't really see the advantage that would

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

> I can personally not see any need to add other configuration Famous last words? Just kidding, 95% confidence is more than enough to me (and better to optimize for current design than for hypothetical future changes). One last question I have then is about the operator/store/repartition naming

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Interesting idea, Sophie. So far, we tried to reuse existing config objects and only add new ones when needed to avoid creating "redundant" classes. This is of course a reactive approach (with the drawback to deprecate stuff if we change it, as you described). I can personally not see any need

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Thanks for the explanation, makes sense to me! As for the API, one other thought I had is might we ever want or need to introduce any other configs or parameters in the future? Obviously that's difficult to say now (or maybe the answer seems obviously "no") but we seem to often end up needing to

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Walker, Thanks for picking up the KIP! I hope you don't mind if I pile on a question also. The motivation section depicts the current "non-optimal" case as a multi-way KTable equi-join, but the proposed API looks like a multi-way KStream windowed join+aggregate. My question is specifically

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Sophie, Thank you for your comments. As for the different methods signatures I have not really considered any other options but while I do agree it is confusing, I don't see any obvious solutions. The problem is that the cogroup essentially pairs a group stream with an aggregator and when it

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hey Walker, Thanks for the KIP! I have just a couple of questions: 1) It seems a little awkward to me that with the current API, we have a nearly identical "add stream to cogroup" method, except for the first which has a different signature (ie the first stream is joined as

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Good catch I updated that. I have made a PR for this KIP I then am splitting it into 3 parts, first cogroup for a key-value store ( here ), then for a timeWindowedStore, and then a sessionWindowedStore + ensuring partitioning. Walker On Tue, Oct 15,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Walker, thanks for picking up the KIP and reworking it for the changed API. Overall, the updated API suggestions make sense to me. The seem to align quite nicely with our current API design. One nit: In `CogroupedKStream#aggregate(...)` the type parameter of `Materialized` should be `V`, not

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+Kafka-Streams+Cogroup here is a link On Mon, Oct 14, 2019 at 2:52 PM Walker Carlson wrote: > Hello all, > > I have picked up and updated KIP-150. Due to changes to the project since > KIP #150 was written there are a few items that

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

The KIP and PR have been updated please go take a look and vote. For those worried about the [DISCUSS] Streams DSL/StateStore Refactoring email thread affecting this I believe the cogroup methods fit well into the streams dsl and won't need to change. We can update the aggregate methods in the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

+1 Thanks, Bill On Wed, Jun 14, 2017 at 8:10 PM, Xavier Léauté wrote: > +1 from me > > any stream should be able to initialize the cogroup > > On Wed, Jun 14, 2017 at 3:44 PM Kyle Winkelman > wrote: > > > I will update the kip to have only the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

+1 from me any stream should be able to initialize the cogroup On Wed, Jun 14, 2017 at 3:44 PM Kyle Winkelman wrote: > I will update the kip to have only the aggregator in the first cogroup call > and the initializer and serde in the aggregate calls. > > This seems to

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I will update the kip to have only the aggregator in the first cogroup call and the initializer and serde in the aggregate calls. This seems to be the consensus on the email chain. Thanks, Kyle On Jun 14, 2017 5:41 PM, wrote: That is not the case. No matter which stream the record comes in on

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

That is not the case. No matter which stream the record comes in on the initializer is called if there is not currently an object in the store. On Jun 14, 2017 5:12 PM, "Guozhang Wang" wrote: > While regarding where we should ask users to set serdes: I think I'm > convinced

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

While regarding where we should ask users to set serdes: I think I'm convinced by Xavier that they should be in the `aggregate` call (but only those does not pass in a state store supplier) instead of the `KStream#cogroup` call to be consistent with other aggregate functions. BTW another

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

To clarify it isn't required to have the initializer in the first cogroup because the first aggregator will have the value type. I like how the initializer makes it abundantly clear that the final type will be that. Right now I'm split because the case may be made that you want to supply a

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I'd suggest we do not block this KIP until the serde work has been sorted out: we cannot estimate yet how long it will take yet. Instead let's say make an agreement on where we want to specify the serdes: whether on the first co-group call or on the aggregate call. Also about the initializer

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

+1 on deferring discussion on Serdes until API improvements are ironed out. On Tue, Jun 13, 2017 at 2:06 PM, Matthias J. Sax wrote: > Hi, > > I am just catching up on this thread. (1) as most people agree, we > should not add anything to KStreamBuilder (btw: we actually

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi, I am just catching up on this thread. (1) as most people agree, we should not add anything to KStreamBuilder (btw: we actually plan to move #merge() to KStream and deprecate it on KStreamBuilder as it's a quite unnatural API atm). About specifying Serdes: there is still the idea to improve

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

You're right, I haven't thought of that. Cheers, Michał On 13/06/17 13:00, Kyle Winkelman wrote: First, I would prefer not calling it aggregate because there are already plenty of aggregate methods. Second, I dont think this would really work because after each aggregate you now have a

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

First, I would prefer not calling it aggregate because there are already plenty of aggregate methods. Second, I dont think this would really work because after each aggregate you now have a unique KTable (someone may want a table with 4 streams and reuse those 4 in another table but with one more

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I think we are discussing two separate things here, so it might be worth clarifying: 1) the position of the initializer with respect to the aggregators. If I understand correctly, Guozhang seems to think it is more natural to specify the initializer first, despite it not bearing any relation to

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Note that although the internal `AbstractStoreSupplier` does maintain the key-value serdes, we do not enforce the interface of `StateStoreSupplier` to always retain that information, and hence we cannot assume that StateStoreSuppliers always retain key / value serdes. On Thu, Jun 8, 2017 at 11:51

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I chose the current way so if you make multiple tables you don't need to supply the serde and initializer multiple times. It is true that you wouldnt need the serde if you use a statestoresupplier but I think we could note that in the method call. I am fine with the first option if thats what

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Another reason for the serde not to be in the first cogroup call, is that the serde should not be required if you pass a StateStoreSupplier to aggregate() Regarding the aggregated type I don't the why initializer should be favored over aggregator to define the type. In my mind separating the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

On a second thought... This is the current proposal API ``` CogroupedKStream cogroup(final Initializer initializer, final Aggregator aggregator, final Serde aggValueSerde) ``` If we do not have the initializer in the first co-group it might be a bit awkward for users to specify the

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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é wrote: > I have a minor suggestion to make the API a little bit more symmetric. > I feel it would make more sense to

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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 wrote: > This makes much more sense to me. +1 > > > On Jun 1, 2017, at 10:33 AM, Kyle Winkelman > wrote: > > > > I have updated

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

This makes much more sense to me. +1 > On Jun 1, 2017, at 10:33 AM, Kyle Winkelman 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,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Kyle, it might make sense to start a VOTE thread, it doesn’t seem there are more comments. Thanks Eno > On May 16, 2017, at 6:42 PM, Damian Guy wrote: > > Yeah - i wish we'd named KGroupedStream GroupedKStream, similarly for > KGroupedTable. > > On Tue, 16 May 2017 at

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Yeah - i wish we'd named KGroupedStream GroupedKStream, similarly for KGroupedTable. On Tue, 16 May 2017 at 17:59 Kyle Winkelman wrote: > I have added code blocks and a note about the partial results. > > Can I ask why you dont like KCogroupedStream? I just think that

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I have added code blocks and a note about the partial results. Can I ask why you dont like KCogroupedStream? I just think that because it is created from a KGroupedStream we should keep a similar name format. On May 16, 2017 9:23 AM, "Damian Guy" wrote: Hi Kyle, Can you

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Hi Kyle, Can you put the code examples etc in {code} blocks to make it easier to read? I think this is probably a pretty common use-case and therefore a worthwhile addition to the API. I'd suggest dropping the K from KCogroupedStream and calling it CogroupedStream or CogroupedKStream. In your

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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" wrote: > Hi Kyle, > > Sorry for the delay in reviews, tomorrow is feature freeze deadline ( >

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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 ) so people are busier than usual. Stay tuned. Eno > On 15 May 2017,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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" wrote: > Eno, is there anyone

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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"

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

Thanks Kyle, good arguments. Eno > On May 7, 2017, at 5:06 PM, Kyle Winkelman 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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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,

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

I’ll look as well asap, sorry, been swamped. Eno > On May 4, 2017, at 6:17 PM, Damian Guy 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 >

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

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"