def run_task(factory, **kwargs):
context = TaskContext(factory, **kwargs)<EOL>pstats_dir = kwargs.get("<STR_LIT>", os.getenv(PSTATS_DIR))<EOL>if pstats_dir:<EOL><INDENT>import cProfile<EOL>import tempfile<EOL>import pydoop.hdfs as hdfs<EOL>hdfs.mkdir(pstats_dir)<EOL>fd, pstats_fn = tempfile.mkstemp(suffix="<STR_LIT>")<EOL>os.close(fd)<EOL>cProfile.runctx(<EOL>"<STR_LIT>", globals(), locals(),<EOL>filename=pstats_fn<EOL>)<EOL>pstats_fmt = kwargs.get(<EOL>"<STR_LIT>",<EOL>os.getenv(PSTATS_FMT, DEFAULT_PSTATS_FMT)<EOL>)<EOL>name = pstats_fmt % (<EOL>context.task_type,<EOL>context.get_task_partition(),<EOL>os.path.basename(pstats_fn)<EOL>)<EOL>hdfs.put(pstats_fn, hdfs.path.join(pstats_dir, name))<EOL><DEDENT>else:<EOL><INDENT>_run(context, **kwargs)<EOL><DEDENT>
\
    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.