done per partition now using mapParitions