class_docstr: ExternalSorter will divide the elements into chunks, sort them in
  memory and dump them into disks, finally merge them back.
  
  The spilling will only happen when the used memory goes above
  the limit.
  
  Examples
  --------
  >>> sorter = ExternalSorter(1)  # 1M
  >>> import random
  >>> l = list(range(1024))
  >>> random.shuffle(l)
  >>> sorted(l) == list(sorter.sorted(l))
  True
  >>> sorted(l) == list(sorter.sorted(l, key=lambda x: -x, reverse=True))
  True
class_docstr_tok: ["ExternalSorter","will","divide","the","elements","into","chunks",",","sort","them","in","memory","and","dump","them","into","disks",",","finally","merge","them","back",".","The","spilling","will","only","happen","when","the","used","memory","goes","above","the","limit",".","Examples","--","--","--","--",">",">",">","sorter","=","ExternalSorter","(","1",")","#","1M",">",">",">","import","random",">",">",">","l","=","list","(","range","(","1024",")",")",">",">",">","random.shuffle","(","l",")",">",">",">","sorted","(","l",")","==","list","(","sorter.sorted","(","l",")",")","True",">",">",">","sorted","(","l",")","==","list","(","sorter.sorted","(","l",",","key=lambda","x",":","-x",",","reverse=True",")",")","True"]
docstr: Sort the elements in iterator, do external sort when the memory
  goes above the limit.
docstr_tok: ["Sort","the","elements","in","iterator",",","do","external","sort","when","the","memory","goes","above","the","limit","."]
code: def sorted(self, iterator, key=None, reverse=False):
      """
      Sort the elements in iterator, do external sort when the memory
      goes above the limit.
      """
      global MemoryBytesSpilled, DiskBytesSpilled
      batch, limit = 100, self._next_limit()
      chunks, current_chunk = [], []
      iterator = iter(iterator)
      while True:
          # pick elements in batch
          chunk = list(itertools.islice(iterator, batch))
          current_chunk.extend(chunk)
          if len(chunk) < batch:
              break
  
          used_memory = get_used_memory()
          if used_memory > limit:
              # sort them inplace will save memory
              current_chunk.sort(key=key, reverse=reverse)
              path = self._get_path(len(chunks))
              with open(path, "wb") as f:
                  self.serializer.dump_stream(current_chunk, f)
  
              def load(f):
                  for v in self.serializer.load_stream(f):
                      yield v
                  # close the file explicit once we consume all the items
                  # to avoid ResourceWarning in Python3
                  f.close()
  
              chunks.append(load(open(path, "rb")))
              current_chunk = []
              MemoryBytesSpilled += (
                  max(used_memory - get_used_memory(), 0) << 20
              )
              DiskBytesSpilled += os.path.getsize(path)
              os.unlink(path)  # data will be deleted after close
  
          elif not chunks:
              batch = min(int(batch * 1.5), 10000)
  
      current_chunk.sort(key=key, reverse=reverse)
      if not chunks:
          return current_chunk
  
      if current_chunk:
          chunks.append(iter(current_chunk))
  
      return heapq.merge(*chunks, key=key, reverse=reverse)
  
imports: import os
  import platform
  import shutil
  import warnings
  import gc
  import itertools
  import operator
  import random
  import sys
  import heapq
  from pyspark.serializers import BatchedSerializer, CPickleSerializer, FlattenedValuesSerializer, CompressedSerializer, AutoBatchedSerializer
  from pyspark.util import fail_on_stopiteration
label_desc: Use image node_id 4 for calling the ExternalSorter obj's underlying member method code with example usage: obj.sorted(iterator, key, reverse) and returns: heapq, current_chunk