code: 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)