class Distinct implements CombinerAggregator<Set<Integer>> {

  @Override
  public Set<Integer> init(TridentTuple tuple) {
    LOG.debug("get tuple: " + tuple);
    Set<Integer> singleton = new HashSet<Integer>();
    singleton.add(tuple.getInteger(1));
    return singleton;
  }

}