Hey Flávio, Thanks! I haven't got anything usable yet, but I'm working on it now. I'm hoping to push up my branch by the end of the day.
I don't know if you've seen it but Streams actually already has something like this, in the form of caching on materialized stores. If you pass in a "Materialized.withCachingEnabled()", you should be able to get a POC working by setting the max cache size pretty high and setting the commit interval for your desired rate: https://docs.confluent.io/current/streams/developer-guide/memory-mgmt.html#streams-developer-guide-memory-management . There are a couple of cases in joins and whatnot where it doesn't work, but for the aggregations we discussed, it should. The reason for KIP-328 is to provide finer control and hopefully a more straightforward API. Let me know if that works, and I'll drop a message in here when I create the draft PR for KIP-328. I'd really appreciate your feedback. Thanks, -John On Wed, Jul 4, 2018 at 10:17 PM flaviost...@gmail.com <flaviost...@gmail.com> wrote: > John, that was fantastic, man! > Have you built any custom implementation of your KIP in your machine so > that I could test it out here? I wish I could test it out. > If you need any help implementing this feature, please tell me. > > Thanks. > > -Flávio Stutz > > > > > On 2018/07/03 18:04:52, John Roesler <j...@confluent.io> wrote: > > Hi Flávio, > > Thanks! I think that we can actually do this, but the API could be > better. > > I've included Java code below, but I'll copy and modify your example so > > we're on the same page. > > > > EXERCISE 1: > > - The case is "total counting of events for a huge website" > > - Tasks from Application A will have something like: > > .stream(/site-events) > > .transform( re-key s.t. the new key is the partition id) > > .groupByKey() // you have to do this before count > > .count() > > // you explicitly published to a one-partition topic here, but > > it's actually sufficient just > > // to re-group onto one key. You could name and pre-create the > > intermediate topic here, > > // but you don't need a separate application for the final > > aggregation. > > .groupBy((partitionId, partialCount) -> new KeyValue("ALL", > > partialCount)) > > .aggregate(sum up the partialCounts) > > .publish(/counter-total) > > > > I've left out the suppressions, but they would go right after the count() > > and the aggregate(). > > > > With this program, you don't have to worry about the double-aggregation > you > > mentioned in the last email. The KTable produced by the first count() > will > > maintain the correct count per partition. If the value changes for any > > partition, it'll emit a retraction of the old value and then the new > value > > downstream, so that the final aggregation can update itself properly. > > > > I think we can optimize both the execution and the programability by > adding > > a "global aggregation" concept. But In principle, it seems like this > usage > > of the current API will support your use case. > > > > Once again, though, this is just to present an alternative. I haven't > done > > the math on whether your proposal would be more efficient. > > > > Thanks, > > -John > > > > Here's the same algorithm written in Java: > > > > final KStream<String, String> siteEvents = > builder.stream("/site-events"); > > > > // here we re-key the events so that the key is actually the partition > id. > > // we don't need the value to do a count, so I just set it to "1". > > final KStream<Integer, Integer> keyedByPartition = > siteEvents.transform(() > > -> new Transformer<String, String, KeyValue<Integer, Integer>>() { > > private ProcessorContext context; > > > > @Override > > public void init(final ProcessorContext context) { > > this.context = context; > > } > > > > @Override > > public KeyValue<Integer, Integer> transform(final String key, final > > String value) { > > return new KeyValue<>(context.partition(), 1); > > } > > }); > > > > // Note that we can't do "count()" on a KStream, we have to group it > first. > > I'm grouping by the key, so it will produce the count for each key. > > // Since the key is actually the partition id, it will produce the > > pre-aggregated count per partition. > > // Note that the result is a KTable<PartitionId,Count>. It'll always > > contain the most recent count for each partition. > > final KTable<Integer, Long> countsByPartition = > > keyedByPartition.groupByKey().count(); > > > > // Now we get ready for the final roll-up. We re-group all the > constituent > > counts > > final KGroupedTable<String, Long> singlePartition = > > countsByPartition.groupBy((key, value) -> new KeyValue<>("ALL", value)); > > > > final KTable<String, Long> totalCount = singlePartition.reduce((l, r) -> > l > > + r, (l, r) -> l - r); > > > > totalCount.toStream().foreach((k, v) -> { > > // k is always "ALL" > > // v is always the most recent total value > > System.out.println("The total event count is: " + v); > > }); > > > > > > On Tue, Jul 3, 2018 at 9:21 AM flaviost...@gmail.com < > flaviost...@gmail.com> > > wrote: > > > > > Great feature you have there! > > > > > > I'll try to exercise here how we would achieve the same functional > > > objectives using your KIP: > > > > > > EXERCISE 1: > > > - The case is "total counting of events for a huge website" > > > - Tasks from Application A will have something like: > > > .stream(/site-events) > > > .count() > > > .publish(/single-partitioned-topic-with-count-partials) > > > - The published messages will be, for example: > > > ["counter-task1", 2345] > > > ["counter-task2", 8495] > > > ["counter-task3", 4839] > > > - Single Task from Application B will have something like: > > > .stream(/single-partitioned-topic-with-count-partials) > > > .aggregate(by messages whose key starts with "counter") > > > .publish(/counter-total) > > > - FAIL HERE. How would I know what is the overall partitions? Maybe > two > > > partials for the same task will arrive before other tasks and it maybe > > > aggregated twice. > > > > > > I tried to think about using GlobalKTables, but I didn't get an easy > way > > > to aggregate the keys from that table. Do you have any clue? > > > > > > Thanks. > > > > > > -Flávio Stutz > > > > > > > > > > > > > > > > > > > > > /partial-counters-to-single-partitioned-topic > > > > > > On 2018/07/02 20:03:57, John Roesler <j...@confluent.io> wrote: > > > > Hi Flávio, > > > > > > > > Thanks for the KIP. I'll apologize that I'm arriving late to the > > > > discussion. I've tried to catch up, but I might have missed some > nuances. > > > > > > > > Regarding KIP-328, the idea is to add the ability to suppress > > > intermediate > > > > results from all KTables, not just windowed ones. I think this could > > > > support your use case in combination with the strategy that Guozhang > > > > proposed of having one or more pre-aggregation steps that ultimately > push > > > > into a single-partition topic for final aggregation. Suppressing > > > > intermediate results would solve the problem you noted that today > > > > pre-aggregating doesn't do much to staunch the flow up updates. > > > > > > > > I'm not sure if this would be good enough for you overall; I just > wanted > > > to > > > > clarify the role of KIP-328. > > > > In particular, the solution you mentioned is to have the downstream > > > KTables > > > > actually query the upstream ones to compute their results. I'm not > sure > > > > whether it's more efficient to do these queries on the schedule, or > to > > > have > > > > the upstream tables emit their results, on the same schedule. > > > > > > > > What do you think? > > > > > > > > Thanks, > > > > -John > > > > > > > > On Sun, Jul 1, 2018 at 10:03 PM flaviost...@gmail.com < > > > flaviost...@gmail.com> > > > > wrote: > > > > > > > > > For what I understood, that KIP is related to how KStreams will > handle > > > > > KTable updates in Windowed scenarios to optimize resource usage. > > > > > I couldn't see any specific relation to this KIP. Had you? > > > > > > > > > > -Flávio Stutz > > > > > > > > > > > > > > > On 2018/06/29 18:14:46, "Matthias J. Sax" <matth...@confluent.io> > > > wrote: > > > > > > Flavio, > > > > > > > > > > > > thanks for cleaning up the KIP number collision. > > > > > > > > > > > > With regard to KIP-328 > > > > > > ( > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-328%3A+Ability+to+suppress+updates+for+KTables > > > > > ) > > > > > > I am wondering how both relate to each other? > > > > > > > > > > > > Any thoughts? > > > > > > > > > > > > > > > > > > -Matthias > > > > > > > > > > > > On 6/29/18 10:23 AM, flaviost...@gmail.com wrote: > > > > > > > Just copying a follow up from another thread to here (sorry > about > > > the > > > > > mess): > > > > > > > > > > > > > > From: Guozhang Wang <wangg...@gmail.com> > > > > > > > Subject: Re: [DISCUSS] KIP-323: Schedulable KTable as Graph > source > > > > > > > Date: 2018/06/25 22:24:17 > > > > > > > List: dev@kafka.apache.org > > > > > > > > > > > > > > Flávio, thanks for creating this KIP. > > > > > > > > > > > > > > I think this "single-aggregation" use case is common enough > that we > > > > > should > > > > > > > consider how to efficiently supports it: for example, for KSQL > > > that's > > > > > built > > > > > > > on top of Streams, we've seen lots of query statements whose > > > return is > > > > > > > expected a single row indicating the "total aggregate" etc. See > > > > > > > https://github.com/confluentinc/ksql/issues/430 for details. > > > > > > > > > > > > > > I've not read through > > > https://issues.apache.org/jira/browse/KAFKA-6953, > > > > > but > > > > > > > I'm wondering if we have discussed the option of supporting it > in a > > > > > > > "pre-aggregate" manner: that is we do partial aggregates on > > > parallel > > > > > tasks, > > > > > > > and then sends the partial aggregated value via a single topic > > > > > partition > > > > > > > for the final aggregate, to reduce the traffic on that single > > > > > partition and > > > > > > > hence the final aggregate workload. > > > > > > > Of course, for non-commutative aggregates we'd probably need to > > > provide > > > > > > > another API in addition to aggregate, like the `merge` > function for > > > > > > > session-based aggregates, to let users customize the > operations of > > > > > merging > > > > > > > two partial aggregates into a single partial aggregate. What's > its > > > > > pros and > > > > > > > cons compared with the current proposal? > > > > > > > > > > > > > > > > > > > > > Guozhang > > > > > > > On 2018/06/26 18:22:27, Flávio Stutz <flaviost...@gmail.com> > > > wrote: > > > > > > >> Hey, guys, I've just created a new KIP about creating a new > DSL > > > graph > > > > > > >> source for realtime partitioned consolidations. > > > > > > >> > > > > > > >> We have faced the following scenario/problem in a lot of > > > situations > > > > > with > > > > > > >> KStreams: > > > > > > >> - Huge incoming data being processed by numerous > application > > > > > instances > > > > > > >> - Need to aggregate different fields whose records span all > > > topic > > > > > > >> partitions (something like “total amount spent by people aged > > 30 > > > > > yrs” > > > > > > >> when processing a topic partitioned by userid). > > > > > > >> > > > > > > >> The challenge here is to manage this kind of situation > without any > > > > > > >> bottlenecks. We don't need the “global aggregation” to be > > > processed > > > > > at each > > > > > > >> incoming message. On a scenario of 500 instances, each > handling 1k > > > > > > >> messages/s, any single point of aggregation (single > partitioned > > > > > topics, > > > > > > >> global tables or external databases) would create a > bottleneck of > > > 500k > > > > > > >> messages/s for single threaded/CPU elements. > > > > > > >> > > > > > > >> For this scenario, it is possible to store the partial > > > aggregations on > > > > > > >> local stores and, from time to time, query those states and > > > aggregate > > > > > them > > > > > > >> as a single value, avoiding bottlenecks. This is a way to > create a > > > > > "timed > > > > > > >> aggregation barrier”. > > > > > > >> > > > > > > >> If we leverage this kind of built-in feature we could greatly > > > enhance > > > > > the > > > > > > >> ability of KStreams to better handle the CAP Theorem > > > characteristics, > > > > > so > > > > > > >> that one could choose to have Consistency over Availability > when > > > > > needed. > > > > > > >> > > > > > > >> We started this discussion with Matthias J. Sax here: > > > > > > >> https://issues.apache.org/jira/browse/KAFKA-6953 > > > > > > >> > > > > > > >> If you want to see more, go to KIP-326 at: > > > > > > >> > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-326%3A+Schedulable+KTable+as+Graph+source > > > > > > >> > > > > > > >> -Flávio Stutz > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > >
