Added aggregate(...) to PCollection