docs: def run(mapper=None, reducer=None, combiner=None, script_path=None, jobconfs=(), **kw):
      """Hadoopy entrance function
  
      This is to be called in all Hadoopy job's.  Handles arguments passed in,
      calls the provided functions with input, and stores the output.
  
      TypedBytes are used if the following is True
      os.environ['stream_map_input'] == 'typedbytes'
  
      It is *highly* recommended that TypedBytes be used for all non-trivial
      tasks.  Keep in mind that the semantics of what you can safely emit from
      your functions is limited when using Text (i.e., no \\t or \\n).  You can use
      the base64 module to ensure that your output is clean.
  
      If the HADOOPY_CHDIR environmental variable is set, this will immediately
      change the working directory to the one specified.  This is useful if your
      data is provided in an archive but your program assumes it is in that
      directory.
  
      As hadoop streaming relies on stdin/stdout/stderr for communication,
      anything that outputs on them in an unexpected way (especially stdout) will
      break the pipe on the Java side and can potentially cause data errors.  To
      fix this problem, hadoopy allows file descriptors (integers) to be provided
      to each task.  These will be used instead of stdin/stdout by hadoopy.  This
      is designed to combine with the 'pipe' command.
  
      To use the pipe functionality, instead of using
      `your_script.py map` use `your_script.py pipe map`
      which will call the script as a subprocess and use the read_fd/write_fd
      command line arguments for communication.  This isolates your script and
      eliminates the largest source of errors when using hadoop streaming.
  
      The pipe functionality has the following semantics
      stdin: Always an empty file
      stdout: Redirected to stderr (which is visible in the hadoop log)
      stderr: Kept as stderr
      read_fd: File descriptor that points to the true stdin
      write_fd: File descriptor that points to the true stdout
  
      | **Command Interface**
      | The command line switches added to your script (e.g., script.py) are
  
      python script.py *map* (read_fd) (write_fd)
          Use the provided mapper, optional read_fd/write_fd.
      python script.py *reduce* (read_fd) (write_fd)
          Use the provided reducer, optional read_fd/write_fd.
      python script.py *combine* (read_fd) (write_fd)
          Use the provided combiner, optional read_fd/write_fd.
      python script.py *freeze* <tar_path> <-Z add_file0 -Z add_file1...>
          Freeze the script to a tar file specified by <tar_path>.  The extension
          may be .tar or .tar.gz.  All files are placed in the root of the tar.
          Files specified with -Z will be added to the tar root.
      python script.py info
          Prints a json object containing 'tasks' which is a list of tasks which
          can include 'map', 'combine', and 'reduce'.  Also contains 'doc' which is
          the provided documentation through the doc argument to the run function.
          The tasks correspond to provided inputs to the run function.
  
      | **Specification of mapper/reducer/combiner**
      | Input Key/Value Types
      |     For TypedBytes/SequenceFileInputFormat, the Key/Value are the decoded TypedBytes
      |     For TextInputFormat, the Key is a byte offset (int) and the Value is a line without the newline (string)
      |
      | Output Key/Value Types
      |     For TypedBytes, anything Pickle-able can be used
      |     For Text, types are converted to string.  Note that neither may contain \\t or \\n as these are used in the encoding.  Output is key\\tvalue\\n
      |
      | Expected arguments
      |     mapper(key, value) or mapper.map(key, value)
      |     reducer(key, values) or reducer.reduce(key, values)
      |     combiner(key, values) or combiner.reduce(key, values)
      |
      | Optional methods
      |     func.configure(): Called before any input read.  Returns None.
      |     func.close():  Called after all input read.  Returns None or Iterator of (key, value)
      |
      | Expected return
      |     None or Iterator of (key, value)
  
      :param mapper: Function or class following the above spec
      :param reducer: Function or class following the above spec
      :param combiner: Function or class following the above spec
      :param doc: If specified, on error print this and call sys.exit(1)
      """
      if script_path is None:
          script_path = inspect.getfile(inspect.currentframe().f_back)
      script_path = os.path.abspath(script_path)
  
      parser = argparse.ArgumentParser()
      parser.add_argument('--log', help='Default log level to use', choices=('debug', 'info', 'warning', 'error', 'critical'), default='info')
      subparsers = parser.add_subparsers(help='Job Commands (additional help available inside each)')
  
      parser_freeze = subparsers.add_parser('freeze', help='Freeze the script to a tar file.')
      parser_freeze.add_argument('tar_path', help='Path to .tar or .tar.gz file.')
      parser_freeze.add_argument('-Z', help='Path to a file to be placed in the tar root (may be repeated for many files).', action='append')
      parser_freeze.set_defaults(func=run_freeze)
  
      parser_info = subparsers.add_parser('info', help='Display job info as JSON')
      parser_info.set_defaults(func=lambda : run_info(mapper, reducer, combiner, jobconfs, kw))
  
      parser_launch_frozen = subparsers.add_parser('launch_frozen', help='Run Hadoop job (freezes script)')
      parser_launch_frozen.add_argument('in_name', help='Input HDFS path')
      parser_launch_frozen.add_argument('out_name', help='Output HDFS path')
      parser_launch_frozen.add_argument('-jobconf', help='Jobconf', action='append', dest='jobconfs', default=[])
      parser_launch_frozen.set_defaults(func=lambda *args, **kw: hadoopy.launch_frozen(*args, script_path=script_path, **kw))
  
      parser_pipe = subparsers.add_parser('pipe', help='Internal: Run map/combine/reduce task using "pipe hopping" to make stdout redirect to stderr.')
      parser_pipe.add_argument('command', help='Command to run', choices=('map', 'reduce', 'combine'))
      parser_pipe.set_defaults(func=run_pipe)
  
      parser_map = subparsers.add_parser('map', help='Internal: Run map task.')
      parser_map.add_argument('read_fd', type=int, help='Read file descriptor', nargs='?')
      parser_map.add_argument('write_fd', type=int, help='Write file descriptor', nargs='?')
      parser_map.set_defaults(func=lambda **y: run_task(mapper, reducer, combiner, command='map', **y))
  
      parser_combine = subparsers.add_parser('combine', help='Internal: Run combine task.')
      parser_combine.add_argument('read_fd', type=int, help='Read file descriptor', nargs='?')
      parser_combine.add_argument('write_fd', type=int, help='Write file descriptor', nargs='?')
      parser_combine.set_defaults(func=lambda **y: run_task(mapper, reducer, combiner, command='combine', **y))
  
      parser_reduce = subparsers.add_parser('reduce', help='Internal: Run reduce task.')
      parser_reduce.add_argument('read_fd', type=int, help='Read file descriptor', nargs='?')
      parser_reduce.add_argument('write_fd', type=int, help='Write file descriptor', nargs='?')
      parser_reduce.set_defaults(func=lambda **y: run_task(mapper, reducer, combiner, command='reduce', **y))
  
      args = vars(parser.parse_args())
      # Handle logging arguments
      if 'log' in args:
          numeric_level = getattr(logging, args['log'].upper(), None)
          logging.basicConfig(level=numeric_level)
          del args['log']
      # Call function with all arguments except for itself
      func = args['func']
      del args['func']
      func(**args)