0

I'm writing an application where I'm trying to count the number of users who visit a page every hour. I'm trying to filter to specific events, group by the userId and event hour time, then group by just the hour to get the number of users. But grouping the KTable causes excessive cpu burn and locks when trying to close the streams. Is there a better way to do this?

    events
   .groupBy(...)
   .aggregate(...)
   .groupBy(...);
   .count();
3
  • Since you are into streaming, what if we use something like spark streaming to do all the processing on top of the incoming data consumed from a Kafka queue processed ( in your case, double grouped aggregates) and publish it into something like Druid, Elastic search etc. and get a idea of the event traffic from there ?
    – Infamous
    Jun 26, 2018 at 20:07
  • Just to get it straight: Do you ever need to retrieve the specific user events in an hour time window? Or are your aggregations simply counting a) the number of records per user in a given hour, and b) the total number of events in a given hour Do you ever need to go back in time to see how many events happened in a previous hour? Or is this always real-time, like what is my hourly event rate at the current time? Jun 26, 2018 at 20:52
  • I never need specific user events, I just want to know within an hour time window the number of users that performed a specific action. Jun 27, 2018 at 15:17

1 Answer 1

2

Given the answer to your question above "I just want to know within an hour time window the number of users that perfomed a specific action", I would suggest the following.

Assuming you have a record something like this:

class ActionRecord {
  String actionType;
  String user;
}

You can define an aggregate class something like this:

class ActionRecordAggregate {
  private Set<String> users = new HashSet<>();

  public void add(ActionRecord rec) {
    users.add(rec.getUser());
  }

  public int count() {
    return users.size();
  }

}

Then your streaming app can:

  • accept the events
  • rekey them according to event type (the .map() )
  • group them by event type (.groupByKey())
  • window them by time (selected 1 minute but YMMV)
  • aggregate them into ActionRecordAggregate
  • materialize them into a StateStore

so this looks something like:

stream()
.map((key, val) -> KeyValue.pair(val.actionType, val)) 
.groupByKey() 
.windowedBy(TimeWindows.of(60*1000)) 
.aggregate(
  ActionRecordAggregate::new, 
  (key, value, agg) -> agg.add(value),
  Materialized
      .<String, ActionRecordAggregate, WindowStore<Bytes, byte[]>>as("actionTypeLookup")
      .withValueSerde(getSerdeForActionRecordAggregate())
);

Then, to get the events back, you can query your state store:

ReadOnlyWindowStore<String, ActionRecordAggregate> store = 
  streams.store("actionTypeLookup", QueryableStoreTypes.windowStore());

WindowStoreIterator<ActionRecordAggregate> wIt = 
  store.fetch("actionTypeToGet", startTimestamp, endTimestamp);

int totalCount = 0;
while(wIt.hasNext()) {
  totalCount += wIt.next().count();
}

// totalCount is the number of distinct users in your 
// time interval that raised action type "actionTypeToGet"

Hope this helps!

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.