Aggregate stream with Storm