def top(self, num, key=None):
        def topIterator(iterator):
            yield heapq.nlargest(num, iterator, key=key)
        def merge(a, b):
            return heapq.nlargest(num, a + b, key=key)
        return self.mapPartitions(topIterator).reduce(merge)