beam_trigger.java

PCollection<KV<Integer, Iterable<Alert>>> sessionedAlerts = alerts
        .apply(Window.remerge())
        .apply(Window.into(Sessions.withGapDuration(Duration.standardHours(1))))
        .apply(MapElements.via((Alert event) -> KV.of(event.userId, event)).withOutputType(new TypeDescriptor<KV<Integer, Alert>>() {}))
        .apply(GroupByKey.create());

PCollection<Alert> throttledAlerts = sessionedAlerts
        .apply(FlatMapElements.via((KV<Integer, Iterable<Alert>> pair) -> {
            ArrayList<Alert> acc = new ArrayList<>();
            FluentIterable
                .from(pair.getValue())
                .toSortedList(Comparator.comparing(a -> a.timestamp))
                .forEach((Alert alert) -> {
                if (acc.isEmpty() || Iterables.getLast(acc).timestamp.plus(Duration.standardMinutes(60)).isBefore(alert.timestamp)) {
                    acc.add(alert);
                }
                });
            return acc;
        }).withOutputType(new TypeDescriptor<Alert>() {}));