Streamify reduce_by_key