<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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)