Hi Jan, Thanks very much for the input.
On Tue, 4 Jul 2017 at 08:54 Jan Filipiak <jan.filip...@trivago.com> wrote: > Hi Damian, > > I do see your point of something needs to change. But I fully agree with > Gouzhang when he says. > --- > > But since this is a incompatibility change, and we are going to remove the > compatibility annotations soon it means we only have one chance and we > really have to make it right. > ---- > > I think we all agree on this one! Hence the discussion. > I fear all suggestions do not go far enough to become something that will > carry on for very much longer. > I am currently working on KAFKA-3705 and try to find the most easy way for > the user to give me all the required functionality. The easiest interface I > could come up so far can be looked at here. > > > https://github.com/Kaiserchen/kafka/blob/3da2b8f787a5d30dee2de71cf0f125ab3e57d89b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java#L622 > > And its already horribly complicated. I am currently unable to find the > right abstraction level to have everything falling into place naturally. To > be honest I already think introducing > > To be fair that is not a particularly easy problem to solve! > > https://github.com/Kaiserchen/kafka/blob/3da2b8f787a5d30dee2de71cf0f125ab3e57d89b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java#L493 > > was unideal and makes everything a mess. I'm not sure i agree that it makes everything a mess, but It could have been done differently. The JoinType:Whatever is also not really flexible. 2 things come to my mind: > > 1. I don't think we should rule out config based decisions say configs like > streams.$applicationID.joins.$joinname.conf = value > Is this just for config? Or are you suggesting that we could somehow "code" the join in a config file? > This can allow for tremendous changes without single API change and IMO it > was not considered enough yet. > > 2. Push logic from the DSL to the Callback classes. A ValueJoiner for > example can be used to implement different join types as the user wishes. > Do you have an example of how this might look? > As Gouzhang said: stopping to break users is very important. Of course. We want to make it as easy as possible for people to use streams. especially with this changes + All the plans I sadly only have in my head > but hopefully the first link can give a glimpse. > > Thanks for preparing the examples made it way clearer to me what exactly > we are talking about. I would argue to go a bit slower and more carefull on > this one. At some point we need to get it right. Peeking over to the hadoop > guys with their hughe userbase. Config files really work well for them. > > Best Jan > > > > > > On 30.06.2017 09:31, Damian Guy wrote: > > Thanks Matthias > > > > On Fri, 30 Jun 2017 at 08:05 Matthias J. Sax <matth...@confluent.io> > wrote: > > > >> I am just catching up on this thread, so sorry for the long email in > >> advance... Also, it's to some extend a dump of thoughts and not always a > >> clear proposal. Still need to think about this in more detail. But maybe > >> it helps other to get new ideas :) > >> > >> > >>>> However, I don't understand your argument about putting aggregate() > >>>> after the withXX() -- all the calls to withXX() set optional > parameters > >>>> for aggregate() and not for groupBy() -- but a groupBy().withXX() > >>>> indicates that the withXX() belongs to the groupBy(). IMHO, this might > >>>> be quite confusion for developers. > >>>> > >>>> > >>> I see what you are saying, but the grouped stream is effectively a > no-op > >>> until you call one of the aggregate/count/reduce etc functions. So the > >>> optional params are ones that are applicable to any of the operations > you > >>> can perform on this grouped stream. Then the final > >>> count()/reduce()/aggregate() call has any of the params that are > >>> required/specific to that function. > >>> > >> I understand your argument, but you don't share the conclusion. If we > >> need a "final/terminal" call, the better way might be > >> > >> .groupBy().count().withXX().build() > >> > >> (with a better name for build() though) > >> > >> > > The point is that all the other calls, i.e,withBlah, windowed, etc apply > > too all the aggregate functions. The terminal call being the actual type > of > > aggregation you want to do. I personally find this more natural than > > groupBy().count().withBlah().build() > > > > > >>> groupedStream.count(/** non windowed count**/) > >>> groupedStream.windowed(TimeWindows.of(10L)).count(...) > >>> groupedStream.sessionWindowed(SessionWindows.of(10L)).count(...) > >> > >> I like this. However, I don't see a reason to have windowed() and > >> sessionWindowed(). We should have one top-level `Windows` interface that > >> both `TimeWindows` and `SessionWindows` implement and just have a single > >> windowed() method that accepts all `Windows`. (I did not like the > >> separation of `SessionWindows` in the first place, and this seems to be > >> an opportunity to clean this up. It was hard to change when we > >> introduced session windows) > >> > > Yes - true we should look into that. > > > > > >> Btw: we do you the imperative groupBy() and groupByKey(), and thus we > >> might also want to use windowBy() (instead of windowed()). Not sure how > >> important this is, but it seems to be inconsistent otherwise. > >> > >> > > Makes sense > > > > > >> About joins: I don't like .withJoinType(JoinType.LEFT) at all. I think, > >> defining an inner/left/outer join is not an optional argument but a > >> first class concept and should have a proper representation in the API > >> (like the current methods join(), leftJoin, outerJoin()). > >> > >> > > Yep, i did originally have it as a required param and maybe that is what > we > > go with. It could have a default, but maybe that is confusing. > > > > > > > >> About the two join API proposals, the second one has too much boiler > >> plate code for my taste. Also, the actual join() operator has only one > >> argument what is weird to me, as in my thinking process, the main > >> operator call, should have one parameter per mandatory argument but your > >> proposal put the mandatory arguments into Joins.streamStreamJoin() call. > >> This is far from intuitive IMHO. > >> > >> > > This is the builder pattern, you only need one param as the builder has > > captured all of the required and optional arguments. > > > > > >> The first join proposal also seems to align better with the pattern > >> suggested for aggregations and having the same pattern for all operators > >> is important (as you stated already). > >> > >> > > This is why i offered two alternatives as i started out with. 1 is the > > builder pattern, the other is the more fluent pattern. > > > > > >> > >> Coming back to the config vs optional parameter. What about having a > >> method withConfig[s](...) that allow to put in the configuration? > >> > >> > > Sure, it is currently called withLogConfig() as that is the only thing > that > > is really config. > > > > > >> This also raises the question if until() is a windows property? > >> Actually, until() seems to be a configuration parameter and thus, should > >> not not have it's own method. > >> > >> > > Hmmm, i don't agree. Until is a property of the window. It is going to be > > potentially different for every window operation you do in a streams app. > > > > > >> > >> Browsing throw your example DSL branch, I also saw this one: > >> > >>> final KTable<Windowed<String>, Long> windowed> > >> groupedStream.counting() > >>> .windowed(TimeWindows.of(10L).until(10)) > >>> .table(); > >> This is an interesting idea, and it remind my on some feedback about "I > >> wanted to count a stream, but there was no count() method -- I first > >> needed to figure out, that I need to group the stream first to be able > >> to count it. It does make sense in hindsight but was not obvious in the > >> beginning". Thus, carrying out this thought, we could also do the > >> following: > >> > >> stream.count().groupedBy().windowedBy().table(); > >> > >> -> Note, I use "grouped" and "windowed" instead of imperative here, as > >> it comes after the count() > >> > >> This would be more consistent than your proposal (that has grouping > >> before but windowing after count()). It might even allow us to enrich > >> the API with a some syntactic sugar like `stream.count().table()` to get > >> the overall count of all records (this would obviously not scale, but we > >> could support it -- if not now, maybe later). > >> > >> > > I guess i'd prefer > > stream.groupBy().windowBy().count() > > stream.groupBy().windowBy().reduce() > > stream.groupBy().count() > > > > As i said above, everything that happens before the final aggregate call > > can be applied to any of them. So it makes sense to me to do those things > > ahead of the final aggregate call. > > > > > >> Last about builder pattern. I am convinced that we need some "terminal" > >> operator/method that tells us when to add the processor to the topology. > >> But I don't see the need for a plain builder pattern that feels alien to > >> me (see my argument about the second join proposal). Using .stream() / > >> .table() as use in many examples might work. But maybe a more generic > >> name that we can use in all places like build() or apply() might also be > >> an option. > >> > >> > > Sure, a generic name might be ok. > > > > > > > > > >> -Matthias > >> > >> > >> > >> On 6/29/17 7:37 AM, Damian Guy wrote: > >>> Thanks Kyle. > >>> > >>> On Thu, 29 Jun 2017 at 15:11 Kyle Winkelman <winkelman.k...@gmail.com> > >>> wrote: > >>> > >>>> Hi Damian, > >>>> > >>>>>>>> When trying to program in the fluent API that has been discussed > >> most > >>>> it > >>>>>>>> feels difficult to know when you will actually get an object you > can > >>>> reuse. > >>>>>>>> What if I make one KGroupedStream that I want to reuse, is it > legal > >> to > >>>>>>>> reuse it or does this approach expect you to call grouped each > time? > >>>>>> I'd anticipate that once you have a KGroupedStream you can re-use it > >> as > >>>> you > >>>>>> can today. > >>>> You said it yourself in another post that the grouped stream is > >>>> effectively a no-op until a count, reduce, or aggregate. The way I see > >> it > >>>> you wouldn’t be able to reuse anything except KStreams and KTables, > >> because > >>>> most of this fluent api would continue returning this (this being the > >>>> builder object currently being manipulated). > >>> So, if you ever store a reference to anything but KStreams and KTables > >> and > >>>> you use it in two different ways then its possible you make > conflicting > >>>> withXXX() calls on the same builder. > >>>> > >>>> > >>> No necessarily true. It could return a new instance of the builder, > i.e., > >>> the builders being immutable. So if you held a reference to the builder > >> it > >>> would always be the same as it was when it was created. > >>> > >>> > >>>> GroupedStream<K,V> groupedStreamWithDefaultSerdes = kStream.grouped(); > >>>> GroupedStream<K,V> groupedStreamWithDeclaredSerdes = > >>>> groupedStreamsWithDefaultSerdes.withKeySerde(…).withValueSerde(…); > >>>> > >>>> I’ll admit that this shouldn’t happen but some user is going to do it > >>>> eventually… > >>>> Depending on implementation uses of groupedStreamWithDefaultSerdes > would > >>>> most likely be equivalent to the version withDeclaredSerdes. One work > >>>> around would be to always make copies of the config objects you are > >>>> building, but this approach has its own problem because now we have to > >>>> identify which configs are equivalent so we don’t create repeated > >>>> processors. > >>>> > >>>> The point of this long winded example is that we always have to be > >>>> thinking about all of the possible ways it could be misused by a user > >>>> (causing them to see hard to diagnose problems). > >>>> > >>> Exactly! That is the point of the discussion really. > >>> > >>> > >>>> In my attempt at a couple methods with builders I feel that I could > >>>> confidently say the user couldn’t really mess it up. > >>>>> // Count > >>>>> KTable<String, Long> count = > >>>>> > kGroupedStream.count(Count.count().withQueryableStoreName("my-store")); > >>>> The kGroupedStream is reusable and if they attempted to reuse the > Count > >>>> for some reason it would throw an error message saying that a store > >> named > >>>> “my-store” already exists. > >>>> > >>>> > >>> Yes i agree and i think using builders is my preferred pattern. > >>> > >>> Cheers, > >>> Damian > >>> > >>> > >>>> Thanks, > >>>> Kyle > >>>> > >>>> From: Damian Guy > >>>> Sent: Thursday, June 29, 2017 3:59 AM > >>>> To: d...@kafka.apache.org > >>>> Subject: Re: [DISCUSS] Streams DSL/StateStore Refactoring > >>>> > >>>> Hi Kyle, > >>>> > >>>> Thanks for your input. Really appreciated. > >>>> > >>>> On Thu, 29 Jun 2017 at 06:09 Kyle Winkelman <winkelman.k...@gmail.com > > > >>>> wrote: > >>>> > >>>>> I like more of a builder pattern even though others have voiced > against > >>>>> it. The reason I like it is because it makes it clear to the user > that > >> a > >>>>> call to KGroupedStream#count will return a KTable not some > intermediate > >>>>> class that I need to undetstand. > >>>>> > >>>> Yes, that makes sense. > >>>> > >>>> > >>>>> When trying to program in the fluent API that has been discussed most > >> it > >>>>> feels difficult to know when you will actually get an object you can > >>>> reuse. > >>>>> What if I make one KGroupedStream that I want to reuse, is it legal > to > >>>>> reuse it or does this approach expect you to call grouped each time? > >>>> > >>>> I'd anticipate that once you have a KGroupedStream you can re-use it > as > >> you > >>>> can today. > >>>> > >>>> > >>>>> This question doesn’t pop into my head at all in the builder pattern > I > >>>>> assume I can reuse everything. > >>>>> Finally, I like .groupByKey and .groupBy(KeyValueMapper) not a big > fan > >> of > >>>>> the grouped. > >>>>> > >>>>> Yes, grouped() was more for demonstration and because groupBy() and > >>>> groupByKey() were taken! So i'd imagine the api would actually want to > >> be > >>>> groupByKey(/** no required args***/).withOptionalArg() and > >>>> groupBy(KeyValueMapper m).withOpitionalArg(...) of course this all > >> depends > >>>> on maintaining backward compatibility. > >>>> > >>>> > >>>>> Unfortunately, the below approach would require atleast 2 (probably > 3) > >>>>> overloads (one for returning a KTable and one for returning a KTable > >> with > >>>>> Windowed Key, probably would want to split windowed and > sessionwindowed > >>>> for > >>>>> ease of implementation) of each count, reduce, and aggregate. > >>>>> Obviously not exhaustive but enough for you to get the picture. > Count, > >>>>> Reduce, and Aggregate supply 3 static methods to initialize the > >> builder: > >>>>> // Count > >>>>> KTable<String, Long> count = > >>>>> > groupedStream.count(Count.count().withQueryableStoreName("my-store")); > >>>>> > >>>>> // Windowed Count > >>>>> KTable<Windowed<String>, Long> windowedCount = > >>>>> > >> > groupedStream.count(Count.windowed(TimeWindows.of(10L).until(10)).withQueryableStoreName("my-windowed-store")); > >>>>> // Session Count > >>>>> KTable<Windowed<String>, Long> sessionCount = > >>>>> > >> > groupedStream.count(Count.sessionWindowed(SessionWindows.with(10L)).withQueryableStoreName("my-session-windowed-store")); > >>>>> > >>>> Above and below, i think i'd prefer it to be: > >>>> groupedStream.count(/** non windowed count**/) > >>>> groupedStream.windowed(TimeWindows.of(10L)).count(...) > >>>> groupedStream.sessionWindowed(SessionWindows.of(10L)).count(...) > >>>> > >>>> > >>>> > >>>> > >>>>> // Reduce > >>>>> Reducer<Long> reducer; > >>>>> KTable<String, Long> reduce = groupedStream.reduce(reducer, > >>>>> Reduce.reduce().withQueryableStoreName("my-store")); > >>>>> > >>>>> // Aggregate Windowed with Custom Store > >>>>> Initializer<String> initializer; > >>>>> Aggregator<String, Long, String> aggregator; > >>>>> KTable<Windowed<String>, String> aggregate = > >>>>> groupedStream.aggregate(initializer, aggregator, > >>>>> > >> > Aggregate.windowed(TimeWindows.of(10L).until(10)).withStateStoreSupplier(stateStoreSupplier))); > >>>>> // Cogroup SessionWindowed > >>>>> KTable<String, String> cogrouped = > groupedStream1.cogroup(aggregator1) > >>>>> .cogroup(groupedStream2, aggregator2) > >>>>> .aggregate(initializer, aggregator, > >>>>> Aggregate.sessionWindowed(SessionWindows.with(10L), > >>>>> sessionMerger).withQueryableStoreName("my-store")); > >>>>> > >>>>> > >>>>> > >>>>> public class Count { > >>>>> > >>>>> public static class Windowed extends Count { > >>>>> private Windows windows; > >>>>> } > >>>>> public static class SessionWindowed extends Count { > >>>>> private SessionWindows sessionWindows; > >>>>> } > >>>>> > >>>>> public static Count count(); > >>>>> public static Windowed windowed(Windows windows); > >>>>> public static SessionWindowed sessionWindowed(SessionWindows > >>>>> sessionWindows); > >>>>> > >>>>> // All withXXX(...) methods. > >>>>> } > >>>>> > >>>>> public class KGroupedStream { > >>>>> public KTable<K, Long> count(Count count); > >>>>> public KTable<Windowed<K>, Long> count(Count.Windowed count); > >>>>> public KTable<Windowed<K>, Long> count(Count.SessionWindowed > >> count); > >>>>> … > >>>>> } > >>>>> > >>>>> > >>>>> Thanks, > >>>>> Kyle > >>>>> > >>>>> From: Guozhang Wang > >>>>> Sent: Wednesday, June 28, 2017 7:45 PM > >>>>> To: d...@kafka.apache.org > >>>>> Subject: Re: [DISCUSS] Streams DSL/StateStore Refactoring > >>>>> > >>>>> I played the current proposal a bit with > >> https://github.com/dguy/kafka/ > >>>>> tree/dsl-experiment < > https://github.com/dguy/kafka/tree/dsl-experiment > >>> , > >>>>> and here are my observations: > >>>>> > >>>>> 1. Personally I prefer > >>>>> > >>>>> "stream.group(mapper) / stream.groupByKey()" > >>>>> > >>>>> than > >>>>> > >>>>> "stream.group().withKeyMapper(mapper) / stream.group()" > >>>>> > >>>>> Since 1) withKeyMapper is not enforced programmatically though it is > >> not > >>>>> "really" optional like others, 2) syntax-wise it reads more natural. > >>>>> > >>>>> I think it is okay to add the APIs in ( > >>>>> > >>>>> > >> > https://github.com/dguy/kafka/blob/dsl-experiment/streams/src/main/java/org/apache/kafka/streams/kstream/GroupedStream.java > >>>>> ) > >>>>> in KGroupedStream. > >>>>> > >>>>> > >>>>> 2. For the "withStateStoreSupplier" API, are the user supposed to > pass > >> in > >>>>> the most-inner state store supplier (e.g. then one whose get() return > >>>>> RocksDBStore), or it is supposed to return the most-outer supplier > with > >>>>> logging / metrics / etc? I think it would be more useful to only > >> require > >>>>> users pass in the inner state store supplier while specifying > caching / > >>>>> logging through other APIs. > >>>>> > >>>>> In addition, the "GroupedWithCustomStore" is a bit suspicious to me: > we > >>>> are > >>>>> allowing users to call other APIs like "withQueryableName" multiple > >> time, > >>>>> but only call "withStateStoreSupplier" only once in the end. Why is > >> that? > >>>>> > >>>>> 3. The current DSL seems to be only for aggregations, what about > joins? > >>>>> > >>>>> > >>>>> 4. I think it is okay to keep the "withLogConfig": for the > >>>>> StateStoreSupplier it will still be user code specifying the topology > >> so > >>>> I > >>>>> do not see there is a big difference. > >>>>> > >>>>> > >>>>> 5. "WindowedGroupedStream" 's withStateStoreSupplier should take the > >>>>> windowed state store supplier to enforce typing? > >>>>> > >>>>> > >>>>> Below are minor ones: > >>>>> > >>>>> 6. "withQueryableName": maybe better "withQueryableStateName"? > >>>>> > >>>>> 7. "withLogConfig": maybe better "withLoggingTopicConfig()"? > >>>>> > >>>>> > >>>>> > >>>>> Guozhang > >>>>> > >>>>> > >>>>> > >>>>> On Wed, Jun 28, 2017 at 3:59 PM, Matthias J. Sax < > >> matth...@confluent.io> > >>>>> wrote: > >>>>> > >>>>>> I see your point about "when to add the processor to the topology". > >>>> That > >>>>>> is indeed an issue. Not sure it we could allow "updates" to the > >>>>> topology... > >>>>>> I don't see any problem with having all the withXX() in KTable > >>>> interface > >>>>>> -- but this might be subjective. > >>>>>> > >>>>>> > >>>>>> However, I don't understand your argument about putting aggregate() > >>>>>> after the withXX() -- all the calls to withXX() set optional > >> parameters > >>>>>> for aggregate() and not for groupBy() -- but a groupBy().withXX() > >>>>>> indicates that the withXX() belongs to the groupBy(). IMHO, this > might > >>>>>> be quite confusion for developers. > >>>>>> > >>>>>> > >>>>>> -Matthias > >>>>>> > >>>>>> On 6/28/17 2:55 AM, Damian Guy wrote: > >>>>>>>> I also think that mixing optional parameters with configs is a bad > >>>>> idea. > >>>>>>>> Have not proposal for this atm but just wanted to mention it. Hope > >>>> to > >>>>>>>> find some time to come up with something. > >>>>>>>> > >>>>>>>> > >>>>>>> Yes, i don't like the mix of config either. But the only real > config > >>>>> here > >>>>>>> is the logging config - which we don't really need as it can > already > >>>> be > >>>>>>> done via a custom StateStoreSupplier. > >>>>>>> > >>>>>>> > >>>>>>>> What I don't like in the current proposal is the > >>>>>>>> .grouped().withKeyMapper() -- the current solution with > >>>> .groupBy(...) > >>>>>>>> and .groupByKey() seems better. For clarity, we could rename to > >>>>>>>> .groupByNewKey(...) and .groupByCurrentKey() (even if we should > find > >>>>>>>> some better names). > >>>>>>>> > >>>>>>>> > >>>>>>> it could be groupByKey(), groupBy() or something different bt > >>>>>>> > >>>>>>> > >>>>>>> > >>>>>>>> The proposed pattern "chains" grouping and aggregation too close > >>>>>>>> together. I would rather separate both more than less, ie, do into > >>>> the > >>>>>>>> opposite direction. > >>>>>>>> > >>>>>>>> I am also wondering, if we could so something more "fluent". The > >>>>> initial > >>>>>>>> proposal was like: > >>>>>>>> > >>>>>>>>>> groupedStream.count() > >>>>>>>>>> .withStoreName("name") > >>>>>>>>>> .withCachingEnabled(false) > >>>>>>>>>> .withLoggingEnabled(config) > >>>>>>>>>> .table() > >>>>>>>> The .table() statement in the end was kinda alien. > >>>>>>>> > >>>>>>> I agree, but then all of the withXXX methods need to be on KTable > >>>> which > >>>>>> is > >>>>>>> worse in my opinion. You also need something that is going to > "build" > >>>>> the > >>>>>>> internal processors and add them to the topology. > >>>>>>> > >>>>>>> > >>>>>>>> The current proposal put the count() into the end -- ie, the > >>>> optional > >>>>>>>> parameter for count() have to specified on the .grouped() call -- > >>>> this > >>>>>>>> does not seems to be the best way either. > >>>>>>>> > >>>>>>>> > >>>>>>> I actually prefer this method as you are building a grouped stream > >>>> that > >>>>>> you > >>>>>>> will aggregate. So > >>>> table.grouped(...).withOptionalStuff().aggregate(..) > >>>>>> etc > >>>>>>> seems natural to me. > >>>>>>> > >>>>>>> > >>>>>>>> I did not think this through in detail, but can't we just do the > >>>>> initial > >>>>>>>> proposal with the .table() ? > >>>>>>>> > >>>>>>>> groupedStream.count().withStoreName("name").mapValues(...) > >>>>>>>> > >>>>>>>> Each .withXXX(...) return the current KTable and all the > .withXXX() > >>>>> are > >>>>>>>> just added to the KTable interface. Or do I miss anything why this > >>>>> wont' > >>>>>>>> work or any obvious disadvantage? > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>> See above. > >>>>>>> > >>>>>>> > >>>>>>>> -Matthias > >>>>>>>> > >>>>>>>> On 6/22/17 4:06 AM, Damian Guy wrote: > >>>>>>>>> Thanks everyone. My latest attempt is below. It builds on the > >>>> fluent > >>>>>>>>> approach, but i think it is slightly nicer. > >>>>>>>>> I agree with some of what Eno said about mixing configy stuff in > >>>> the > >>>>>> DSL, > >>>>>>>>> but i think that enabling caching and enabling logging are things > >>>>> that > >>>>>>>>> aren't actually config. I'd probably not add withLogConfig(...) > >>>> (even > >>>>>>>>> though it is below) as this is actually config and we already > have > >>>> a > >>>>>> way > >>>>>>>> of > >>>>>>>>> doing that, via the StateStoreSupplier. Arguably we could use the > >>>>>>>>> StateStoreSupplier for disabling caching etc, but as it stands > that > >>>>> is > >>>>>> a > >>>>>>>>> bit of a tedious process for someone that just wants to use the > >>>>> default > >>>>>>>>> storage engine, but not have caching enabled. > >>>>>>>>> > >>>>>>>>> There is also an orthogonal concern that Guozhang alluded to.... > If > >>>>> you > >>>>>>>>> want to plug in a custom storage engine and you want it to be > >>>> logged > >>>>>> etc, > >>>>>>>>> you would currently need to implement that yourself. Ideally we > can > >>>>>>>> provide > >>>>>>>>> a way where we will wrap the custom store with logging, metrics, > >>>>> etc. I > >>>>>>>>> need to think about where this fits, it is probably more > >>>> appropriate > >>>>> on > >>>>>>>> the > >>>>>>>>> Stores API. > >>>>>>>>> > >>>>>>>>> final KeyValueMapper<String, String, Long> keyMapper = null; > >>>>>>>>> // count with mapped key > >>>>>>>>> final KTable<Long, Long> count = stream.grouped() > >>>>>>>>> .withKeyMapper(keyMapper) > >>>>>>>>> .withKeySerde(Serdes.Long()) > >>>>>>>>> .withValueSerde(Serdes.String()) > >>>>>>>>> .withQueryableName("my-store") > >>>>>>>>> .count(); > >>>>>>>>> > >>>>>>>>> // windowed count > >>>>>>>>> final KTable<Windowed<String>, Long> windowedCount = > >>>> stream.grouped() > >>>>>>>>> .withQueryableName("my-window-store") > >>>>>>>>> .windowed(TimeWindows.of(10L).until(10)) > >>>>>>>>> .count(); > >>>>>>>>> > >>>>>>>>> // windowed reduce > >>>>>>>>> final Reducer<String> windowedReducer = null; > >>>>>>>>> final KTable<Windowed<String>, String> windowedReduce = > >>>>>> stream.grouped() > >>>>>>>>> .withQueryableName("my-window-store") > >>>>>>>>> .windowed(TimeWindows.of(10L).until(10)) > >>>>>>>>> .reduce(windowedReducer); > >>>>>>>>> > >>>>>>>>> final Aggregator<String, String, Long> aggregator = null; > >>>>>>>>> final Initializer<Long> init = null; > >>>>>>>>> > >>>>>>>>> // aggregate > >>>>>>>>> final KTable<String, Long> aggregate = stream.grouped() > >>>>>>>>> .withQueryableName("my-aggregate-store") > >>>>>>>>> .aggregate(aggregator, init, Serdes.Long()); > >>>>>>>>> > >>>>>>>>> final StateStoreSupplier<KeyValueStore<String, Long>> > >>>>>> stateStoreSupplier > >>>>>>>> = null; > >>>>>>>>> // aggregate with custom store > >>>>>>>>> final KTable<String, Long> aggWithCustomStore = stream.grouped() > >>>>>>>>> .withStateStoreSupplier(stateStoreSupplier) > >>>>>>>>> .aggregate(aggregator, init); > >>>>>>>>> > >>>>>>>>> // disable caching > >>>>>>>>> stream.grouped() > >>>>>>>>> .withQueryableName("name") > >>>>>>>>> .withCachingEnabled(false) > >>>>>>>>> .count(); > >>>>>>>>> > >>>>>>>>> // disable logging > >>>>>>>>> stream.grouped() > >>>>>>>>> .withQueryableName("q") > >>>>>>>>> .withLoggingEnabled(false) > >>>>>>>>> .count(); > >>>>>>>>> > >>>>>>>>> // override log config > >>>>>>>>> final Reducer<String> reducer = null; > >>>>>>>>> stream.grouped() > >>>>>>>>> .withLogConfig(Collections.singletonMap("segment.size", > >>>>> "10")) > >>>>>>>>> .reduce(reducer); > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> If anyone wants to play around with this you can find the code > >>>> here: > >>>>>>>>> https://github.com/dguy/kafka/tree/dsl-experiment > >>>>>>>>> > >>>>>>>>> Note: It won't actually work as most of the methods just return > >>>> null. > >>>>>>>>> Thanks, > >>>>>>>>> Damian > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> On Thu, 22 Jun 2017 at 11:18 Ismael Juma <ism...@juma.me.uk> > >>>> wrote: > >>>>>>>>>> Thanks Damian. I think both options have pros and cons. And both > >>>> are > >>>>>>>> better > >>>>>>>>>> than overload abuse. > >>>>>>>>>> > >>>>>>>>>> The fluent API approach reads better, no mention of builder or > >>>> build > >>>>>>>>>> anywhere. The main downside is that the method signatures are a > >>>>> little > >>>>>>>> less > >>>>>>>>>> clear. By reading the method signature, one doesn't necessarily > >>>>> knows > >>>>>>>> what > >>>>>>>>>> it returns. Also, one needs to figure out the special method > >>>>>> (`table()` > >>>>>>>> in > >>>>>>>>>> this case) that gives you what you actually care about (`KTable` > >>>> in > >>>>>> this > >>>>>>>>>> case). Not major issues, but worth mentioning while doing the > >>>>>>>> comparison. > >>>>>>>>>> The builder approach avoids the issues mentioned above, but it > >>>>> doesn't > >>>>>>>> read > >>>>>>>>>> as well. > >>>>>>>>>> > >>>>>>>>>> Ismael > >>>>>>>>>> > >>>>>>>>>> On Wed, Jun 21, 2017 at 3:37 PM, Damian Guy < > damian....@gmail.com > >>>>>>>> wrote: > >>>>>>>>>>> Hi, > >>>>>>>>>>> > >>>>>>>>>>> I'd like to get a discussion going around some of the API > choices > >>>>>> we've > >>>>>>>>>>> made in the DLS. In particular those that relate to stateful > >>>>>> operations > >>>>>>>>>>> (though this could expand). > >>>>>>>>>>> As it stands we lean heavily on overloaded methods in the API, > >>>> i.e, > >>>>>>>> there > >>>>>>>>>>> are 9 overloads for KGroupedStream.count(..)! It is becoming > >>>> noisy > >>>>>> and > >>>>>>>> i > >>>>>>>>>>> feel it is only going to get worse as we add more optional > >>>> params. > >>>>> In > >>>>>>>>>>> particular we've had some requests to be able to turn caching > >>>> off, > >>>>> or > >>>>>>>>>>> change log configs, on a per operator basis (note this can be > >>>> done > >>>>>> now > >>>>>>>>>> if > >>>>>>>>>>> you pass in a StateStoreSupplier, but this can be a bit > >>>>> cumbersome). > >>>>>>>>>>> So this is a bit of an open question. How can we change the DSL > >>>>>>>> overloads > >>>>>>>>>>> so that it flows, is simple to use and understand, and is > easily > >>>>>>>> extended > >>>>>>>>>>> in the future? > >>>>>>>>>>> > >>>>>>>>>>> One option would be to use a fluent API approach for providing > >>>> the > >>>>>>>>>> optional > >>>>>>>>>>> params, so something like this: > >>>>>>>>>>> > >>>>>>>>>>> groupedStream.count() > >>>>>>>>>>> .withStoreName("name") > >>>>>>>>>>> .withCachingEnabled(false) > >>>>>>>>>>> .withLoggingEnabled(config) > >>>>>>>>>>> .table() > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> Another option would be to provide a Builder to the count > method, > >>>>> so > >>>>>> it > >>>>>>>>>>> would look something like this: > >>>>>>>>>>> groupedStream.count(new > >>>>>>>>>>> CountBuilder("storeName").withCachingEnabled(false).build()) > >>>>>>>>>>> > >>>>>>>>>>> Another option is to say: Hey we don't need this, what are you > on > >>>>>>>> about! > >>>>>>>>>>> The above has focussed on state store related overloads, but > the > >>>>> same > >>>>>>>>>> ideas > >>>>>>>>>>> could be applied to joins etc, where we presently have many > join > >>>>>>>> methods > >>>>>>>>>>> and many overloads. > >>>>>>>>>>> > >>>>>>>>>>> Anyway, i look forward to hearing your opinions. > >>>>>>>>>>> > >>>>>>>>>>> Thanks, > >>>>>>>>>>> Damian > >>>>>>>>>>> > >>>>>>>> > >>>>>> > >>>>> > >>>>> -- > >>>>> -- Guozhang > >>>>> > >>>>> > >>>> > >> > >
