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;
  }
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.