+1
Thanks,
Bill
On Wed, Jun 14, 2017 at 8:10 PM, Xavier Léauté <xav...@confluent.io> wrote:
> +1 from me
>
> any stream should be able to initialize the cogroup
>
> On Wed, Jun 14, 2017 at 3:44 PM Kyle Winkelman <winkelman.k...@gmail.com>
> 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 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 the
> > initializer is called if there is not currently an object in the store.
> >
> > On Jun 14, 2017 5:12 PM, "Guozhang Wang" <wangg...@gmail.com> wrote:
> >
> > 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 motivation for me to suggest keeping the initializer on the
> > first stream is that by reviewing the PR (some time ago though, so again
> I
> > might be wrong) we will trigger the initializer only when we received an
> > incoming record from the first stream whose key is not in the state store
> > yet, while for other streams we will just drop it on the floor. If that
> is
> > actually not the case, that we call initializer on any one of the
> > co-grouped streams' incoming records, then I'm open to set the
> initializer
> > at the `aggregate` call as well.
> >
> >
> > Guozhang
> >
> > On Wed, Jun 14, 2017 at 2:23 PM, Guozhang Wang <wangg...@gmail.com>
> wrote:
> >
> > > 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 specification I actually felt that the first
> > > cogrouped stream is special (Kyle please feel free to correct me if I'm
> > > wrong) and that is why I thought it is better to specify the
> initializer
> > at
> > > the beginning: since from the typing you can see that the final
> > aggregated
> > > value type is defined to be the same as the first co-grouped stream,
> and
> > > for any intermediate stream to co-group, their value types not be
> > inherited
> > > but the value be "incorporated" into the original stream:
> > >
> > > <T> CogroupedKStream<K, V> cogroup(final KGroupedStream<K, T>
> > > groupedStream, final Aggregator<? super K, ? super T, V> aggregator)
> > >
> > > Note that we do not have a cogroup function that returns
> > > CogroupedKStream<K, T>.
> > >
> > >
> > > Guozhang
> > >
> > >
> > > On Tue, Jun 13, 2017 at 2:31 PM, Bill Bejeck <bbej...@gmail.com>
> wrote:
> > >
> > >> +1 on deferring discussion on Serdes until API improvements are ironed
> > >> out.
> > >>
> > >> On Tue, Jun 13, 2017 at 2:06 PM, Matthias J. Sax <
> matth...@confluent.io
> > >
> > >> 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 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 to
> overall
> > >> > API from the current "we are adding more overloads"-pattern to a
> > >> > builder-like pattern. This might make the whole discussion void if
> we
> > do
> > >> > this. Thus, it might make sense to keep this in mind (or even delay
> > this
> > >> > KIP?). It seems a waste of time to discuss all this if we are going
> to
> > >> > chance it in 2 month anyway... Just saying.
> > >> >
> > >> >
> > >> > -Matthias
> > >> >
> > >> > On 6/13/17 8:05 AM, Michal Borowiecki wrote:
> > >> > > 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 unique KTable (someone may want a table with 4
> > streams
> > >> > and
> > >> > >> reuse those 4 in another table but with one more stream added)
> and
> > >> > unless
> > >> > >> we completely duplicate everything every time this isnt really
> > >> possible.
> > >> > >> Additionally, the cogroup way just requires 1 more call to create
> > two
> > >> > >> different tables (normal, windowed, and session windowed) this
> new
> > >> way
> > >> > >> would require copying the aggregate chain.
> > >> > >>
> > >> > >> Another way to think about it is with cogroup we know that when
> > they
> > >> > call
> > >> > >> aggregate they arent going to be adding any more aggregators to
> > that
> > >> > table
> > >> > >> but your way requires us to assume they are done adding
> aggregators
> > >> > after
> > >> > >> each call so we must return a ktable just to possibly not need to
> > >> have
> > >> > >> created it.
> > >> > >>
> > >> > >> On Jun 13, 2017 5:20 AM, "Michal Borowiecki" <
> > >> > michal.borowie...@openbet.com>
> > >> > >> wrote:
> > >> > >>
> > >> > >>> Actually, just had a thought. It started with naming.
> > >> > >>>
> > >> > >>> Are we actually co-grouping these streams or are we
> co-aggregating
> > >> > them?
> > >> > >>>
> > >> > >>> After all, in each of the cogroup calls we are providing an
> > >> Aggregator
> > >> > >>> implementation.
> > >> > >>>
> > >> > >>>
> > >> > >>> If they are really co-aggregated, why don't we turn this around:
> > >> > >>> 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> coagg = grouped1.aggregate(initializer1,
> > aggregator1,
> > >> > >>> aggValueSerde1) // this is the unchanged aggregate method
> > >> > >>> .aggregate(grouped2, aggregator2) // this is a new
> method
> > >> > >>> .aggregate(grouped3, aggregator3); // ditto
> > >> > >>>
> > >> > >>> This means instead of adding cogroup methods on KGroupStream
> > >> interface,
> > >> > >>> adding aggregate method on KTable interface.
> > >> > >>>
> > >> > >>> Is that feasible?
> > >> > >>>
> > >> > >>> Cheers,
> > >> > >>> Michał
> > >> > >>>
> > >> > >>> On 13/06/17 10:56, Michal Borowiecki wrote:
> > >> > >>>
> > >> > >>> Also, I still feel that putting initializer on the first cogroup
> > can
> > >> > >>> mislead users into thinking the first stream is in some way
> > special.
> > >> > >>> Just my 5c.
> > >> > >>> Michał
> > >> > >>>
> > >> > >>> On 13/06/17 09:54, Michal Borowiecki wrote:
> > >> > >>>
> > >> > >>> Agree completely with the argument for serdes belonging in the
> > same
> > >> > place
> > >> > >>> as the state store name, which is in the aggregate method.
> > >> > >>>
> > >> > >>> Cheers,
> > >> > >>>
> > >> > >>> Michał
> > >> > >>>
> > >> > >>> On 12/06/17 18:20, Xavier Léauté wrote:
> > >> > >>>
> > >> > >>> 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
> the
> > >> first
> > >> > >>> aggregator. I can see the argument for specifying the
> initializer
> > >> > first,
> > >> > >>> but I think it is debatable whether mixing it into the first
> > cogroup
> > >> > call
> > >> > >>> leads to a cleaner API or not.
> > >> > >>>
> > >> > >>> 2) where the serde should be defined (if necessary). Looking at
> > our
> > >> > >>> existing APIs in KGroupedStreams, we always offer two
> aggregate()
> > >> > >>> methods. The first one takes the name of the store and
> associated
> > >> > aggregate
> > >> > >>> value serde e.g. KGroupedStream.aggregate(Initializer<VR>
> > >> initializer,
> > >> > >>> Aggregator<? super K, ? super V, VR> aggregator, Serde<VR>
> > >> > aggValueSerde,
> > >> > >>> String queryableStoreName)
> > >> > >>> The second one only takes a state store supplier, and does not
> > >> specify
> > >> > any
> > >> > >>> serde, e.g. KGroupedStream.aggregate(Initializer<VR>
> > >> > >>> initializer, Aggregator<? super K, ? super V, VR> aggregator,
> > final
> > >> > >>> StateStoreSupplier<KeyValueStore> storeSupplier)
> > >> > >>> Presumably, when specifying a state store supplier it shouldn't
> be
> > >> > >>> necessary to specify an aggregate value serde, since the
> provided
> > >> > >>> statestore might not need to serialize the values (e.g. it may
> > just
> > >> > keep
> > >> > >>> them as regular objects in heap) or it may have its own
> > >> > >>> internal serialization format.
> > >> > >>>
> > >> > >>> For consistency I think it would be valuable to preserve the
> same
> > >> two
> > >> > >>> aggregate methods for cogroup as well. Since the serde is only
> > >> > required in
> > >> > >>> one of the two cases, I believe the serde has no place in the
> > first
> > >> > >>> cogroup() call and should only have to be specified as part of
> the
> > >> > >>> aggregate() method that takes a state store name. In the case
> of a
> > >> > state
> > >> > >>> store supplier, no serde would be necessary.
> > >> > >>>
> > >> > >>>
> > >> > >>> On Sat, Jun 10, 2017 at 4:09 PM Guozhang Wang <
> wangg...@gmail.com
> > >
> > >> > wrote:
> > >> > >>>
> > >> > >>>> I'd agree that the aggregate value serde and the initializer
> does
> > >> not
> > >> > >>>> bear direct relationship with the first `cogroup` calls, but
> > after
> > >> I
> > >> > tried
> > >> > >>>> to write some example code with these two different set of
> APIs I
> > >> > felt the
> > >> > >>>> current APIs just program more naturally.
> > >> > >>>>
> > >> > >>>> I know it is kinda subjective, but I do think that user
> > experience
> > >> > may be
> > >> > >>>> more important as a deciding factor than the logical argument
> for
> > >> > public
> > >> > >>>> interfaces. So I'd recommend people to also try out writing
> some
> > >> > example
> > >> > >>>> lines also and we can circle back and discuss which one feels
> > more
> > >> > natural
> > >> > >>>> to write code.
> > >> > >>>>
> > >> > >>>>
> > >> > >>>> Guozhang
> > >> > >>>>
> > >> > >>>> On Fri, Jun 9, 2017 at 1:59 AM, Michal Borowiecki <
> > >> > >>>> michal.borowie...@openbet.com> wrote:
> > >> > >>>>
> > >> > >>>>> 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.
> > >> > >>>>>
> > >> > >>>>> +1 for moving initializer and serde from cogroup() to the
> > >> aggregate()
> > >> > >>>>> for the reasons mentioned above.
> > >> > >>>>>
> > >> > >>>>> Cheers,
> > >> > >>>>>
> > >> > >>>>> Michał
> > >> > >>>>>
> > >> > >>>>> On 08/06/17 22:44, Guozhang Wang wrote:
> > >> > >>>>>
> > >> > >>>> 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 AM, Xavier Léauté <
> > >> xav...@confluent.io>
> > >> > <xav...@confluent.io> wrote:
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>> 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 <T> I don't the why initializer
> > >> should
> > >> > be
> > >> > >>>>> favored over aggregator to define the type. In my mind
> > separating
> > >> the
> > >> > >>>>> initializer into the last aggregate call clearly indicates
> that
> > >> the
> > >> > >>>>> initializer is independent of any of the aggregators or
> streams
> > >> and
> > >> > that we
> > >> > >>>>> don't wait for grouped1 events to initialize the co-group.
> > >> > >>>>>
> > >> > >>>>> On Thu, Jun 8, 2017 at 11:14 AM Guozhang Wang <
> > wangg...@gmail.com
> > >
> > >> <
> > >> > wangg...@gmail.com> wrote:
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>> On a second thought... This is the current proposal API
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>> ```
> > >> > >>>>>
> > >> > >>>>> <T> CogroupedKStream<K, T> cogroup(final Initializer<T>
> > >> initializer,
> > >> > >>>>>
> > >> > >>>>> final
> > >> > >>>>>
> > >> > >>>>> Aggregator<? super K, ? super V, T> aggregator, final Serde<T>
> > >> > >>>>> aggValueSerde)
> > >> > >>>>>
> > >> > >>>>> ```
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>> If we do not have the initializer in the first co-group it
> might
> > >> be
> > >> > a bit
> > >> > >>>>> awkward for users to specify the aggregator that returns a
> typed
> > >> <T>
> > >> > >>>>>
> > >> > >>>>> value?
> > >> > >>>>>
> > >> > >>>>> Maybe it is still better to put these two functions in the
> same
> > >> api?
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>>
> > >> > >>>>> Guozhang
> > >> > >>>>>
> > >> > >>>>> On Thu, Jun 8, 2017 at 11:08 AM, Guozhang Wang <
> > >> wangg...@gmail.com>
> > >> > <wangg...@gmail.com>
> > >> > >>>>>
> > >> > >>>>> wrote:
> > >> > >>>>>
> > >> > >>>>> 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>
> > >> > <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>
> > >> <
> > >> > 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>
> > <
> > >> > 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> <
> > >> > 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
> > >> > >>>>>
> > >> > >>>>> ...
> > >> > >
> > >> > > --
> > >> > > Signature
> > >> > > <http://www.openbet.com/> Michal Borowiecki
> > >> > > Senior Software Engineer L4
> > >> > > T: +44 208 742 1600 <+44%2020%208742%201600>
> > >> > >
> > >> > >
> > >> > > +44 203 249 8448 <+44%2020%203249%208448>
> > >> > >
> > >> > >
> > >> > >
> > >> > > E: michal.borowie...@openbet.com
> > >> > > W: www.openbet.com <http://www.openbet.com/>
> > >> > >
> > >> > >
> > >> > > OpenBet Ltd
> > >> > >
> > >> > > Chiswick Park Building 9
> > >> > >
> > >> > > 566 Chiswick High Rd
> > >> > >
> > >> > > London
> > >> > >
> > >> > > W4 5XT
> > >> > >
> > >> > > UK
> > >> > >
> > >> > >
> > >> > > <https://www.openbet.com/email_promo>
> > >> > >
> > >> > > This message is confidential and intended only for the addressee.
> If
> > >> you
> > >> > > have received this message in error, please immediately notify the
> > >> > > postmas...@openbet.com <mailto:postmas...@openbet.com> and delete
> > it
> > >> > > from your system as well as any copies. The content of e-mails as
> > well
> > >> > > as traffic data may be monitored by OpenBet for employment and
> > >> security
> > >> > > purposes. To protect the environment please do not print this
> e-mail
> > >> > > unless necessary. OpenBet Ltd. Registered Office: Chiswick Park
> > >> Building
> > >> > > 9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A
> company
> > >> > > registered in England and Wales. Registered no. 3134634. VAT no.
> > >> > > GB927523612
> > >> > >
> > >> >
> > >> >
> > >>
> > >
> > >
> > >
> > > --
> > > -- Guozhang
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
>
