StreamAggregator { public static StreamAggregatorBuilder builder() { return new StreamAggregatorBuilder(); }  static StreamAggregatorBuilder builder();  }
@Test public void test() throws Exception { SiddhiAggregator aggregator = StreamAggregator.builder() .columnDef("host", "string") .columnDef("timestamp", "long") .columnDef("metric", "string") .columnDef("site", "string") .columnDef("value", "double") .filter("metric==\"missingblocks\"") .groupby("host") .agg("avg", "value", "avg") .timeColumn("timestamp") .window("1 min", 0) .aggregateHandler(new AggregateHandler() { @Override public void onAggregate(List<AggregateResult> result) { System.out.println(result); } }) .build(); aggregator.add(new Object[] {"host1", 1000L, "missingblocks", "site1", 10.0}); aggregator.add(new Object[] {"host2", 2000L, "missingblocks", "site1", 16.0}); aggregator.add(new Object[] {"host3", 2000L, "missingblocks", "site1", 11.0}); aggregator.add(new Object[] {"host1", 21000L, "missingblocks", "site1", 20.0}); aggregator.add(new Object[] {"host1", 61000L, "missingblocks", "site1", 14.0}); aggregator.add(new Object[] {"host2", 61500L, "missingblocks", "site1", 14.0}); aggregator.add(new Object[] {"host3", 62000L, "missingblocks", "site1", 13.0}); aggregator.add(new Object[] {"host2", 63500L, "missingblocks", "site1", 19.0}); aggregator.add(new Object[] {"host1", 121000L, "missingblocks", "site1", 14.0}); aggregator.add(new Object[] {"host2", 121000L, "missingblocks", "site1", 14.0}); aggregator.add(new Object[] {"host3", 122000L, "missingblocks", "site1", 13.0}); }