docs: def run_task(factory, **kwargs):
      """\
      Run a MapReduce task.
  
      Available keyword arguments:
  
      * ``raw_keys`` (default: :obj:`False`): pass map input keys to context
        as byte strings (ignore any type information)
      * ``raw_values`` (default: :obj:`False`): pass map input values to context
        as byte strings (ignore any type information)
      * ``private_encoding`` (default: :obj:`True`): automatically serialize map
        output k/v and deserialize reduce input k/v (pickle)
      * ``auto_serialize`` (default: :obj:`True`): automatically serialize reduce
        output (map output in map-only jobs) k/v (call str/unicode then encode as
        utf-8)
  
      Advanced keyword arguments:
  
      * ``pstats_dir``: run the task with cProfile and store stats in this dir
      * ``pstats_fmt``: use this pattern for pstats filenames (experts only)
  
      The pstats dir and filename pattern can also be provided via ``pydoop
      submit`` arguments, with lower precedence in case of clashes.
      """
      context = TaskContext(factory, **kwargs)
      pstats_dir = kwargs.get("pstats_dir", os.getenv(PSTATS_DIR))
      if pstats_dir:
          import cProfile
          import tempfile
          import pydoop.hdfs as hdfs
          hdfs.mkdir(pstats_dir)
          fd, pstats_fn = tempfile.mkstemp(suffix=".pstats")
          os.close(fd)
          cProfile.runctx(
              "_run(context, **kwargs)", globals(), locals(),
              filename=pstats_fn
          )
          pstats_fmt = kwargs.get(
              "pstats_fmt",
              os.getenv(PSTATS_FMT, DEFAULT_PSTATS_FMT)
          )
          name = pstats_fmt % (
              context.task_type,
              context.get_task_partition(),
              os.path.basename(pstats_fn)
          )
          hdfs.put(pstats_fn, hdfs.path.join(pstats_dir, name))
      else:
          _run(context, **kwargs)