cleaned and replaced read master aggregation with pipeTo