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 >>>>> >>>> >>> >> >> >
signature.asc
Description: OpenPGP digital signature