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