basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/aggregation/aggregator/Aggregators.java
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator, N extends Number> GroupByAggregator<A, N> createGroupBy(
      String name, Class<? extends Aggregator> aKlass
  ) {
    Utils.checkState(!started, "Already started");
    GroupByAggregator<A, N> aggregator = new GroupByAggregator(name, aKlass, this);
    dataProvider.addAggregator(aggregator);
    aggregator.setDataProvider(dataProvider);
    return aggregator;
  }
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator, N extends Number> GroupByAggregator<A, N> createGroupBy(
      String name, Class<? extends Aggregator> aKlass
  ) {
    Utils.checkState(!started, "Already started");
    GroupByAggregator<A, N> aggregator = new GroupByAggregator(name, aKlass, this);
    dataProvider.addAggregator(aggregator);
    aggregator.setDataProvider(dataProvider);
    return aggregator;
  }
@ SuppressWarnings ( "unchecked" ) public < A extends SimpleAggregator , N extends Number > GroupByAggregator < A , N > createGroupBy ( String name , Class < ? extends Aggregator > aKlass ) { Utils . checkState ( ! started , "Already started" ) ; GroupByAggregator < A , N > aggregator = new GroupByAggregator ( name , aKlass , this ) ; dataProvider . addAggregator ( aggregator ) ; aggregator . setDataProvider ( dataProvider ) ; return aggregator ; }
Creates a group-by Agregator.

@param name name of the group-by Aggregator.
@param aKlass simple aggregator class backing each group-by element.
@return a group-by aggregator.
Creates a group - by Agregator .