beam_sources.java

SlidingWindows window = SlidingWindows.of(Duration.standardMinutes(60)).every(Duration.standardMinutes(5));

// create the sources and map them into POJOs (and window them by the sliding window)
PCollection<KV<Integer, BloodPressureEvent>> bpEvents =
        p.apply(Create.of(bloodPressureRawEvents))
            .apply(ParDo.of(new BloodPressureEventCaster()))
            .apply(Window.into(window))
            .apply(MapElements.via((BloodPressureEvent event) -> KV.of(event.userId, event)).withOutputType(new TypeDescriptor<KV<Integer, BloodPressureEvent>>() {}));
PCollection<KV<Integer, HeartRateEvent>> hrEvents =
        p.apply(Create.of(heartRateRawEvents))
            .apply(ParDo.of(new HeartRateEventCaster()))
            .apply(Window.into(window))
            .apply(MapElements.via((HeartRateEvent event) -> KV.of(event.userId, event)).withOutputType(new TypeDescriptor<KV<Integer, HeartRateEvent>>() {}));