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!