def run_pipes(executable, input_path, output_path, more_args=None,<EOL>properties=None, force_pydoop_submitter=False,<EOL>hadoop_conf_dir=None, logger=None, keep_streams=False):
if logger is None:<EOL><INDENT>logger = utils.NullLogger()<EOL><DEDENT>if not hdfs.path.exists(executable):<EOL><INDENT>raise IOError("<STR_LIT>" % executable)<EOL><DEDENT>if not hdfs.path.exists(input_path) and not (set(input_path) & GLOB_CHARS):<EOL><INDENT>raise IOError("<STR_LIT>" % input_path)<EOL><DEDENT>if properties is None:<EOL><INDENT>properties = {}<EOL><DEDENT>properties.setdefault('<STR_LIT>', '<STR_LIT:true>')<EOL>properties.setdefault('<STR_LIT>', '<STR_LIT:true>')<EOL>if force_pydoop_submitter:<EOL><INDENT>use_pydoop_submit = True<EOL><DEDENT>else:<EOL><INDENT>use_pydoop_submit = False<EOL>ver = pydoop.hadoop_version_info()<EOL>if ver.has_security():<EOL><INDENT>if ver.is_cdh_mrv2() and hdfs.default_is_local():<EOL><INDENT>raise RuntimeError("<STR_LIT>")<EOL><DEDENT>use_pydoop_submit = hdfs.default_is_local()<EOL><DEDENT><DEDENT>args = [<EOL>"<STR_LIT>", executable,<EOL>"<STR_LIT>", input_path,<EOL>"<STR_LIT>", output_path,<EOL>]<EOL>if more_args is not None:<EOL><INDENT>args.extend(more_args)<EOL><DEDENT>if use_pydoop_submit:<EOL><INDENT>submitter = "<STR_LIT>"<EOL>pydoop_jar = pydoop.jar_path()<EOL>args.extend(("<STR_LIT>", pydoop_jar))<EOL>return run_class(submitter, args, properties,<EOL>classpath=pydoop_jar, logger=logger,<EOL>keep_streams=keep_streams)<EOL><DEDENT>else:<EOL><INDENT>return run_mapred_cmd("<STR_LIT>", args=args, properties=properties,<EOL>hadoop_conf_dir=hadoop_conf_dir, logger=logger,<EOL>keep_streams=keep_streams)<EOL><DEDENT>
Run a pipes command.

``more_args`` (after setting input/output path) and ``properties``
are passed to :func:`run_cmd`.

If not specified otherwise, this function sets the properties
``mapreduce.pipes.isjavarecordreader`` and
``mapreduce.pipes.isjavarecordwriter`` to ``"true"``.

This function works around a bug in Hadoop pipes that affects
versions of Hadoop with security when the local file system is
used as the default FS (no HDFS); see
https://issues.apache.org/jira/browse/MAPREDUCE-4000.  In those
set-ups, the function uses Pydoop's own pipes submitter
application.  You can force the use of Pydoop's submitter by
passing the argument force_pydoop_submitter=True.