TStream<Integer> stream = top.generate(() -> {
  try{
    Thread.sleep(1500);
  }catch(InterruptedException ex){
    //DO NOTHING
  }
  return (int)(Math.random() * 100);
});

TWindow<Integer, Integer> window = stream.last(5, TimeUnit.SECONDS, zero());

TStream<Double> average = window.batch((values, key) -> {
  int sum = values.stream().mapToInt(Integer::intValue).sum();
  return (double)sum / values.size();
});

average.peek(average1 -> LOGGER.info("Calculated average : " + average1));
job = dp.submit(top);