basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/aggregation/aggregator/Aggregators.java
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator> A createSimple(String name, Class<? extends Aggregator> klass) {
    Utils.checkState(!started, "Already started");
    try {
      A aggregator = (A) CONSTRUCTORS.get(klass).newInstance(name);
      dataProvider.addAggregator(aggregator);
      aggregator.setDataProvider(dataProvider);
      return aggregator;
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator> A createSimple(String name, Class<? extends Aggregator> klass) {
    Utils.checkState(!started, "Already started");
    try {
      A aggregator = (A) CONSTRUCTORS.get(klass).newInstance(name);
      dataProvider.addAggregator(aggregator);
      aggregator.setDataProvider(dataProvider);
      return aggregator;
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
Creates a simple Aggregator.

@param name name of the aggregator.
@param klass Aggregator class
@return a simple aggregator instance.